# Transactions & EOS (/connectors/kafka/how-to/transactions) Exactly-once semantics (EOS) on the Kafka connector runs on the same coordinator protocol real Kafka uses: a transactional producer completes `InitProducerId` → `AddPartitionsToTxn` → transactional `Produce` → `EndTxn(commit|abort)`, a `read_committed` consumer never sees an aborted record, and a stale producer instance is fenced rather than silently allowed to keep writing. This page is the practical how-to; the full API-key/version table and the KIP-890 scope note live at [Capabilities](/connectors/kafka/reference/capabilities), and every wire error code at [Error Codes](/connectors/kafka/reference/error-codes). **EOS is V1 scope — no KIP-890 transaction protocol V2.** `EndTxn` writes a real in-log COMMIT/ABORT control marker and gives the same `(PID, epoch)` fencing real Kafka's coordinator does, but the producer epoch is **not** bumped on every `EndTxn` the way TV2 (`transaction.version=2`) requires. The practical residual: a stray, delayed produce from the *same* epoch, arriving after that transaction's own `EndTxn` has already resolved, can still be admitted into the producer's *next* transaction — the same upstream-shared ceiling any Kafka-protocol clone inherits until it implements TV2. See [Capabilities](/connectors/kafka/reference/capabilities) for the full scope statement. ## The transactional producer round-trip [#the-transactional-producer-round-trip] Set a stable `transactional.id`, and the client library drives the coordinator handshake for you — your code only calls begin, produce, and end. `kcat` drives a transaction only as a whole batch (it begins on the first record and commits when its input stream closes), which doesn't fit this step-by-step begin/produce/end walkthrough; the pinned Ruby client (`rdkafka`, the `karafka/rdkafka-ruby` gem) doesn't expose a transactional producer API at all today — both are omitted below rather than faked. `InitProducerId` itself is never called directly: the first begin-transaction call issues it for you, allocating the `(PID, epoch)` pair every subsequent call in this session is fenced against. ```go package main import ( "context" "log" "github.com/twmb/franz-go/pkg/kgo" ) func main() { ctx := context.Background() cl, err := kgo.NewClient( kgo.SeedBrokers("localhost:9092"), kgo.TransactionalID("orders-producer"), ) if err != nil { log.Fatalf("client: %v", err) } defer cl.Close() // BeginTransaction lazily drives InitProducerId on the first call. if err := cl.BeginTransaction(); err != nil { log.Fatalf("begin transaction: %v", err) } record := &kgo.Record{Topic: "orders", Value: []byte("txn-value")} if err := cl.ProduceSync(ctx, record).FirstErr(); err != nil { _ = cl.EndTransaction(ctx, kgo.TryAbort) log.Fatalf("produce: %v", err) } if err := cl.EndTransaction(ctx, kgo.TryCommit); err != nil { log.Fatalf("commit: %v", err) } log.Println("committed: txn-value") } ``` ```python from confluent_kafka import KafkaException, Producer producer = Producer({ "bootstrap.servers": "localhost:9092", "transactional.id": "orders-producer", }) producer.init_transactions() producer.begin_transaction() try: producer.produce("orders", value=b"txn-value") producer.commit_transaction() print("committed: txn-value") except KafkaException as err: if err.args[0].txn_requires_abort(): producer.abort_transaction() else: raise ``` ```java import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.errors.ProducerFencedException; import org.apache.kafka.common.serialization.StringSerializer; import java.util.Properties; public final class TransactionalProduce { public static void main(String[] args) { Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("transactional.id", "orders-producer"); props.put("key.serializer", StringSerializer.class.getName()); props.put("value.serializer", StringSerializer.class.getName()); try (KafkaProducer producer = new KafkaProducer<>(props)) { producer.initTransactions(); try { producer.beginTransaction(); producer.send(new ProducerRecord<>("orders", "txn-value")); producer.commitTransaction(); System.out.println("committed: txn-value"); } catch (ProducerFencedException fenced) { // A zombie holding a stale (PID, epoch) cannot recover from this — give up. throw fenced; } catch (Exception e) { producer.abortTransaction(); } } } } ``` ```javascript const { Kafka } = require("kafkajs"); const kafka = new Kafka({ brokers: ["localhost:9092"] }); const producer = kafka.producer({ transactionalId: "orders-producer", maxInFlightRequests: 1, idempotent: true, }); async function main() { await producer.connect(); const transaction = await producer.transaction(); try { await transaction.send({ topic: "orders", messages: [{ value: "txn-value" }] }); await transaction.commit(); console.log("committed: txn-value"); } catch (err) { await transaction.abort(); throw err; } finally { await producer.disconnect(); } } main().catch((err) => { console.error(err); process.exit(1); }); ``` ```csharp using Confluent.Kafka; var config = new ProducerConfig { BootstrapServers = "localhost:9092", TransactionalId = "orders-producer", }; using var producer = new ProducerBuilder(config).Build(); producer.InitTransactions(TimeSpan.FromSeconds(10)); producer.BeginTransaction(); try { producer.Produce("orders", new Message { Value = "txn-value" }); producer.CommitTransaction(); Console.WriteLine("committed: txn-value"); } catch (KafkaException) { producer.AbortTransaction(); throw; } ``` ```rust use rdkafka::config::ClientConfig; use rdkafka::producer::{BaseProducer, BaseRecord, Producer}; use rdkafka::util::Timeout; use std::time::Duration; fn main() { let producer: BaseProducer = ClientConfig::new() .set("bootstrap.servers", "localhost:9092") .set("transactional.id", "orders-producer") .set("enable.idempotence", "true") .create() .expect("producer creation failed"); producer.init_transactions(Timeout::Never).expect("init_transactions failed"); producer.begin_transaction().expect("begin_transaction failed"); producer .send(BaseRecord::to("orders").payload("txn-value").key("order-1")) .expect("send failed"); producer.flush(Duration::from_secs(10)).expect("flush failed"); match producer.commit_transaction(Timeout::Never) { Ok(()) => println!("committed: txn-value"), Err(err) => { eprintln!("commit failed, aborting: {err}"); producer .abort_transaction(Duration::from_secs(10)) .expect("abort_transaction failed"); } } } ``` ## `read_committed` isolation and the Last Stable Offset [#read_committed-isolation-and-the-last-stable-offset] A `read_committed` consumer never gets handed an aborted record. Internally, `Fetch` computes a **Last Stable Offset (LSO)** — the offset up to which every transaction has already decided — and, under `read_committed`, clamps what it serves to that boundary; `ListOffsets(latest)` returns the LSO instead of the high watermark while a transaction is still open. The filtering itself happens **client-side**: the broker still serves the raw aborted batch below the LSO, tagged in `AbortedTransactions`, and a conforming `read_committed` client (Java's `Fetcher`, franz-go) drops those records itself — never a server-side record filter. Setting the isolation level is a one-line client config on any librdkafka-based client: ```python consumer = Consumer({ "bootstrap.servers": "localhost:9092", "group.id": "orders-consumer", "isolation.level": "read_committed", # default is read_uncommitted }) ``` `kcat` exposes the same librdkafka property through `-X`, which makes it a handy way to verify `read_committed` isolation behavior on the consumer side: ```bash kcat -C -b localhost:9092 -t orders -G orders-consumer -X isolation.level=read_committed ``` kafkajs takes a different shape for the same setting — a boolean `readUncommitted` option on the consumer (default `false`, i.e. `read_committed` behavior), rather than a string-valued `isolation.level`. ## Consume-transform-produce [#consume-transform-produce] A consume-transform-produce loop needs one more coordinator round-trip beyond a plain transactional produce: the consumer's **input offsets** have to commit atomically with the **output records**, or a crash between the two would either lose or double-process a batch. `AddOffsetsToTxn` adds the consumer group's offset-commit partition to the open transaction, and `TxnOffsetCommit` stages the offsets themselves — both are resolved on `EndTxn(commit)` alongside the produced records, and both are discarded together on `EndTxn(abort)`. franz-go wraps the whole pattern in `GroupTransactSession`, so application code never calls `AddOffsetsToTxn`/`TxnOffsetCommit` directly — `Begin`/`End` drive them: ```go package main import ( "context" "log" "github.com/twmb/franz-go/pkg/kgo" ) func main() { ctx := context.Background() sess, err := kgo.NewGroupTransactSession( kgo.SeedBrokers("localhost:9092"), kgo.TransactionalID("orders-etl"), kgo.ConsumerGroup("orders-group"), kgo.ConsumeTopics("orders"), ) if err != nil { log.Fatalf("session: %v", err) } defer sess.Close() for { fetches := sess.PollFetches(ctx) if errs := fetches.Errors(); len(errs) > 0 { log.Fatalf("fetch: %v", errs) } if err := sess.Begin(); err != nil { log.Fatalf("begin: %v", err) } fetches.EachRecord(func(r *kgo.Record) { sess.Produce(ctx, &kgo.Record{Topic: "orders-processed", Value: r.Value}, nil) }) // End commits the produced records AND the consumed offsets atomically // (AddOffsetsToTxn + TxnOffsetCommit happen here), or aborts both together. if _, err := sess.End(ctx, kgo.TryCommit); err != nil { log.Fatalf("end: %v", err) } } } ``` The other pinned clients expose the same two-request pattern as a single call on the producer, taking the consumer's group metadata as an argument: | Client | Offset-commit call | | ------------------------ | ----------------------------------------------------------------------------------- | | Go (franz-go) | `GroupTransactSession.End` (wraps `AddOffsetsToTxn`+`TxnOffsetCommit` internally) | | Python (confluent-kafka) | `producer.send_offsets_to_transaction(offsets, consumer.consumer_group_metadata())` | | Java (kafka-clients) | `producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata())` | | JavaScript (kafkajs) | `transaction.sendOffsets({ consumerGroupId, topics })` | | C# (Confluent.Kafka) | `producer.SendOffsetsToTransaction(offsets, consumerGroupMetadata, timeout)` | | Rust (rdkafka) | `producer.send_offsets_to_transaction(&tpl, &consumer.group_metadata(), timeout)` | See [Consuming](/connectors/kafka/how-to/consuming) for manual offset control outside a transaction, and [Consumer Groups](/connectors/kafka/concepts/consumer-groups) for the group protocol underneath. **Consume-transform-produce needs consumer-group Write, not Read.** Real Kafka authorizes `AddOffsetsToTxn`/`TxnOffsetCommit` against the group's Read ACL. This connector requires **Write** on that route instead. A standard EOS client — `GroupTransactSession`, `sendOffsetsToTransaction`, `send_offsets_to_transaction` — commits its input offsets *only* through the producer's transaction and never calls plain `OffsetCommit`, so if the group principal has only Read, the first `TxnOffsetCommit` fails fatally with `GROUP_AUTHORIZATION_FAILED`(30). Grant the consumer group Write before running a consume-transform-produce workload against an authorized cluster. ## Producer fencing [#producer-fencing] Two producer instances sharing the same `transactional.id` — most commonly an application restarted without a clean shutdown of the previous instance — can't both be authoritative. Each successful `InitProducerId` bumps the epoch on record for that `transactional.id`; once a newer instance has taken over, the older one is **fenced**, not silently allowed to keep writing: | Code | Error | Where it fires | | ---- | ------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------ | | 47 | `INVALID_PRODUCER_EPOCH` | A `Produce` arrives carrying an epoch below the live `(PID, epoch)` on record — the zombie's own writes are the giveaway. Non-retriable. | | 90 | `PRODUCER_FENCED` | An `InitProducerId` names an epoch strictly above the live durable epoch, surfacing the same outcome on the coordinator RPC path instead of `Produce`. | Both are terminal for that producer instance: there is no retry that fixes a fenced producer short of the application creating a brand-new one. A related, non-fencing limit worth setting sensibly: the server enforces a configurable ceiling on the client's `transaction.timeout.ms` — `900000` (15 min) by default, operator-adjustable up to `86400000` (24 h) — and a negotiated timeout above that ceiling answers `INVALID_TRANSACTION_TIMEOUT`(50) at `InitProducerId`. See [Configuration reference](/configure/reference/connectors#kafka) for the exact field names, and [Error Codes](/connectors/kafka/reference/error-codes) for the rest of the transaction coordinator's error surface — `INVALID_TXN_STATE`(48), `CONCURRENT_TRANSACTIONS`(51), and `TRANSACTIONAL_ID_AUTHORIZATION_FAILED`(53) among them. ## Related [#related]