DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
No code changes are required for existing users.
2.4 Integration with KafkaShareConsumer
...
For KIP-1289 transactional acknowledgments, the share consumer accepts a TransactionSession:
```...
and Kafka Consumer
void bindTransactionSession(TransactionSession session);
void unbindTransactionSession();
...
3. Proposed Changes
3.1 Architecture: Before and After
Before (current):
After (this KIP):
3. Proposed Changes
3.1 Architecture: Before and After
Before (current):
After (this KIP):
3.2 Extracting TransactionSession from TransactionManager
| Component | Responsibility |
| TransactionSession (New) | Identity (ID/Epoch), Lifecycle State Machine, Coordinator RPCs, and Heartbeat thread. |
| TransactionManager (Slimmed) | Sequence numbers, in-flight partition management, and Sender thread interaction. |
State mapping:
| Current TransactionManager.State | New TransactionSession.State |
UNINITIALIZED / INITIALIZING | UNINITIALIZED / INITIALIZING |
READY / IN_TRANSACTION | READY / IN_TRANSACTION |
PREPARED_TRANSACTION | PREPARED |
COMMITTING_TRANSACTION | COMMITTING |
ABORTING_TRANSACTION | ABORTING |
ABORTABLE_ERROR / FATAL_ERROR | ABORTABLE_ERROR / FATAL_ERROR |
3.3 No Wire Protocol Changes
This KIP reuses existing RPCs and schemas. TransactionSession becomes the new sender for all transaction-related requests:
| RPC | Current Sender | New Sender |
FindCoordinator / InitProducerId | TransactionManager | TransactionSession |
AddPartitionsToTxn / EndTxn | TransactionManager | TransactionSession |
AddOffsetsToTxn / TxnOffsetCommit | TransactionManager | TransactionSession |
TxnHeartbeat (KIP-1309) | N/A | TransactionSession |
AddShareAcksToTxn (KIP-1289) | N/A | TransactionSession |
3.4 Internal Network Client
TransactionSession uses a lightweight NetworkClient for coordinator communication, similar to the consumer's HeartbeatThread.
Dedicated Connection - Maintains its own connection to the coordinator broker.
4. Use Cases
4.1 Apache Flink: Lightweight Transaction Completion
...
```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();
```
4.
...
2 Consumer / Share Consumer
```
// 1. Open shared session
TransactionSession session = new TransactionSession(configs);
session.initialize();
session.beginTransaction();
// 2. Producer writes records using session identity
KafkaProducer<K, V> producer = new KafkaProducer<>(producerConfigs, session);
producer.send(outputRecord);
// 3. Share consumer acknowledges within the SAME transaction
shareConsumer.acknowledgeTransactionally(session, acknowledgments);
// 4. Atomic commit for both entities
session.commitTransaction();
...
| Test | Description | ||
| Session Lifecycle | Verify transitions: UNINITIALIZED → INITIALIZING → READY → IN_TRANSACTION → COMMITTING → READY. | ||
| Session Resume | Verify resume() starts in IN_TRANSACTION and can execute commitTransaction(). | ||
| Heartbeat Logic | Verify heartbeat thread lifecycle based on transaction.session.timeout.ms. | ||
| Producer Fencing | Ensure KafkaProducer throws IllegalStateException on lifecycle calls when using an external session. | ||
| Backward Compatibility | Verify standard KafkaProducer transaction methods work via internal session delegation. | ||
| Epoch Fencing | Verify that a new session with the same transactional.id correctly fences the older session. | Identity Accessors | Ensure producerId(), producerEpoch(), and transactionalId() are accurate post-initialization. |
7.2 Integration Tests
Producer + external session E2E
...