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

[ernesthill]

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

[kiview]

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().

[bsideup]

@gAmUssA any idea? :)

[gAmUssA]

@bsideup @kiview
Gents
Let me take a look

[gAmUssA]

@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?

[ernesthill]

Thanks for the information.

[debop]

Thanks for this information !!!

[seglo]

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

[zenglian]

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.

[zaanwer]

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"

[zaanwer]

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);
}