Kafka InvalidTxnStateException: Causes and How to Fix It

InvalidTxnStateException is the transactional producer telling you it received a request it cannot serve in its current state. It is not a networking error, not fencing, and not PID expiry. It's a state-machine violation: you (or a framework on your behalf) called beginTransaction(), send(), commitTransaction(), or sendOffsetsToTransaction() at a point where the transaction coordinator — or the client's own TransactionManager — was not in a state that permits it.

This one bites hardest in three places: hand-rolled transactional producers in a POC, Spring Kafka apps that mix transactional and non-transactional templates, and EOS sinks (Flink, Kafka Connect, SeaTunnel) after a prior error left the producer in an abortable state. Let's nail down the exact triggers and the fix.

1. The Error

The client-side stack trace, from kafka-clients:

org.apache.kafka.common.errors.InvalidTxnStateException: The producer attempted
a transactional operation in an invalid state.

Wrapped through KafkaProducer.commitTransaction() it usually looks like:

org.apache.kafka.common.KafkaException: Cannot execute transactional method
because we are in an error state
	at org.apache.kafka.clients.producer.internals.TransactionManager.maybeFailWithError(TransactionManager.java:1010)
	at org.apache.kafka.clients.producer.internals.TransactionManager.beginCommit(TransactionManager.java:311)
	at org.apache.kafka.clients.producer.KafkaProducer.commitTransaction(KafkaProducer.java:788)
Caused by: org.apache.kafka.common.errors.InvalidTxnStateException: The producer
attempted a transactional operation in an invalid state.

In Spring Kafka the same underlying exception surfaces from KafkaTemplate.executeInTransaction(...) or from the listener container's transaction manager:

org.apache.kafka.common.errors.InvalidTxnStateException: The producer attempted
a transactional operation in an invalid state.

The broker returns error code 48, INVALID_TXN_STATE. It is a fatal, non-retriable error for the current transaction — it extends ApiException, not RetriableException. Retrying the same call in a loop will not help; you must abort and (in most cases) rebuild the producer.

Where it shows up:

• kafka-clients 2.x–4.x, any broker version that supports transactions (0.11+).
• Local Docker single-broker POCs using transactional.id.
• Spring Kafka (spring-kafka 2.x/3.x) with transaction-id-prefix set, or executeInTransaction.
• Flink KafkaSink / Kafka Connect / SeaTunnel with exactly-once semantics.
• Kafka Streams with processing.guarantee=exactly_once_v2 after a task error.

2. How to Reproduce It

Spin up a single-broker KRaft cluster with the internal topics forced to RF=1 (transactions need __transaction_state):

# docker-compose.yml
services:
  kafka:
    image: apache/kafka:3.9.0
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1

docker compose up -d
docker exec -it $(docker ps -qf name=kafka) \
  /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 \
  --create --topic orders --partitions 1 --replication-factor 1

Trigger A — send without beginning a transaction

The single most common POC mistake: transactional.id is set, initTransactions() runs, but you forget beginTransaction().

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, "orders-tx-1");

try (KafkaProducer<String, String> producer = new KafkaProducer<>(p)) {
    producer.initTransactions();
    // BUG: no beginTransaction()
    producer.send(new ProducerRecord<>("orders", "k", "v"));
    producer.commitTransaction();   // throws
}

The commitTransaction() (and often the send() before it) fails because the client TransactionManager is in READY, not IN_TRANSACTION.

Trigger B — commit after an abortable error

Any send inside a transaction can fail with an abortable error (NotEnoughReplicasException, a batch timeout, etc.). The transaction is now poisoned. If you call commitTransaction() instead of abortTransaction(), you get InvalidTxnStateException:

producer.beginTransaction();
try {
    producer.send(record).get();     // fails -> abortable error latched
    producer.commitTransaction();    // BUG: commit on a poisoned txn -> throws
} catch (KafkaException e) {
    // correct path is abortTransaction() here
}

Trigger C — a stale producer replays a request the coordinator already fenced/expired

In Flink/Connect EOS, a producer that was cleanly closed or whose transaction the coordinator already timed out (transaction.timeout.ms) tries to commit/continue. The coordinator's TransactionMetadata is in CompleteAbort/Empty, so the incoming EndTxn or AddPartitionsToTxn is rejected with INVALID_TXN_STATE.

