Skip to content

KafkaContainer transactions: Timeout expired while initializing transactional state in 60000ms. #1816

New issue
New issue
Open
#1984
Open
KafkaContainer transactions: Timeout expired while initializing transactional state in 60000ms.#1816
#1984
Labels
modules/kafkaresolution/acknowledgedtype/bug

Description

@ernesthill
ernesthill
opened on Aug 30, 2019

I'm trying create a test that uses kafka transactions. If I use a local instance of Kafka instead of KafkaContainer everything works fine, but when I use KafkaContainer it fails and I see the following:
Timeout expired while initializing transactional state in 60000ms.
org.apache.kafka.common.errors.TimeoutException: Timeout expired while initializing transactional state in 60000ms.
Attached are the source for my test and the logfile from the run.
kafka.log
Test.java.txt

Activity

  1. kiview commented on Sep 9, 2019

    @kiview
    kiview
    on Sep 9, 2019
    Member

    I just had a quick check with our KafkaContainerTest and added a transaction example and I see the same behaviour, but with a slightly different error mesage:

    Timeout expired after 60000milliseconds while awaiting InitProducerId
org.apache.kafka.common.errors.TimeoutException: Timeout expired after 60000milliseconds while awaiting InitProducerId

    It happens on producer.initTransactions().

  2. kiview
    added
    modules/kafka
    resolution/acknowledged
    type/bug
    on Sep 9, 2019
  3. kiview
    changed the title [-]Timeout expired while initializing transactional state in 60000ms.[/-] [+]KafkaContainer transactions: Timeout expired while initializing transactional state in 60000ms.[/+] on Sep 9, 2019

  4. bsideup commented on Sep 9, 2019

    @bsideup
    bsideup
    on Sep 9, 2019
    Member

    @gAmUssA any idea? :)

  5. gAmUssA commented on Sep 10, 2019

    @gAmUssA
    gAmUssA
    on Sep 10, 2019

    @bsideup @kiview
    Gents
    Let me take a look

  6. gAmUssA commented on Sep 10, 2019

    @gAmUssA
    gAmUssA
    on Sep 10, 2019

    @ernesthill

    you need to configure some of the broker parameters in order transaction state store will be initialized correctly.
    Here's correct test

    NOTE: it's always good to enable log output from the container to debug this kind of issues.
    I use Slf4jLogConsumer from TC.

    @Slf4j
public class ProducerTransactionTest {

  public static KafkaContainer kafka = new KafkaContainer("5.2.1")
      .withLogConsumer(new Slf4jLogConsumer(log));

  @BeforeClass
  public static void prep() {
    // see https://cwiki.apache.org/confluence/display/KAFKA/KIP-98+-+Exactly+Once+Delivery+and+Transactional+Messaging#KIP-98-ExactlyOnceDeliveryandTransactionalMessaging-NewConfigurations
    kafka.addEnv("KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR", "1");
    kafka.addEnv("KAFKA_TRANSACTION_STATE_LOG_MIN_ISR", "1");
    kafka.start();
  }

