Kafka transactional producer throws 'Timeout expired while initializing transactional state in 60000ms'

I have configured a Kafka ProducerFactory with a transactionIdPrefix, in order to enable transaction synchronization using @Transactional (see Spring documentation on producer-only transactions).

I'm running an EmbeddedKafka in my integration test, to see how it behaves.

The logs show the following :

DEBUG 8384 --- [ad | producer-1] o.a.k.clients.producer.internals.Sender  :
    [Producer clientId=producer-1, transactionalId=tx-0-0]
    Sending transactional request (type=FindCoordinatorRequest, coordinatorKey=tx-0-0, coordinatorType=TRANSACTION) to node 127.0.0.1:61445 (id: -1 rack: null)

DEBUG 8384 --- [ad | producer-1] o.a.k.c.p.internals.TransactionManager   :
    [Producer clientId=producer-1, transactionalId=tx-0-0]
    Enqueuing transactional request (type=FindCoordinatorRequest, coordinatorKey=tx-0-0, coordinatorType=TRANSACTION)

Timeout expired while initializing transactional state in 60000ms.

This is thrown when DefaultKafkaProducerFactory executes newProducer.initTransactions().

My configuration is the following :

IntegrationTest

@EmbeddedKafka(brokerProperties = { "transaction.state.log.replication.min.isr=1", "transaction.state.log.replication.factor=1" })

ProducerConfig

@Bean
  public ProducerFactory<String, String> transactionalProducerFactory() {
    Map<String, Object> configuration = new HashMap<>();

    configuration.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServers());
    configuration.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    configuration.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    String transactionIdPrefix = "tx-0-";
    configuration.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
    DefaultKafkaProducerFactory<String, String> factory = new DefaultKafkaProducerFactory<>(configuration);
    factory.setTransactionIdPrefix(transactionIdPrefix);

    return factory;
  }

@Bean
  public KafkaTemplate<String, String> transactionalKafka() {
    return new KafkaTemplate<>(transactionalProducerFactory());
  }

Spring-Kafka version : 2.2.7.RELEASE

I don't see how to move forward, I think that I followed every step from the documentation and the communication between the Kafka client and the broker should be fine during transaction initialization. Could anyone please help me fix this?

I could solve the problem thanks to the embedded kafka server logs.

Property transaction.state.log.min.isr defaulted to 2, I had to overwrite it with transaction.state.log.min.isr = 1 to fix the server error. After that my integration test passed.