This page is meant as a template for writing a KIP. To create a KIP choose Tools->Copy on this page and modify with your content and replace the heading with the next KIP number and a description of your issue. Replace anything in italics with your own description.
Current state: Under Discussion
Discussion thread: here [Change the link from the KIP proposal email archive to your own email thread]
JIRA: [KAFKA-19883]()
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
Share Groups currently support only immediate acknowledgement modes (IMPLICIT/EXPLICIT). This causes data loss in distributed streaming frameworks:
1. Worker acknowledges records → Records removed from Kafka
2. Checkpoint fails before sink write
3. Records lost (acknowledged but never persisted)
Goal: Enable exactly-once read semantics via transactional acknowledgements.
We will be following the similar existing pattern we have in Kafka Producer Transactions.
- Apache Flink: Exactly-once checkpointing with Share Group sources
- Apache Spark: Structured Streaming with Share Group consumers
- Any coordinator-worker streaming framework requiring atomic acknowledgements
```java
public interface ShareConsumer<K, V> {
// Existing
void acknowledge(ConsumerRecord<K, V> record, AcknowledgeType type);
// New: Transactional acknowledgement
void acknowledge(ConsumerRecord<K, V> record, AcknowledgeType type,
String transactionId);
}
```
```java
public interface ShareTransactionCoordinator {
void beginTransaction(String transactionId, Duration timeout);
void prepareTransaction(String transactionId) throws TransactionPrepareException;
void commitTransaction(String transactionId) throws TransactionCommitException;
void abortTransaction(String transactionId);
List<PendingTransaction> listPendingTransactions(String groupId);
}
```
Config Default Description
share.acknowledgement.mode explicit Values: implicit, explicit, transactional
share.transaction.timeout.ms 60000 Maximum transaction duration before auto-abort
share.transaction.max.pending 5 Maximum concurrent transactions per group
Request Description
ShareAcknowledgeTransactional Acknowledge records within a transaction
ShareBeginTransaction Start a new batch transaction
SharePrepareTransaction Phase 1: Prepare transaction
ShareCommitTransaction Phase 2: Commit transaction
ShareAbortTransaction Abort transaction and release records
Metric Type Description
share-transaction-active Gauge Active transactions
share-transaction-prepare-time-ms Histogram Prepare latency
share-transaction-commit-time-ms Histogram Commit latency
share-transaction-abort-total Counter Aborted transactions
share-transaction-timeout-total Counter Timed-out transactions
Introduce a new `LOCKED` state for records within a transaction:
Record State Machine:
AVAILABLE --[poll()]--> ACQUIRED
ACQUIRED --[txn.acknowledge()]--> LOCKED
ACQUIRED --[acquisitionLockTimeout, no txn]--> AVAILABLE
LOCKED --[commit()]--> ACKNOWLEDGED (removed)
LOCKED --[abort() or timeout]--> AVAILABLE (retry)
Key: `LOCKED` state is governed by **transaction timeout**, not acquisition lock.
- EMPTY -> ONGOING -> PREPARE -> COMMITTED (success)
- EMPTY -> ONGOING -> PREPARE -> ABORTED (abort after prepare)
- EMPTY -> ONGOING -> ABORTED (early abort)
Two-Phase Commit Protocol Flow:
Step 1: Coordinator sends BeginBatchTxn(id) to Broker
Step 2: Executor-1 calls poll() to Broker - gets records
Step 3: Executor-2 calls poll() to Broker - gets records
Step 4: Executor-1 sends ack(txnId, records) to Broker - records become LOCKED
Step 5: Executor-2 sends ack(txnId, records) to Broker - records become LOCKED
Step 6: Coordinator sends PrepareBatchTxn(id) to Broker - state becomes PREPARE
Step 7: Broker responds "prepared" to Coordinator
Step 8: Coordinator sends CommitBatchTxn(id) to Broker - LOCKED records become ACKNOWLEDGED
Step 9: Broker responds "committed" to Coordinator
```
{groupId}-{transactionalIdPrefix}-{checkpointId}
Example: my-share-group-flink-job-42
```
Extend `__share_group_state` topic with transaction records:
```java
public class ShareTransactionState {
String transactionId;
String groupId;
TransactionState state; // ONGOING, PREPARE, COMMITTED, ABORTED
long createTimeMs;
long prepareTimeMs;
Map<TopicPartition, List<AcknowledgementBatch>> pendingAcks;
}
```
- ONGOING: Abort → Release records
- PREPARE: Coordinator decides (commit/abort)
- COMMITTED/ABORTED: Cleanup after retention
```java
// On coordinator startup
List<PendingTransaction> pending = coordinator.listPendingTransactions(groupId);
for (PendingTransaction txn : pending) {
if (txn.state == PREPARE && shouldCommit(txn)) {
coordinator.commitTransaction(txn.id); // Complete Phase 2
} else {
coordinator.abortTransaction(txn.id); // Rollback
}
}
```
All operations are idempotent by transaction ID:
- `beginTransaction(id)` - returns existing if already started
- `prepareTransaction(id)` - no-op if already prepared
- `commitTransaction(id)` - no-op if already committed
- `abortTransaction(id)` - no-op if already aborted/committed
```
ShareAcknowledgeTransactionalRequest => GroupId TransactionId [Topics]
GroupId => STRING
TransactionId => STRING
Topics => TopicId [Partitions]
Partitions => Partition [AcknowledgementBatches]
AcknowledgementBatches => FirstOffset LastOffset AcknowledgeType
```
```
ShareBeginTransactionRequest => GroupId TransactionId TimeoutMs
GroupId => STRING
TransactionId => STRING
TimeoutMs => INT64
```
```
SharePrepareTransactionRequest => GroupId TransactionId
GroupId => STRING
TransactionId => STRING
```
```
ShareCommitTransactionRequest => GroupId TransactionId
GroupId => STRING
TransactionId => STRING
```
- No breaking changes for existing Share Group users
- `transactional` mode is opt-in via `share.acknowledgement.mode` config
- Existing `implicit` and `explicit` modes continue to work unchanged
1. Phase 1: Add `transactional` acknowledgement mode (backward compatible)
2. Phase 2: Streaming frameworks (Flink, Spark) implement transactional sources
3. Phase 3: Documentation and best practices for exactly-once semantics
- No deprecation of existing modes planned
- `transactional` mode recommended for exactly-once use cases
- Transaction state machine transitions
- Idempotency of all transaction operations
- Timeout handling and auto-abort
- Record state transitions (ACQUIRED → LOCKED → ACKNOWLEDGED)
- End-to-end transaction commit/abort flows
- Multi-consumer transactions within same group
- Coordinator failure and recovery
- Network partition scenarios
- Flink checkpoint integration with transactional Share Groups
- Exactly-once delivery verification under failures
- Performance benchmarks comparing transactional vs non-transactional modes
- Stress testing with concurrent transactions
- Broker failure during PREPARE state
- Coordinator crash before/after prepare
- Consumer crash during transaction
- Network partitions between coordinator and brokers
If there are alternative ways of accomplishing the same thing, what were they? The purpose of this section is to motivate why the design is the way it is and not some other way.