Kafka InitProducerId Timeout: Why It Hangs and the Fix

Your producer starts, blocks for exactly 60 seconds, and dies with this:

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

Kafka Streams and the transactional API dress it up slightly differently, but it's the same wall:

org.apache.kafka.streams.errors.StreamsException: stream-thread [app-StreamThread-1] Failed to initialize task 0_0 due to timeout.
Caused by: org.apache.kafka.common.errors.TimeoutException: Timeout expired after 60000ms while awaiting InitProducerId

org.apache.kafka.common.KafkaException: Timeout expired while initializing transactional state in 60000ms.

No stack trace pointing at your code. No broker exception in the client log. Just a 60-second stall on the very first send() or on initTransactions(), then a timeout. This one confuses people because since kafka-clients 3.0 the InitProducerId round trip happens even when you never touched the transactional API — idempotence is on by default. The fix depends entirely on which InitProducerId path you're on, and almost every "just set replication factor to 1" answer online conflates the two. Let's separate them.

1. The Error

The exact symptoms, by client:

• Plain producer (no transactional.id): first send() returns a future that never completes; after max.block.ms (default 60000) the callback fires with TimeoutException: Timeout expired after 60000ms while awaiting InitProducerId.
• Transactional producer (transactional.id set): initTransactions() blocks and throws the same TimeoutException, or KafkaException: Timeout expired while initializing transactional state in 60000ms.
• Kafka Streams with processing.guarantee=exactly_once_v2: StreamThread fails task initialization with the wrapped timeout and shuts the thread down.

Environment where this shows up:

• kafka-clients / Kafka 3.0 through 4.0 (idempotence default-on since 3.0, effective since 3.0.1/3.1.1/3.2.0 due to KAFKA-13598).
• Single-broker Docker Compose, Testcontainers KafkaContainer, or a freshly bootstrapped KRaft cluster.
• Managed Kafka or K8s where one broker in the path is unreachable.

Two mechanically different failures produce the identical message. Confusing the two is why people burn an afternoon on it.

2. How to Reproduce It (step-by-step)

Reproduction A — transactional producer on a single broker

This is the one that generates the most search traffic. Spin up a single KRaft broker but leave the internal topic replication at its production default:

# docker-compose.yml — single broker, DEFAULT internal-topic RF (the trap)
services:
  kafka:
    image: apache/kafka:4.0.0
    container_name: kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      # NOTE: transaction.state.log.replication.factor defaults to 3 — left unset on purpose

docker compose up -d

Now a transactional producer:

// kafka-clients 4.0.0
Properties p = new Properties();
p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
p.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "poc-tx-1"); // this is the trigger

try (Producer<String, String> producer = new KafkaProducer<>(p)) {
    producer.initTransactions();   // hangs 60s, then throws
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("orders", "k", "v"));
    producer.commitTransaction();
}

initTransactions() blocks for 60 seconds and throws. Meanwhile the broker log shows the real cause, which never reaches the client:

[Controller] ... Unable to create topic __transaction_state with replication factor 3:
only 1 broker(s) are registered. INVALID_REPLICATION_FACTOR

Reproduction B — plain idempotent producer that can't reach a broker

Same client defaults (idempotence on), but here the broker is simply unreachable — e.g. advertised.listeners points at a hostname the client can't resolve. Change the advertised listener to a bad name:

      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-internal:9092  # not resolvable from host

Properties p = new Properties();
p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
// no transactional.id — plain producer, but enable.idempotence=true by default
try (Producer<String, String> producer = new KafkaProducer<>(p)) {
    producer.send(new ProducerRecord<>("orders", "k", "v")).get(); // 60s, then InitProducerId timeout
}

Bootstrap succeeds (localhost:9092 is reachable), metadata comes back advertising kafka-internal:9092, the client can't connect there, and the InitProducerId request never lands. Same 60-second timeout, completely different root cause. There is no __transaction_state problem here at all.

3. Why It Happens — Surface Level

