DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
| Code Block |
|---|
// One-time setup
shareConsumer.subscribe(List.of("source-topic"));
producer.initTransactions(); // exactly once per producer instance
// CTP loop
while (running) {
ConsumerRecords<K, V> records = shareConsumer.poll(Duration.ofSeconds(5));
if (records.isEmpty()) continue;
try {
producer.beginTransaction();
Map<TopicIdPartition, Acknowledgements> acks = new HashMap<>();
for (ConsumerRecord<K, V> record : records) {
V output = process(record);
producer.send(new ProducerRecord<>("destination-topic", output));
acks.computeIfAbsent(
new TopicIdPartition(record.topicId(),
new TopicPartition(record.topic(), record.partition())),
k -> Acknowledgements.empty()
).add(record.offset(), AcknowledgeType.ACCEPT); // accumulate
}
// Compress acks to AcknowledgementBatch wire form and stage in the txn.
producer.sendShareAcknowledgementsToTransaction(
toBatches(acks), // Map<TopicIdPartition, List<AcknowledgementBatch>>
shareConsumer.shareGroupMetadata()
);
producer.commitTransaction(); // BLOCKS — true synchronisation point
// Records are now both produced AND acknowledged ATOMICALLY.
// Do NOT call shareConsumer.acknowledge(...) on these records.
// if called mistakenly then non-transactional acknowledge is rejected (TX_PENDING to ACKNOWLEDGED - InvalidRecordStateException); the transactional path works.
} catch (ProducerFencedException | UnsupportedVersionException fatal) {
throw fatal; // unrecoverable — kill the process
} catch (KafkaException abortable) {
producer.abortTransaction(); // BLOCKS — txn reverted; staged records revert to ACQUIRED on broker
// Records will be re-delivered on next poll(); retry naturally.
}
}
|
...
1. Drain in-flight TX_PENDING:
- Halt producers calling sendShareAcknowledgementsToTransaction
- Wait for transaction.timeout.ms (default 60s) for any abandoned txns to clear
2. kafka-features.sh downgrade --feature share.version --version 1
Broker rejects the downgrade RPC if any in-memory TX_PENDING exists, preventing data inconsistency.
Software downgrade (binary): must follow feature downgrade; standard rolling restart.
Test Plan
Rough implementation [The actual implementation will be phasewise with multiple smaller PRs]:
https://github.com/apache/kafka/pull/22357
The verification strategy focuses on state machine integrity and fault tolerance under high-concurrency and failure scenarios.
...