  @Test
  public void testIt() {
    Properties props = new Properties();
    props.put(BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());
    props.put(KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    props.put(VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    props.put(ENABLE_IDEMPOTENCE_CONFIG, "true");
    props.put(TRANSACTIONAL_ID_CONFIG, "prod-0");
    KafkaProducer<String, String> producer = new KafkaProducer<>(props);
    ProducerRecord<String, String> record = new ProducerRecord<>("something", "A message");
    producer.initTransactions();
    producer.beginTransaction();
    producer.send(record);
    producer.commitTransaction();
  }
}

    Let me know if you have any questions.

    @bsideup @kiview do you want me to send PR with a test for this issue?

  7. ernesthill commented on Sep 16, 2019

    @ernesthill
    ernesthill
    on Sep 16, 2019
    Author

    Thanks for the information.

  8. debop commented on Sep 23, 2019

    @debop
    debop
    on Sep 23, 2019

    Thanks for this information !!!

  9. seglo commented on Oct 11, 2019

    @seglo
    seglo
    on Oct 11, 2019
    Contributor

    This should be part of default Kafka test container config, since (AFAIK) it always runs in a single broker configuration.

  10. zenglian commented on Nov 5, 2019

    @zenglian
    zenglian
    on Nov 5, 2019

    The default value of transaction.state.log.replication.factor is 3 and transaction.state.log.min.isr is 2. So if broker count in your cluster is less than 3, kafka server fails to (automatically) create the topic __transaction_state, thus client got timedout error.

  11. zaanwer commented on Nov 11, 2019

    @zaanwer
    zaanwer
    on Nov 11, 2019

    I did have this transaction.state.log.replication.factor set to 1 but still seeing this error in producer.initTransaction()

    "Timeout expired while initializing transactional state in 60000ms"

  12. zaanwer commented on Nov 11, 2019

    @zaanwer
    zaanwer
    on Nov 11, 2019

    It went to following code, got the transactionManager.initializeTransactions into result but timing out in last line result.await(maxBlockTimeMs, TimeUnit.MILLISECONDS);
    public void initTransactions() {
    throwIfNoTransactionManager();
    throwIfProducerClosed();
    TransactionalRequestResult result = transactionManager.initializeTransactions();
    sender.wakeup();
    result.await(maxBlockTimeMs, TimeUnit.MILLISECONDS);
    }

  13. zenglian commented on Nov 13, 2019

    @zenglian
    zenglian
    on Nov 13, 2019

    I did have this transaction.state.log.replication.factor set to 1 but still seeing this error in producer.initTransaction()

    "Timeout expired while initializing transactional state in 60000ms"

    Note there are 2 props to change. Do not try to debug the client code. It’s an server error.

  14. 6 remaining items

  15. gAmUssA commented on Jun 8, 2020

    @gAmUssA
    gAmUssA
    on Jun 8, 2020

    @pancudaniel7 could you share a reproducer?
    Thank you

  16. cemkayan commented on Aug 3, 2020

    @cemkayan
    cemkayan
    on Aug 3, 2020

    I am also getting same exception. I am using 'KafkaTemplate' with 'executeInTransaction'

    props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 10000);
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "prod-0"); // I am really not clear what this is for. I set it once in the producerConfig. Then create KafkaTemplate and use it many times...

    This is how I call Kafka template:

    	    kafkaTemplate.setTransactionIdPrefix(message.getGuid()); // I am also not sure about this line. I tried to put a unique value to prevent same message to be created twice.

	    kafkaTemplate.executeInTransaction(kafkaTemplate -> {

		ListenableFuture<SendResult<String, Event>> future = kafkaTemplate.send(topic, message);

		future.addCallback(new ListenableFutureCallback<SendResult<String, Event>>() {

		    @Override
		    public void onSuccess(SendResult<String, Event> result) {
		    }

		    @Override
		    public void onFailure(Throwable ex) {
		    }
		});

		return null;
	    });

    This is what I receive

    org.apache.kafka.common.errors.TimeoutException: Timeout expired after 10000milliseconds while awaiting InitProducerId

  17. maver1ck commented on Sep 29, 2020

    @maver1ck
    maver1ck
    on Sep 29, 2020

    Same problem here :(

    I'm using Flink job with FlinkKafkaProducer.Semantic.EXACTLY_ONCE.

    When using same job without Test Containers everything is working fine.

  18. bsideup commented on Sep 29, 2020

    @bsideup
    bsideup
    on Sep 29, 2020
    Member

    @maver1ck the transactions are not enabled by default, see @gAmUssA's answer:
    #1816 (comment)

  19. maver1ck commented on Sep 29, 2020

    @maver1ck
    maver1ck
    on Sep 29, 2020

    @bsideup checking this

    EDIT: It works. Thanks.

  20. MrMikeFloyd commented on Nov 12, 2020

    @MrMikeFloyd
    MrMikeFloyd
    on Nov 12, 2020

    @ernesthill

    you need to configure some of the broker parameters in order transaction state store will be initialized correctly.
    Here's correct test

    NOTE: it's always good to enable log output from the container to debug this kind of issues.
    I use Slf4jLogConsumer from TC.

    @Slf4j
public class ProducerTransactionTest {

  public static KafkaContainer kafka = new KafkaContainer("5.2.1")
      .withLogConsumer(new Slf4jLogConsumer(log));

  @BeforeClass
  public static void prep() {
    // see https://cwiki.apache.org/confluence/display/KAFKA/KIP-98+-+Exactly+Once+Delivery+and+Transactional+Messaging#KIP-98-ExactlyOnceDeliveryandTransactionalMessaging-NewConfigurations
    kafka.addEnv("KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR", "1");
    kafka.addEnv("KAFKA_TRANSACTION_STATE_LOG_MIN_ISR", "1");
    kafka.start();
  }

  @Test
  public void testIt() {
    Properties props = new Properties();
    props.put(BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());
    props.put(KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    props.put(VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    props.put(ENABLE_IDEMPOTENCE_CONFIG, "true");
    props.put(TRANSACTIONAL_ID_CONFIG, "prod-0");
    KafkaProducer<String, String> producer = new KafkaProducer<>(props);
    ProducerRecord<String, String> record = new ProducerRecord<>("something", "A message");
    producer.initTransactions();
    producer.beginTransaction();
    producer.send(record);
    producer.commitTransaction();
  }
}

    Let me know if you have any questions.

    @bsideup @kiview do you want me to send PR with a test for this issue?

    This did the trick for me 👆❤️

    In case anyone has the same issue using EmbeddedKafka in Spring Boot test, the values can be set when defining the config for the embedded Kafka instance as follows:

    @EmbeddedKafka(topics = "kafka-test", ports = 9099, brokerProperties = {
    "transaction.state.log.replication.factor=1",
    "transaction.state.log.min.isr=1"
})
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
    properties = {
        "spring.kafka.bootstrap-servers=localhost:9099"
    })
class KafkaEmbeddedIT {
...
}

  21. kcotzen commented on Nov 26, 2020

    @kcotzen
    kcotzen
    on Nov 26, 2020

    Hi everyone, Unfortunately I have the same error:

    2020-11-26 11:27:08.246  INFO [poc-test,,,] 506391 --- [ producer-tx-3] org.apache.kafka.clients.Metadata        : [Producer clientId=producer-tx-3, transactionalId=tx-3] Cluster ID: -nEH5zcySOSTk7pnaSsZOg
2020-11-26 11:28:08.238 ERROR [poc-test,,,] 506391 --- [           main] o.s.c.s.b.k.p.KafkaTopicProvisioner      : Failed to obtain partition information

org.apache.kafka.common.errors.TimeoutException: Timeout expired after 60000milliseconds while awaiting InitProducerId

    In the console output I see this:

    2020-11-26 11:27:08.218  INFO [poc-test,,,] 506391 --- [           main] o.a.k.clients.producer.ProducerConfig    : ProducerConfig values: 
	acks = -1
	batch.size = 16384
	bootstrap.servers = [PLAINTEXT://localhost:33193]
	buffer.memory = 33554432
	client.dns.lookup = default
	client.id = producer-tx-3
	compression.type = none
	connections.max.idle.ms = 540000
	delivery.timeout.ms = 120000
	enable.idempotence = true
	interceptor.classes = []
	key.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer
	linger.ms = 0
	max.block.ms = 60000
	max.in.flight.requests.per.connection = 5
	max.request.size = 1048576
	metadata.max.age.ms = 300000
	metadata.max.idle.ms = 300000
	metric.reporters = []
	metrics.num.samples = 2
	metrics.recording.level = INFO
	metrics.sample.window.ms = 30000
	partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
	receive.buffer.bytes = 32768
	reconnect.backoff.max.ms = 1000
	reconnect.backoff.ms = 50
	request.timeout.ms = 30000
	retries = 1
	retry.backoff.ms = 100
	sasl.client.callback.handler.class = null
	sasl.jaas.config = null
	sasl.kerberos.kinit.cmd = /usr/bin/kinit
	sasl.kerberos.min.time.before.relogin = 60000
	sasl.kerberos.service.name = null
	sasl.kerberos.ticket.renew.jitter = 0.05
	sasl.kerberos.ticket.renew.window.factor = 0.8
	sasl.login.callback.handler.class = null
	sasl.login.class = null
	sasl.login.refresh.buffer.seconds = 300
	sasl.login.refresh.min.period.seconds = 60
	sasl.login.refresh.window.factor = 0.8
	sasl.login.refresh.window.jitter = 0.05
	sasl.mechanism = GSSAPI
	security.protocol = PLAINTEXT
	security.providers = null
	send.buffer.bytes = 131072
	ssl.cipher.suites = null
	ssl.enabled.protocols = [TLSv1.2]
	ssl.endpoint.identification.algorithm = https
	ssl.key.password = null
	ssl.keymanager.algorithm = SunX509
	ssl.keystore.location = null
	ssl.keystore.password = null
	ssl.keystore.type = JKS
	ssl.protocol = TLSv1.2
	ssl.provider = null
	ssl.secure.random.implementation = null
	ssl.trustmanager.algorithm = PKIX
	ssl.truststore.location = null
	ssl.truststore.password = null
	ssl.truststore.type = JKS
	transaction.timeout.ms = 60000
	transactional.id = tx-3
	value.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer

    My configuration of KafkaContainer :

            lateinit var kafka: KafkaContainer

        init {
            configureKafka()
        }

        private fun configureKafka() {
            kafka = KafkaContainer("5.3.2-1")
            kafka.addEnv("KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR", "2")
            kafka.addEnv("KAFKA_TRANSACTION_STATE_LOG_MIN_ISR", "1")
            kafka.start()
        }

    And my configuration in the application.yml looks like this:

    spring:
  cloud:
    stream:
      kafka:
        binder:
          transaction:
            transaction-id-prefix: tx-
          producer-properties:
            retries: 1
            acks: all

    I'm not getting it to start up successfully, has anyone been able to overcome this?
    thanks in advance

  22. maver1ck commented on Nov 26, 2020

    @maver1ck
    maver1ck
    on Nov 26, 2020

    kafka.addEnv("KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR", "2")
    Why 2?

  23. kcotzen commented on Nov 26, 2020

    @kcotzen
    kcotzen
    on Nov 26, 2020

    kafka.addEnv("KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR", "2")
    Why 2?

    If I use :

                kafka.addEnv("KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR", "3")
            kafka.addEnv("KAFKA_TRANSACTION_STATE_LOG_MIN_ISR", "2")

    The error is the same. If I use 1 for both, I got this:

    2020-11-26 13:15:01.309  WARN [poc-test,,,] 523026 --- [| producer-tx-0] org.apache.kafka.clients.NetworkClient   : [Producer clientId=producer-tx-0, transactionalId=tx-0] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.
2020-11-26 13:15:01.309  WARN [poc-test,,,] 523026 --- [| producer-tx-0] org.apache.kafka.clients.NetworkClient   : [Producer clientId=producer-tx-0, transactionalId=tx-0] Bootstrap broker localhost:9092 (id: -1 rack: null) disconnected

  24. bsideup commented on Nov 26, 2020

    @bsideup
    bsideup
    on Nov 26, 2020
    Member

    @kcotzen it should be 1, unless you start & connect two brokers

  25. kcotzen commented on Nov 26, 2020

    @kcotzen
    kcotzen
    on Nov 26, 2020

    I used value 1 for both, and I got this error:

    2020-11-26 13:15:01.309  WARN [poc-test,,,] 523026 --- [| producer-tx-0] org.apache.kafka.clients.NetworkClient   : [Producer clientId=producer-tx-0, transactionalId=tx-0] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.
2020-11-26 13:15:01.309  WARN [poc-test,,,] 523026 --- [| producer-tx-0] org.apache.kafka.clients.NetworkClient   : [Producer clientId=producer-tx-0, transactionalId=tx-0] Bootstrap broker localhost:9092 (id: -1 rack: null) disconnected

  26. kcotzen commented on Nov 26, 2020

    @kcotzen
    kcotzen
    on Nov 26, 2020

    I made some modifications(I removed use of chainedTransactionManager), and it works Zero errors in console output, but when I use chainedTransactionManager :

    2020-11-26 13:15:01.309  WARN [poc-test,,,] 523026 --- [| producer-tx-0] org.apache.kafka.clients.NetworkClient   : [Producer clientId=producer-tx-0, transactionalId=tx-0] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.
2020-11-26 13:15:01.309  WARN [poc-test,,,] 523026 --- [| producer-tx-0] org.apache.kafka.clients.NetworkClient   : [Producer clientId=producer-tx-0, transactionalId=tx-0] Bootstrap broker localhost:9092 (id: -1 rack: null) disconnected

    sounds weird. Any ideas please?

  27. kcotzen commented on Nov 26, 2020

    @kcotzen
    kcotzen
    on Nov 26, 2020

    well, for some reason the problem was caused by the chainedTransactionManager definition, I think the console output is very weird and led me to confusion.
    Thanks anyway.

  28. daoanhvu commented on Jul 31, 2024

    @daoanhvu
    daoanhvu
    on Jul 31, 2024

    I got the same issue while trying to apply transaction producers.
    I fixed it by overriding these two settings at server side:
    transaction.state.log.min.isr = 1
    transaction.state.log.replication.factor = 1

    if you are using Kafka container with docker then most likely you will go with:
    KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
    KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1

  29. aymenjamaaoui commented on Oct 4, 2026

    @aymenjamaaoui
    aymenjamaaoui
    on Oct 4, 2026

    This is fixed since 1.15.2: the Kafka containers set KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR and KAFKA_TRANSACTION_STATE_LOG_MIN_ISR to 1. #12102 adds tests for transactions so this does not break again. I think this issue can be closed.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    modules/kafkaresolution/acknowledgedtype/bug

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions