Status

Current state: Under discussion

Discussion thread: thread1

Voting thread: 

JIRA: 

Summary: Extract transaction identity and lifecycle management from `KafkaProducer` into a first-class
`TransactionSession` client abstraction, enabling any Kafka client (producer, share consumer, external coordinator, or any other entity) 
to participate in transactions without producer coupling or reflection hacks.

1. Motivation

1.1 The Problem: Multi-Participant Transactions Need a Shared Identity Abstraction

Kafka's transaction coordinator is a general-purpose distributed transaction manager. Its metadata (TransactionMetadata) stores:

```
transactionalId   -- session identity
producerId        -- session token
producerEpoch     -- fencing token
state             -- lifecycle state (ONGOING, PREPARE_COMMIT, ...)
topicPartitions   -- participating partitions
txnTimeoutMs      -- correctness bound
```

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 (Accepted) solved. KIP-939 made Kafka a proper 2PC participant by adding initTransactions(keepPreparedTxn=true), prepareTransaction(), and completeTransaction(PreparedTxnState) to KafkaProducer. This provides a public API for the 2PC recovery path: after a crash, a new producer can call initTransactions(true) to resume without aborting the prepared transaction, then completeTransaction() to commit or abort based on the externally-stored PreparedTxnState. KIP-939 also removes the transaction timeout for 2PC transactions (enable2Pc=true sets txnTimeoutMs = MAX_INT). These are essential primitives that this KIP builds upon.

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:

```java
// Flink separates transaction identity from producer instance
class KafkaCommittable {
    long producerId;
    short epoch;
    String transactionalId;
    // Optional: the producer that wrote the records (may be null)
}

// A DIFFERENT entity commits using only the identity
class KafkaCommitter {
    void commit(KafkaCommittable committable) {
        producer.resumeTransaction(committable.getProducerId(), committable.getEpoch());
        producer.commitTransaction();
    }
}
```

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

Every transaction RPC uses the same three identity fields:



```
TransactionalId (string)  -- "who is this transaction session?"
ProducerId      (int64)   -- "what is your session token?"
ProducerEpoch   (int16)   -- "are you the latest holder of this token?"
```

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 ( KAFKA-20381 - Getting issue details... STATUS /KIP-1309) carries these same three fields.

This KIP makes the client API match what the wire protocol already supports.

2. Public Interfaces

2.1 New Class: TransactionSession

Package: org.apache.kafka.clients.transaction

TransactionSession is a lightweight, thread-safe object that holds transaction identity and provides lifecycle operations. It does NOT produce records, consume records, or manage batching/serialization. It owns only the coordinator interaction.

```java
public class TransactionSession implements Closeable {

    // --- Construction ---

    /**
     * Create a transaction session with the given configuration.
     * Connects to the transaction coordinator for the transactionalId.
     */
    public TransactionSession(Map<String, Object> configs);

    /**
     * Create a transaction session from an existing identity.
     * Used by external coordinators (e.g., Flink KafkaCommitter)
     * that need to commit/abort a transaction started elsewhere.
     */
    public static TransactionSession resume(
        String transactionalId,
        long producerId,
        short producerEpoch,
        Map<String, Object> configs
    );

    // --- Lifecycle ---

    /**
     * Initialize the transaction session with the coordinator.
     * Sends InitProducerId, acquires producerId and epoch.
     * Aborts any pending transaction from a previous instance
     * with the same transactionalId (epoch fencing).
     */
    public void initialize();

    /**
     * Begin a new transaction. The session must be in READY state.
     */
    public void beginTransaction();

    /**
     * Prepare the transaction for two-phase commit (KIP-939).
     * Transitions state to PREPARED. The transaction persists
     * until commitTransaction() or abortTransaction() is called.
     */
    public PreparedTxnState prepareTransaction();

    /**
     * Complete a previously prepared transaction (KIP-939).
     * Called by an external coordinator to deliver the final
     * outcome of a 2PC transaction that was prepared with
     * prepareTransaction(). This may be called from a different
     * process using TransactionSession.resume().
     */
    public void completeTransaction(PreparedTxnState preparedTxnState);

    /**
     * Commit the current transaction.
     * Sends EndTxn(COMMIT) to the coordinator.
     */
    public void commitTransaction();

    /**
     * Abort the current transaction.
     * Sends EndTxn(ABORT) to the coordinator.
     */
    public void abortTransaction();

    // --- Identity (read-only after initialize) ---

    /** The transactional ID for this session. */
    public String transactionalId();

    /** The producer ID (session token) assigned by the coordinator. */
    public long producerId();

    /** The current epoch (fencing token). */
    public short producerEpoch();

    // --- Transaction Participation ---

    /**
     * Register topic partitions as part of this transaction.
     * Sends AddPartitionsToTxn to the coordinator.
     * Called by producers before writing to a partition.
     */
    public void addPartitionsToTransaction(Collection<TopicPartition> partitions);

    /**
     * Register consumer group offsets as part of this transaction.
     * Sends AddOffsetsToTxn + TxnOffsetCommit.
     * Called during read-process-write patterns.
     */
    public void sendOffsetsToTransaction(
        Map<TopicPartition, OffsetAndMetadata> offsets,
        ConsumerGroupMetadata groupMetadata
    );

    /**
     * Register share group acknowledgments as part of this transaction.
     * Sends AddShareAcksToTxn (KIP-1289, API).
     */
    public void addShareAcksToTransaction(
        String groupId,
        Collection<ShareAcknowledgment> acknowledgments
    );

    // --- Heartbeat (KIP-1309) ---

    /**
     * Send a liveness heartbeat to the transaction coordinator.
     * Automatically managed by the internal heartbeat thread
     * when transaction.session.timeout.ms > 0.
     * Can also be called explicitly for custom heartbeat patterns.
     */
    public void heartbeat();

    // --- Cleanup ---

    /** Close the session and release resources. */
    public void close();
}
```

2.2 Configuration

TransactionSession accepts a subset of existing producer configs plus one new config:

No new broker-side configs. No new wire protocol RPCs. TransactionSession sends the same RPCs that KafkaProducer sends today: FindCoordinator, InitProducerId, AddPartitionsToTxn, EndTxn, AddOffsetsToTxn, TxnOffsetCommit, TxnHeartbeat (KIP 1309).

2.3 Integration with KafkaProducer

KafkaProducer gains a new constructor and method to accept an external TransactionSession:

```java
public class KafkaProducer<K, V> {

    /**
     * Create a producer that uses an externally-managed transaction session.
     * The producer does NOT own the transaction lifecycle.
     * It uses the session's producerId/epoch for idempotent writes.
     * beginTransaction(), commitTransaction(), abortTransaction()
     * throw IllegalStateException -- use the TransactionSession directly.
     */
    public KafkaProducer(Map<String, Object> configs, TransactionSession session);

    /**
     * Get the transaction session for this producer.
     * Returns null if the producer is not transactional.
     * Returns the internal TransactionSession if transactional.id is configured.
     * Returns the external TransactionSession if constructed with one.
     */
    public TransactionSession transactionSession();
}
```

Backward compatibility: When KafkaProducer is constructed with transactional.id in the config (the current pattern), it internally creates a TransactionSession and delegates to it. The existing initTransactions(), beginTransaction(), commitTransaction(), abortTransaction(), sendOffsetsToTransaction() methods continue to work unchanged. They are convenience wrappers around the internal TransactionSession.

2.4 Integration with KafkaShareConsumer

For KIP-1289 transactional acknowledgments, the share consumer accepts a TransactionSession:

```java
public class KafkaShareConsumer<K, V> {

    /**
     * Acknowledge records transactionally within the given session.
     * Sends AddShareAcksToTxn (API 83) + TxnShareAcknowledge (API 84)
     * using the session's identity.
     */
    public void acknowledgeTransactionally(
        TransactionSession session,
        Map<TopicPartition, Set<Long>> acknowledgments
    );
}
```

3. Proposed Changes

3.1 Architecture: Before and After

Before (current):

After (this KIP):

3.2 Extracting TransactionSession from TransactionManager

The existing TransactionManager (1999 lines) is refactored into two components:

TransactionManager retains a reference to TransactionSession for identity information (producerId, epoch) and delegates lifecycle calls to it.

State machine mapping:

```
Current TransactionManager.State    -->    TransactionSession.State
---------------------------------          ----------------------
UNINITIALIZED                              UNINITIALIZED
INITIALIZING                               INITIALIZING
READY                                      READY
IN_TRANSACTION                             IN_TRANSACTION
PREPARED_TRANSACTION                       PREPARED
COMMITTING_TRANSACTION                     COMMITTING
ABORTING_TRANSACTION                       ABORTING
ABORTABLE_ERROR                            ABORTABLE_ERROR
FATAL_ERROR                                FATAL_ERROR
```

The state transitions are identical. The extraction is mechanical: move identity fields and lifecycle methods out of TransactionManager, leave produce-path integration in place.

3.3 No Wire Protocol Changes

This KIP does NOT introduce new RPCs or modify existing RPC schemas. TransactionSession sends the same RPCs that TransactionManager inside KafkaProducer sends today:

3.4 Internal Network Client

TransactionSession uses its own lightweight NetworkClient instance for coordinator communication. This is the same pattern used by the consumer's HeartbeatThread and the admin client's internal network handling. The NetworkClient is configured with:

• The same bootstrap.servers and security settings as the producer/consumer.

• A separate connection to the transaction coordinator broker.

• No RecordAccumulator, no batching, no Sender thread -- just request/response for transaction RPCs.

When a KafkaProducer is constructed with an external TransactionSession, the producer's Sender thread does NOT send transaction RPCs. It only sends Produce RPCs. The TransactionSession handles all coordinator communication independently. This clean separation eliminates the priority queue (PriorityQueue<TxnRequestHandler>) in the current TransactionManager that interleaves transaction RPCs with produce RPCs on the same Sender thread.

3.5 TransactionSession.resume() -- Eliminating Reflection

The static factory method TransactionSession.resume() replaces Flink's resumeTransaction() reflection hack:

```java
// Before (Flink -- reflection, fragile)
FlinkKafkaInternalProducer producer = new FlinkKafkaInternalProducer(configs);
producer.resumeTransaction(producerId, epoch);  // reflection on TransactionManager internals
producer.commitTransaction();

// After (this KIP -- public API, stable)
TransactionSession session = TransactionSession.resume(
    transactionalId, producerId, epoch, configs
);
session.commitTransaction();
session.close();
```

resume() creates a TransactionSession in IN_TRANSACTION state with the given identity, connects to the coordinator, and allows commitTransaction(), abortTransaction(), or completeTransaction() to be called.

Relationship to KIP-939's initTransactions(keepPreparedTxn=true): KIP-939 provides a public API for resuming a prepared transaction: call initTransactions(true), which sends InitProducerId with keepPreparedTxn=true and receives both the new epoch and the ongoing transaction's (OngoingTxnProducerId, OngoingTxnEpoch). This works correctly but requires a full KafkaProducer instance. TransactionSession.resume() achieves the same result -- re-establishing the transaction identity -- without the producer overhead. Internally, resume() may call InitProducerId(keepPreparedTxn=true) under the covers, but it does so through a lightweight network client rather than a full producer infrastructure. The two approaches are functionally equivalent; TransactionSession.resume() is an efficiency optimization, not a semantic change.

4. Use Cases

4.1 Apache Flink: Lightweight Transaction Completion

Flink's exactly-once KafkaSink separates the write path (writer subtask) from the commit path (committer). The committer must complete a transaction that was started by a different entity.

Three approaches compared:

A. Current Flink pattern (reflection -- pre-KIP-939):

```java
// Committer creates a full KafkaProducer, then uses reflection
FlinkKafkaInternalProducer<byte[], byte[]> committer = new FlinkKafkaInternalProducer<>(configs);
committer.resumeTransaction(producerId, epoch);  // Reflection on TransactionManager internals
committer.commitTransaction();
// Problem: fragile, breaks on Kafka client upgrades
```

B. KIP-939 pattern (public API -- no reflection, but heavyweight):

```java
// Committer creates a full KafkaProducer, then uses KIP-939's public API
KafkaProducer<byte[], byte[]> committer = new KafkaProducer<>(configs);
committer.initTransactions(true);  // keepPreparedTxn=true, sends InitProducerId RPC
// Internally: creates RecordAccumulator, Sender thread, serializers, compression...
// ...all unused, because we only need to send EndTxn

PreparedTxnState savedState = new PreparedTxnState(dbValue);
committer.completeTransaction(savedState);  // Sends EndTxn (COMMIT or ABORT)
committer.close();
// Problem: creates and destroys full producer infrastructure for a single RPC
```

KIP-939 eliminates the reflection hack -- this is a significant improvement. But the committer still instantiates a full KafkaProducer with RecordAccumulator (batching infrastructure), Sender thread (network I/O loop), serializers, partitioners, interceptors, and compression codecs. None of these are used. The committer sends exactly one RPC: EndTxn.

C. TransactionSession pattern (this KIP -- lightweight, public API):

```java
// Committer creates a lightweight TransactionSession, no producer infrastructure
TransactionSession session = TransactionSession.resume(txnId, producerId, epoch, configs);
// Internally: connects to transaction coordinator only, no RecordAccumulator, no Sender
session.commitTransaction();  // Sends EndTxn (COMMIT)
session.close();
// ~Simple lines of code vs  TransactionManager + full KafkaProducer
```

What changes in Flink connector:

• FlinkKafkaInternalProducer.resumeTransaction() reflection code deleted.

• KafkaCommitter uses TransactionSession.resume() instead of creating a KafkaProducer.

• KafkaCommittable stores TransactionSession identity fields (already does: producerId, epoch, transactionalId).

• Writer subtask uses TransactionSession + KafkaProducer(configs, session) for the write path. The session is the shared identity; the producer is used only for send().

Relationship to KIP-939: This KIP does not replace KIP-939. It builds on KIP-939's primitives (keepPreparedTxn, PreparedTxnState, completeTransaction) and provides a lighter-weight client for the recovery/commit path. The underlying RPCs are identical.

4.2 Share Group Consumer: Transactional Acknowledgments (KIP-1289)

KIP-1289 defines AddShareAcksToTxnRequest (API 83) and TxnShareAcknowledgeRequest (API 84), but has no client-side implementation. With TransactionSession, the pattern is clean:

```java
// Producer opens the transaction
TransactionSession session = new TransactionSession(configs);
session.initialize();
session.beginTransaction();

KafkaProducer<K, V> producer = new KafkaProducer<>(producerConfigs, session);
producer.send(outputRecord);

// Share consumer acknowledges within the SAME transaction
shareConsumer.acknowledgeTransactionally(session, acknowledgments);

// Commit atomically: output records + share acks in one transaction
session.commitTransaction();
```

Without TransactionSession, the share consumer would need to receive the producerId and epoch from the producer through an out-of-band channel, then construct the AddShareAcksToTxnRequest manually. TransactionSession provides the shared identity object that makes this natural.

4.3 Kafka Streams: Cleaner Task Transaction Management

Current Streams pattern:

```

java
// StreamsProducer wraps KafkaProducer, manages transaction state
class StreamsProducer {
    private KafkaProducer<byte[], byte[]> producer;
    private boolean transactionInFlight;

    void initTransaction() { producer.initTransactions(); }
    void beginTransaction() {
        if (!transactionInFlight) {
            producer.beginTransaction();
            transactionInFlight = true;
        }
    }
    void commitTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
                           ConsumerGroupMetadata metadata) {
        producer.sendOffsetsToTransaction(offsets, metadata);
        producer.commitTransaction();
        transactionInFlight = false;
    }
}
```

With TransactionSession:

```

java
class StreamsProducer {
    private TransactionSession session;
    private KafkaProducer<byte[], byte[]> producer;

    void initTransaction() { session.initialize(); }
    void beginTransaction() { session.beginTransaction(); }
    void commitTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
                           ConsumerGroupMetadata metadata) {
        session.sendOffsetsToTransaction(offsets, metadata);
        session.commitTransaction();
    }
    // session.heartbeat() runs automatically via background thread (KIP 1309 / 
                    KAFKA-20381
                            -
            Getting issue details...
                                                STATUS
                )
}
```

The manual transactionInFlight tracking is replaced by TransactionSession's internal state machine. The heartbeat (KIP 1309) is automatic.

4.4 Kafka Connect: Exactly-Once Source Tasks

Current Connect pattern:

```java
// ExactlyOnceWorkerSourceTask
class ExactlyOnceWorkerSourceTask {
    private KafkaProducer<byte[], byte[]> producer;
    private boolean transactionOpen;

    void maybeBeginTransaction() {
        if (!transactionOpen) {
            producer.beginTransaction();
            transactionOpen = true;
        }
    }
    void commitTransaction() {
        producer.commitTransaction();
        transactionOpen = false;
    }
}
```

With TransactionSession:

```java
class ExactlyOnceWorkerSourceTask {
    private TransactionSession session;
    private KafkaProducer<byte[], byte[]> producer;

    void maybeBeginTransaction() { session.beginTransaction(); }
    void commitTransaction() { session.commitTransaction(); }
    // Transaction state is managed by session, not duplicated here
}
```

4.5 Custom 2PC Coordinators

Any application implementing distributed transactions across Kafka and external systems (databases, object stores) can use TransactionSession as the Kafka leg of the 2PC:

```java
// Phase 1: Prepare
TransactionSession kafkaSession = new TransactionSession(configs);
kafkaSession.initialize();
kafkaSession.beginTransaction();
producer.send(records);               // write to Kafka
database.prepareTransaction(dbTxnId); // prepare database leg
kafkaSession.prepareTransaction();    // prepare Kafka leg (KIP-939)

// Phase 2: Commit (possibly from a different process)
TransactionSession resumed = TransactionSession.resume(txnId, pid, epoch, configs);
resumed.commitTransaction();          // commit Kafka leg
database.commitTransaction(dbTxnId);  // commit database leg
```

4.6 Kafka Connect Sink with Share Groups: Exactly-Once Kafka-to-Kafka (KIP-1302)

KIP-1302 introduces WorkerShareSinkTask -- a Connect runtime class that drives SinkTask using a KafkaShareConsumer instead of a KafkaConsumer. This enables queue semantics (elastic scaling independent of partition count, no rebalances, no head-of-line blocking) for all sink connectors.

KIP-1302 defines two delivery modes:

The exactly-once mode is the critical integration point with this KIP. It requires three entities to participate in a single transaction:

1. KafkaProducer -- writes transformed output records to the output topic.

2. KafkaShareConsumer -- sends transactional acknowledgments (AddShareAcksToTxn, API 83) for the input records.

3. Transaction Coordinator -- coordinates the atomic commit of both output records and source acknowledgments.

The Problem Without KIP:

KIP-1302's exactly-once code path (from the KIP):

```java
// KIP-1302 exactly-once pattern -- WITHOUT TransactionSession
void iterationExactlyOnce() {
    ConsumerRecords<byte[], byte[]> records = shareConsumer.poll(pollTimeout);
    if (records.isEmpty()) return;

    producer.beginTransaction();

    try {
        for (SinkRecord record : convertMessages(records)) {
            producer.send(new ProducerRecord<>(outputTopic, record.key(), record.value()));
        }

        // Problem: to send AddShareAcksToTxnRequest (API), the caller
        // needs the transaction's producerId and producerEpoch.
        // Without this KIP, the only entity holding these credentials is the
        // KafkaProducer (via TransactionManager). This forces a hypothetical
        // sendShareAcksToTransaction() method onto the Producer interface --
        // a share consumer operation proxied through a record producer.
        // This method does NOT currently exist in the Kafka codebase.
        producer.sendShareAcksToTransaction(           // proposed API (not yet implemented)
            ShareAcknowledgements.fromRecords(records, AcknowledgeType.ACCEPT),
            shareConsumer.groupMetadata()
        );

        producer.commitTransaction();
    } catch (Exception e) {
        producer.abortTransaction();
    }
}
```

This works but has fundamental coupling issues:

1. sendShareAcksToTransaction() does not exist yet and would have to be added to KafkaProducer. This method has no business being on a record producer. It is a share consumer operation that happens to need the transaction identity. Without this KIP, KIP-1302's EOS phase has no clean home for this API -- it must either add it to the Producer<K, V> interface (polluting the record producer abstraction) or send AddShareAcksToTxnRequest directly from the worker with credentials extracted from the producer.

2. The transaction identity (producerId, epoch) flows through the producer, making it impossible for the share consumer to participate in the transaction independently.

3. If the Connect runtime wants to separate the poll/process thread from the commit thread (as Flink does), there is no way to hand off the transaction identity without reflection.

4. The heartbeat for transaction liveness (KIP 1309) must run inside the producer, even though the long-running phase is often the task.put() call (controlled by the Connect runtime, not the producer).

The Solution With this KIP:

```java
// KIP-1302 exactly-once pattern -- WITH TransactionSession
void iterationExactlyOnce() {
    ConsumerRecords<byte[], byte[]> records = shareConsumer.poll(pollTimeout);
    if (records.isEmpty()) return;

    // TransactionSession is the shared identity object.
    // Both producer and share consumer use it.
    session.beginTransaction();

    try {
        // Producer writes output records using the session's identity
        for (SinkRecord record : convertMessages(records)) {
            producer.send(new ProducerRecord<>(outputTopic, record.key(), record.value()));
        }

        // Share consumer acknowledges input records within the SAME transaction.
        // Natural API: the consumer calls the session directly.
        // No need for the producer to proxy share consumer operations.
        session.addShareAcksToTransaction(
            shareConsumer.groupMetadata().groupId(),
            ShareAcknowledgements.fromRecords(records, AcknowledgeType.ACCEPT)
        );

        session.commitTransaction();
        // Output records AND source acknowledgments commit atomically.
    } catch (Exception e) {
        session.abortTransaction();
        // Both output records AND source acknowledgments roll back.
        // Records will be re-delivered by the share coordinator.
    }
}
```

What changes in WorkerShareSinkTask:

```

java
class WorkerShareSinkTask extends WorkerTask {
    private final TransactionSession session;          // Shared identity
    private final KafkaProducer<byte[], byte[]> producer; // Uses session for writes
    private final KafkaShareConsumer<byte[], byte[]> shareConsumer;
    private final SinkTask task;

    void initialize() {
        // 1. Create transaction session (owns identity + heartbeat)
        this.session = new TransactionSession(txnConfigs);
        session.initialize();

        // 2. Create producer bound to session (no internal TransactionManager)
        this.producer = new KafkaProducer<>(producerConfigs, session);

        // 3. Create share consumer (independent, no transaction coupling)
        this.shareConsumer = new KafkaShareConsumer<>(shareConsumerConfigs);
        shareConsumer.subscribe(topics);

        // 4. Start task
        task.initialize(context);
        task.start(taskConfig);
    }
}
```

Architectural comparison:

Why this matters for KIP-1302 specifically:

1. Gives addShareAcksToTransaction() a proper home. Without this KIP, KIP-1302's EOS implementation must add this method to KafkaProducer or the Producer<K, V> interface -- neither of which is the right abstraction. TransactionSession.addShareAcksToTransaction() is the correct home: it is a transaction lifecycle operation that happens to originate from a share consumer.

2. Enables async commit. KIP-1302's WorkerShareSinkTask may want to separate the poll/process thread from the commit thread (similar to Flink's writer/committer split). With TransactionSession, the commit thread can call session.commitTransaction() without needing a producer instance.

3. Transaction heartbeat covers the full window. With the session owning the heartbeat (KIP 1309), liveness detection covers the entire transaction duration -- including task.put() latency, share consumer polling, and output record production. Without this KIP, the heartbeat would need to be in the producer, which may be idle during task.put().

4. Acquisition lock coordination. KIP-1302 requires share.acquisition.lock.timeout.ms to exceed task.put() + transaction commit latency. With TransactionSession.heartbeat() (KIP 1309) keeping the transaction alive, the acquisition lock timeout only needs to exceed task.put() latency, not the full transaction timeout. This reduces the acquisition lock window and improves re-delivery speed.

5. Compatibility, Deprecation, and Migration Plan

5.1 Full Backward Compatibility

KafkaProducer constructed with transactional.id in the config continues to work exactly as today. Internally, it creates a TransactionSession and delegates to it, but the public API (initTransactions(), beginTransaction(), commitTransaction(), abortTransaction(), sendOffsetsToTransaction()) is unchanged.

5.2 Deprecation Path

No deprecations. The KafkaProducer convenience methods remain the recommended API for simple produce-and-commit patterns. TransactionSession is for advanced use cases: 2PC, cross-entity transactions, external coordinators.

5.3 Migration Path

Phase 1: Introduce TransactionSession class. Internal refactoring of TransactionManager. All existing code continues to work.

Phase 2: Add KafkaProducer(configs, TransactionSession) constructor. Enables external session management. Flink, Connect, Streams can optionally adopt.

Phase 3: Add KafkaShareConsumer.acknowledgeTransactionally(session, acks). Enables KIP-1289 transactional ack implementation.

Each phase is independently deployable. Phase 1 is a pure refactoring with no public API changes.

