> ## Content Index > Fetch the complete content index at: https://www.ggorantala.dev/llms.txt > Use this file to discover other available public pages before exploring further. # Kafka InvalidTxnStateException: Causes and How to Fix It - URL: https://www.ggorantala.dev/kafka-invalidtxnstateexception-fix/ - Published: 2026-04-08T09:40:00.000Z - Updated: 2026-08-29T08:28:11.000Z - Description: InvalidTxnStateException means your producer called a transactional API in the wrong state. Here's the state machine, the exact triggers, and the fix. - Author: Gopi Gorantala - Tags: kafka-errors, InvalidTxnStateException, kafka-transactions, spring-kafka, exactly-one, kafka-producer, Java `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`: ```text org.apache.kafka.common.errors.InvalidTxnStateException: The producer attempted a transactional operation in an invalid state. ``` Wrapped through `KafkaProducer.commitTransaction()` it usually looks like: ```text 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: ```text 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`): ```yaml # 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 ``` ```bash 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()`. ```java 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 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`: ```java 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 ```diff 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): ```java producer.beginTransaction(); try { for (ProducerRecord 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): ```properties # 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: ```properties 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: ```java @Bean public KafkaTemplate kafkaTemplate(ProducerFactory pf) { return new KafkaTemplate<>(pf); } @Bean public ProducerFactory producerFactory() { Map 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 pf = new DefaultKafkaProducerFactory<>(cfg); pf.setTransactionIdPrefix("orders-tx-"); // enables transactions return pf; } ``` ```java // 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.