Versions Compared

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.

...

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.

...