5.4 Wire Protocol Compatibility

No wire protocol changes. No new RPCs. No new request/response versions. TransactionSession uses the same RPCs at the same versions as the current TransactionManager. A cluster running any version of Kafka that supports transactions (0.11+) can be used with TransactionSession.

6. Security

6.1 Authorization

TransactionSession requires the same ACLs as the current producer transaction API:

6.2 Session Sharing

When a TransactionSession is shared between a producer and a share consumer (Use Case 4.2), both entities operate under the same transactionalId and require the same ACLs. The session identity is not multiplied -- there is one session, one producerId, one epoch, regardless of how many clients use it.

7. Relationship to KIP-939 (Support Participation in 2PC)

KIP-939 (Accepted, authored by Artem Livshits) is the foundational KIP that enables Kafka to participate in externally-coordinated two-phase commit. This KIP  is explicitly designed as a complement to KIP-939, not a replacement. The following table clarifies the boundary:

What KIP-939 Got Right

1. Implicit prepare. KIP-939 rejected an explicit "prepare" RPC, keeping Kafka's existing "implicit prepare" (flush-as-prepare). The external coordinator tracks prepared state, not the broker. This avoids duplicating state and an extra synchronous operation on the transaction coordinator topic. TransactionSession.prepareTransaction() wraps this same implicit-prepare mechanism.

2. keepPreparedTxn flag. Separating "don't abort the ongoing transaction" from "enable 2PC" was correct. Flink needs keepPreparedTxn=true even without enable2Pc=true (for clusters that don't grant 2PC privileges). TransactionSession uses this flag transparently.

3. PreparedTxnState as serializable {producerId, epoch}. This is the portable transaction identity that can be stored in any database. TransactionSession.resume() accepts the same identity fields.

What KIP-939's Rejected Alternatives Reveal

KIP-939 explicitly rejected a HeartBeat RPC (page 11): "HeartBeat RPC definitely sounds like a 'good thing to do'. It is not clear, though, what would be the cases when we need to handle these situations differently." Since then, KIP-1309 KAFKA-20381 - Getting issue details... STATUS has been proposed to add exactly this heartbeat, with strong community support. The evolution from KIP-939's rejection to KIP-1309's proposal demonstrates that the ecosystem's needs have grown beyond KIP-939's original scope. TransactionSession provides the natural home for the heartbeat thread (KIP 1309), which runs independently of the producer's Sender thread.

Why TransactionSession Is Not Redundant With KIP-939

The key question: "If KIP-939 provides initTransactions(true) + completeTransaction(), why do we need TransactionSession?"

Answer: KIP-939 solved the protocol problem. this KIP solves the abstraction problem.

KIP-939 added the correct primitives to the producer API. But the producer API is the wrong abstraction for three emerging use cases that KIP-939 did not anticipate:

1. KIP-1289 (Transactional Share Acks): A share consumer must send AddShareAcksToTxnRequest using the producer's (producerId, epoch). There is no method for this on KafkaProducer today, and adding one would pollute the record-producer interface with share-consumer semantics.

2. KIP-1302 (Share Groups in Connect Sink): The exactly-once Kafka-to-Kafka path requires three entities (producer, share consumer, transaction coordinator) to participate in one transaction. The shared identity must be accessible to both producer and consumer without one proxying through the other.

3. KAFKA-20381 - Getting issue details... STATUS /KIP-1309 (Transaction Heartbeat): The heartbeat thread should belong to the transaction session, not the producer. The producer may be idle (no records to send) while the transaction is active during a long checkpoint. The heartbeat must continue independently.

TransactionSession is the missing abstraction that connects KIP-939's 2PC primitives, KIP-1289's multi-entity transactions, KIP-1302's Connect sink EOS, and KIP 1309's liveness detection into a coherent client-side architecture.

8. Test Plan

8.1 Unit Tests

8.2 Integration Tests

8.3 Compatibility Tests

9. Reference 

KIP-98 - Exactly Once Delivery and Transactional Messaging

KIP-939: Support Participation in 2PC

KIP-1289 Support Transactional Acknowledgments for Share Groups

KIP-1302: Support Share Groups (Queue Semantics) in Kafka Connect Sink Connectors

KIP-1309: Improve transaction liveness checking

KAFKA-20381 - Getting issue details... STATUS

10. Rejected Alternatives

--