Since kafka-clients 3.0, enable.idempotence=true and acks=all are the defaults (KIP-679). An idempotent or transactional producer must obtain a producer ID (PID) before it can send its first batch, because every record batch is stamped with (producerId, producerEpoch, sequence) for dedup and ordering on the broker. That handshake is the InitProducerIdRequest. If it can't complete within max.block.ms, you get the timeout.

The catch: transactional and non-transactional producers acquire the PID by two different routes, and only one of them touches the transaction coordinator and the __transaction_state topic.

4. Why It Happens — Under the Hood

The non-transactional idempotent path

A producer with no transactional.id has no coordinator. On first send, the client's TransactionManager targets the least-loaded node with an InitProducerIdRequest carrying a null transactional id. On the broker, KafkaApis.handleInitProducerIdRequest sees the null id and allocates a PID directly from the ProducerIdManager — a controller-backed block allocator (a KRaft controller RPC in 3.3+; ZooKeeper in the legacy path). No transaction coordinator. No __transaction_state topic. No replication-factor requirement.

So if a plain idempotent producer times out on InitProducerId, the PID allocator was unreachable — meaning the client couldn't reach any usable broker. This is a connectivity problem: wrong advertised.listeners, a dead broker, a firewall, or a metadata response pointing at an endpoint the client can't route to. Setting transaction.state.log.replication.factor=1 will do absolutely nothing here, because this path never consults that topic.

The transactional path

A producer with transactional.id set must first locate its transaction coordinator. initTransactions() sends a FindCoordinatorRequest(type=TRANSACTION, key=transactional.id). The coordinator is the leader of the __transaction_state partition that the id hashes to. That means __transaction_state has to exist and have a leader before the coordinator can be found.

__transaction_state is auto-created on first use with transaction.state.log.replication.factor (default 3) and transaction.state.log.min.isr (default 2). On a single-broker cluster, creation fails with INVALID_REPLICATION_FACTOR — only in the broker log. The client just sees FindCoordinator return COORDINATOR_NOT_AVAILABLE, retries it every retry.backoff.ms until max.block.ms elapses, and reports Timeout expired ... while awaiting InitProducerId. The RF failure never surfaces to the producer.

The same transactional path fails for three other reasons worth knowing:

• Coordinator leader election in flight — during a rolling restart the __transaction_state partition's leader is briefly none; FindCoordinator returns COORDINATOR_NOT_AVAILABLE and self-heals in seconds on RF≥3.
• Coordinator broker unreachable — __transaction_state exists and has a leader, but that specific broker's advertised.listeners endpoint isn't routable from the client. FindCoordinator succeeds, InitProducerId to the coordinator hangs.
• Broker genuinely down — the coordinator partition's leader is offline and, with RF=1 or unmet min.isr, no replica can take over.

This is why "set RF to 1" fixes the single-broker POC but not a real cluster: those three are availability problems, not arithmetic problems. (The broker-side replication-factor arithmetic itself — why RF=3 can't be satisfied and how the offsets/transaction internal topics get created — is covered in the InvalidReplicationFactorException article; this one is about the client-side InitProducerId symptom.)

5. The Fix

Fix A — single-broker dev/POC (transactional or Streams EOS)

Override the internal-topic replication factors so __transaction_state (and __consumer_offsets) can be created on one broker:

     environment:
       KAFKA_NODE_ID: 1
       KAFKA_PROCESS_ROLES: broker,controller
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1

Restart with a clean volume so the topic gets recreated at the new RF:

docker compose down -v && docker compose up -d

Verify the topic is healthy before you run your app:

docker exec kafka /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 \
  --describe --topic __transaction_state
# ReplicationFactor: 1, Leader: 1 on every partition — good

For Testcontainers, the stock KafkaContainer already sets these to 1, but a custom apache/kafka container does not. Add them explicitly:

KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("apache/kafka:4.0.0"))
    .withEnv("KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR", "1")
    .withEnv("KAFKA_TRANSACTION_STATE_LOG_MIN_ISR", "1")
    .withEnv("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR", "1");

Fix B — plain idempotent producer timeout (connectivity)

There is no RF knob to turn. Fix the routing so the client can reach a broker:

