...

None of these fields are producer-specific. The naming is historical -- producerId and producerEpoch could equally be called sessionId and sessionEpoch. The wire protocol RPCs (InitProducerId, AddPartitionsToTxn, EndTxn, TxnHeartbeat) carry only these identity fields. They do not carry a "client type" discriminator.

What KIP-939

...

Solved

KIP-939

...

enabled Kafka to act as a

...

formal participant in Two-Phase Commit (2PC) transactions by:

• Adding Recovery APIs: Introduced prepareTransaction()

...

• and completeTransaction(

...

• )

...

What remains unsolved. KIP-939 added the correct primitives but placed them on the wrong abstraction. Every transaction lifecycle method -- including prepareTransaction() and completeTransaction(), which have nothing to do with producing records -- lives on KafkaProducer. The implementation is locked inside TransactionManager (1999 lines, declared public but located in org.apache.kafka.clients.producer.internals -- an internal API by package convention, not by access modifier). The Producer<K, V> interface declares all transaction lifecycle methods, but the actual implementation -- coordinator discovery, epoch management, state machine (UNINITIALIZED, INITIALIZING, READY, IN_TRANSACTION, PREPARED_TRANSACTION, COMMITTING_TRANSACTION, ABORTING_TRANSACTION, ABORTABLE_ERROR, FATAL_ERROR), and RPC priority queue (PriorityQueue<TxnRequestHandler>) -- is inaccessible outside the producer package in practice.

This creates three problems that KIP-939 does not address:

1. Heavyweight recovery. To complete a prepared transaction after a crash using KIP-939, the recovery coordinator must create a full KafkaProducer instance (with RecordAccumulator, Sender thread, batching, compression, serializers) just to send a single EndTxn RPC. The Producer<K, V> interface is semantically a record producer -- it carries send(ProducerRecord), partitionsFor(), and serializer semantics that are irrelevant to transaction completion. A lightweight TransactionSession that sends only coordinator RPCs would avoid this overhead (KAFKA-20381).

2. Multi-entity transactions have no shared identity object. KIP-1289's AddShareAcksToTxnRequest (API 83) requires a share consumer to participate in the producer's transaction using the producer's (producerId, epoch). KIP-1302's exactly-once Kafka-to-Kafka sink requires binding share group acknowledgments and producer output records in a single atomic transaction. Neither KIP-939 nor the current Producer<K, V> interface provides a way for non-producer entities to hold and use transaction credentials. A method like addShareAcksToTransaction() has no natural home -- it is a share consumer operation that must be routed through the producer because only the producer holds the identity.

3. Wrapper duplication. Kafka Streams (StreamsProducer.transactionInFlight) and Kafka Connect (ExactlyOnceWorkerSourceTask.transactionOpen) each duplicate transaction state tracking because TransactionManager's state is not externally accessible. KIP-939 does not change this: the state machine remains internal to the producer.

1.2 Six Entities Already Need Transaction Participation

The Kafka ecosystem already has six distinct entity types that interact with the transaction coordinator, each working around the producer-centric API:

1.3 The Evidence: Flink Proved the Separation Works

Flink's KafkaCommitter already implements the pattern this KIP proposes to formalize:

...

• , allowing external systems to coordinate Kafka's commit/abort status.

• Resilient Resumption: Allowed new producers to resume a "prepared" transaction after a crash without automatically aborting it.

• Removing Timeouts: Provided an enable2Pc mode that sets transaction timeouts to infinity, preventing premature aborts during external coordination.

The Unsolved Problem

While KIP-939 added the right tools, it put them in the wrong place:

• Wrong Abstraction: Transaction methods (like preparing and completing) are forced into the KafkaProducer class, even though they don't involve producing records.

• Heavyweight Bloat: To simply commit a transaction, you are forced to instantiate a full KafkaProducer with its entire infrastructure (buffer pools, sender threads, serializers).

• Inaccessible Logic: The core "brain" of transactions—the state machine and coordinator logic—is buried in internal producer packages, making it impossible for other clients (like consumers or custom coordinators) to use transactions independently.

1.2 Entities Already Need Transaction Participation

The Kafka ecosystem already has six distinct entity types that interact with the transaction coordinator, each working around the producer-centric API:

The resumeTransaction() method uses reflection to set TransactionManager's private fields:

```java
void resumeTransaction(long producerId, short epoch) {
    Object txnMgr = getTransactionManager();
    synchronized (txnMgr) {
        setField(txnMgr, "producerIdAndEpoch", createProducerIdAndEpoch(producerId, epoch));
        transitionTransactionManagerStateTo(txnMgr, "READY");
        transitionTransactionManagerStateTo(txnMgr, "IN_TRANSACTION");
        setField(txnMgr, "transactionStarted", true);
    }
}
```

This reflection is fragile, undocumented, and breaks on internal refactors. This KIP replaces it with a supported public API.

1.4 The Wire Protocol Is Already Entity-Agnostic

...

The transaction coordinator does not know or care whether the sender is a producer, a consumer, a Streams task, or an external coordinator. The fencing logic (epoch comparison) works identically regardless of caller identity. The heartbeat RPC (

...

...

2. Public Interfaces

2.1 New Class: TransactionSession

...