3. Why It Happens — Surface Level

Kafka transactions are a strict state machine, enforced on both sides. The client TransactionManager tracks its own state (READY, IN_TRANSACTION, COMMITTING_TRANSACTION, ABORTING_TRANSACTION, ABORTABLE_ERROR, FATAL_ERROR), and the transaction coordinator tracks the durable state in __transaction_state (Empty, Ongoing, PrepareCommit, PrepareAbort, CompleteCommit, CompleteAbort). Every transactional API call is only legal from specific states. Calling one from the wrong state throws InvalidTxnStateException — locally if the client catches it, or remotely as broker error code 48.

The mechanical causes reduce to: (a) you skipped beginTransaction(), (b) you tried to commit a transaction that already hit an abortable error, or (c) a request arrived at the coordinator for a transaction that no longer exists in the state it assumes.

4. Why It Happens — Under the Hood

The transactional protocol has a fixed happy-path sequence, and every message is tagged with a (producerId, producerEpoch):

1. initTransactions() → InitProducerId to the coordinator → client moves UNINITIALIZED → READY, coordinator writes Empty.
2. beginTransaction() → client-only transition READY → IN_TRANSACTION. No RPC yet.
3. First send() to a new partition → AddPartitionsToTxn → coordinator moves Empty → Ongoing.
4. commitTransaction() → EndTxn(commit=true) → coordinator Ongoing → PrepareCommit → writes markers → CompleteCommit.
5. abortTransaction() → EndTxn(commit=false) → Ongoing → PrepareAbort → CompleteAbort.

beginTransaction() is purely a client-side flag flip — this is why forgetting it (Trigger A) doesn't fail there but at the next transactional operation, when the TransactionManager checks currentState == IN_TRANSACTION and finds READY.

For Trigger B: when a send fails with an abortable error, TransactionManager transitions to ABORTABLE_ERROR and latches the exception. From ABORTABLE_ERROR the only legal exit is abortTransaction(). commitTransaction() calls maybeFailWithError(), sees the latched error, and refuses — surfacing as InvalidTxnStateException / "Cannot execute transactional method because we are in an error state."

For Trigger C: the coordinator is the source of truth. If transaction.timeout.ms (default 60s) elapses with an Ongoing transaction, the coordinator proactively aborts it and bumps the producer epoch. A late EndTxn or AddPartitionsToTxn from the old producer now targets a transaction in CompleteAbort/Empty with a stale epoch — rejected INVALID_TXN_STATE. This is the Flink/Connect "commit after checkpoint stall" failure: the checkpoint took longer than the transaction timeout, and the coordinator already gave up.

5. The Fix

Trigger A — always pair begin with commit/abort

  producer.initTransactions();
- producer.send(new ProducerRecord<>("orders", "k", "v"));
- producer.commitTransaction();
+ producer.beginTransaction();
+ producer.send(new ProducerRecord<>("orders", "k", "v"));
+ producer.commitTransaction();

Trigger B — abort on any KafkaException, and treat state errors as fatal

The canonical transactional loop distinguishes fatal errors (rebuild the producer) from abortable ones (abort and continue):

producer.beginTransaction();
try {
    for (ProducerRecord<String, String> r : batch) {
        producer.send(r);
    }
    producer.sendOffsetsToTransaction(offsets, groupMetadata);
    producer.commitTransaction();
} catch (ProducerFencedException | OutOfOrderSequenceException
         | AuthorizationException e) {
    // fatal: producer is unusable, close and rebuild
    producer.close();
    throw e;
} catch (KafkaException e) {
    // abortable: roll back this transaction, keep the producer
    producer.abortTransaction();
}

Never call commitTransaction() in a catch block, and never call it after a send has already thrown. If a send().get() threw, the transaction is already poisoned — go straight to abortTransaction().

Trigger C — size the transaction timeout above your commit cadence

For Flink/Connect/streaming sinks, the transaction timeout must exceed the longest possible time between beginTransaction() and commitTransaction() (i.e., your checkpoint interval plus its worst-case duration):