# Confirm what the broker advertises and whether you can reach it
docker exec kafka /opt/kafka/bin/kafka-broker-api-versions.sh \
  --bootstrap-server localhost:9092
# If this hangs from the host, advertised.listeners is wrong

Make the advertised endpoint resolvable from wherever the client runs. From the host, that's localhost; from another container, it's the service name — use dual listeners so both work:

-      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-internal:9092
+      KAFKA_LISTENERS: INTERNAL://0.0.0.0:19092,EXTERNAL://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
+      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:19092,EXTERNAL://localhost:9092
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT
+      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL

Only as a throwaway escape hatch on a POC where you truly don't need exactly-once, you can turn idempotence off to skip InitProducerId entirely — never do this in production:

p.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, false);
p.put(ProducerConfig.ACKS_CONFIG, "1");

Don't just raise the timeout

Bumping max.block.ms higher only makes you wait longer for the same failure. It's occasionally justified when the coordinator is briefly electing during a rolling restart, but if the root cause is a missing topic or bad listener, a longer timeout changes nothing.

6. Best Practices & The Better Design

Provision the internal topics deliberately instead of relying on lazy auto-create at production RF. On a real cluster, create __transaction_state and __consumer_offsets with RF=3 and min.insync.replicas=2 at bootstrap so the coordinator is available before the first transactional client connects. In dev, pin them to 1 in your compose file as shown. Either way, the replication factor is a decision, not an accident.

For the transactional producer itself, keep a stable, instance-unique transactional.id, initialize once at startup, and separate fatal from retriable handling:

Producer<String, String> producer = new KafkaProducer<>(props);
try {
    producer.initTransactions();               // once, at startup
} catch (TimeoutException e) {
    // coordinator not reachable yet — retry init with backoff, don't spin the send loop
    throw new IllegalStateException("Transaction coordinator unavailable at startup", e);
}

In Spring Kafka, set transaction-id-prefix and let the framework manage the KafkaTransactionManager; the prefix plus the listener-container instance yields a stable id per instance. Provision internal-topic RF at the cluster level, not per app.

The deeper design point: if you enabled a transactional.id only to get idempotence, you didn't need transactions. Idempotence (default-on) gives you no-duplicate, in-order delivery to a single producer session without ever touching the transaction coordinator. Reserve transactional.id for genuine read-process-write atomicity (consume-transform-produce or Streams EOS). Fewer producers on the coordinator path means fewer places this timeout can bite.

7. How to Prevent It Long-Term

• CI smoke test the transactional path with Testcontainers — call initTransactions() and commit one transaction. It catches a missing internal-topic RF override before it reaches a developer's POC.
• Monitor coordinator availability: alert on kafka.server:type=transaction-coordinator-metrics unavailability and on UnderMinIsrPartitionCount / OfflinePartitionsCount for __transaction_state. A transactional producer can only be as available as that topic.
• Lint your compose and Helm values so transaction.state.log.replication.factor is explicit — either 1 (dev) or 3 (prod), never left to default-on-single-broker.
• Standardize listener config as code so advertised.listeners is correct for both in-cluster and external clients; most "plain idempotent InitProducerId timeout" incidents are really listener incidents.
• Alert on producer record-error-rate and construction-time failures so a 60-second startup stall shows up as a failed deploy, not a silent hang.

8. Key Takeaways

• Timeout expired after 60000ms while awaiting InitProducerId means the producer couldn't get its producer ID within max.block.ms — since kafka-clients 3.0 this handshake happens even for non-transactional producers because idempotence is default-on.
• Two different failures, one message. Transactional producers route through the transaction coordinator and __transaction_state; plain idempotent producers get their PID directly from the broker with no coordinator involved.
• On a single broker, the transactional path fails because __transaction_state can't be created at its default RF of 3 — fix with transaction.state.log.replication.factor=1 and min.isr=1 in dev.
• A plain idempotent producer timing out is a connectivity problem (bad advertised.listeners, unreachable broker) — the RF knob does nothing; fix the routing.
• In production, provision internal topics at RF=3/min.isr=2 up front, keep transactional.id only for real exactly-once, and raising max.block.ms is not a fix.