# producer config on the EOS sink
transaction.timeout.ms=900000        # 15 min, must be >= checkpoint interval + slack

And on the broker, the client value is clamped by:

transaction.max.timeout.ms=900000    # broker cap; raise if clients need more

If a client requests a transaction.timeout.ms larger than the broker's transaction.max.timeout.ms, initTransactions() fails outright — so raise the broker cap first.

6. Best Practices & The Better Design

Let the framework own the transaction boundaries. Hand-rolled beginTransaction/commitTransaction is where Triggers A and B come from. In Spring Kafka, configure transaction-id-prefix and let the container manage begin/commit/abort:

@Bean
public KafkaTemplate<String, String> kafkaTemplate(ProducerFactory<String, String> pf) {
    return new KafkaTemplate<>(pf);
}

@Bean
public ProducerFactory<String, String> producerFactory() {
    Map<String, Object> cfg = new HashMap<>();
    cfg.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    cfg.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    cfg.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    DefaultKafkaProducerFactory<String, String> pf =
        new DefaultKafkaProducerFactory<>(cfg);
    pf.setTransactionIdPrefix("orders-tx-");   // enables transactions
    return pf;
}

// begin/commit/abort handled for you; an exception rolls the transaction back
kafkaTemplate.executeInTransaction(t -> {
    t.send("orders", key, value);
    return null;
});

A subtle Spring trap: mixing a transactional and a non-transactional send on the same template inside a transaction, or calling a transactional send from a thread that isn't inside the container's transaction. Keep one template's transactional intent consistent. If you need read-process-write exactly-once, use the listener container's KafkaAwareTransactionManager so consumer offsets commit inside the producer transaction via sendOffsetsToTransaction, rather than committing offsets separately.

Reserve transactions for when you actually need atomicity. If all you want is dedup on retries, plain idempotence (enable.idempotence=true, default since kafka-clients 3.0) needs no transactional.id, no coordinator, and has none of this state machine. Only reach for transactions when you need atomic multi-partition writes or read-process-write exactly-once.

For Streams, use exactly_once_v2 and don't fight the runtime. Kafka Streams manages the entire transaction lifecycle per task. InvalidTxnStateException inside Streams almost always means an infrastructure problem (coordinator unreachable, timeout too low) or a known runtime bug — not something to catch and retry in your topology.

7. How to Prevent It Long-Term

Make the state machine impossible to misuse and observable when it breaks:

• Wrap the transactional loop once in a helper (or use Spring's transaction management) so no application code ever calls beginTransaction/commitTransaction directly. This eliminates Triggers A and B by construction.
• CI check: ban raw commitTransaction() calls that aren't guarded by a matching abortTransaction() in the catch path. An ArchUnit/Checkstyle rule or a simple grep gate catches the poisoned-commit pattern before it ships.
• Align timeouts as config-as-code: transaction.timeout.ms (client) ≤ transaction.max.timeout.ms (broker), and both comfortably above your checkpoint/commit interval. Document the arithmetic next to the config.
• Monitor transaction abort rate and the producer's txn-abort-time-ns / record-error-rate metrics. A rising abort rate with InvalidTxnStateException in logs points at Trigger B (poisoned commits) or C (timeouts).
• Load/chaos test at real latency: run your EOS sink with an injected processing stall longer than the transaction timeout in a Testcontainers cluster and assert the pipeline aborts and recovers cleanly instead of wedging.

8. Key Takeaways

• InvalidTxnStateException (broker error code 48, fatal/non-retriable) = a transactional API called from the wrong state, enforced on both client TransactionManager and the coordinator's __transaction_state.
• The #1 POC cause is missing beginTransaction() — beginTransaction is a client-side flag flip, so the failure surfaces later at send/commit.
• Never commitTransaction() after an abortable send error — the only legal exit from ABORTABLE_ERROR is abortTransaction().
• In Flink/Connect/streaming sinks it usually means transaction.timeout.ms is smaller than your commit cadence; the coordinator already aborted the transaction.
• Let the framework own the boundaries (Spring transaction-id-prefix, Streams exactly_once_v2), and only use transactions when you truly need atomic multi-partition or read-process-write exactly-once — otherwise plain idempotence is enough.