// Kafka-as-destination path (true EOS via TxnShareAcknowledge):
batch boundary (or per-record, depending on tx granularity):
    producer.beginTransaction()
    for each (record, output) in batch: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.send(destinationTopic, output)
    producer.sendShareAcknowledgementsToTransaction(commitTransaction();    // BLOCKS — true synchronisation point
        ackMap,// Records are now both produced AND acknowledged ATOMICALLY.
        // Do NOT call shareConsumer.acknowledge(...) on these records.
		//    if called mistakenly then non-transactional acknowledge is // record→ACCEPT for processed records
        shareConsumer.shareGroupMetadata()
    )rejected (TX_PENDING to ACKNOWLEDGED - InvalidRecordStateException); the transactional path works.

    } catch (ProducerFencedException | UnsupportedVersionException fatal) {
    producer.commitTransaction()
    //throw Nofatal; separate consumer.acknowledge() — the ACK is in the transaction.

-------------------

// Pure-external-sink path (at-least-once + idempotent destination):[this KIP is not changin anything of// this flow]
worker thread per record:unrecoverable — kill the process
    output} =catch process(recordKafkaException abortable) {
      externalSink  producer.write(output, idempotencyKey)abortTransaction();     // BLOCKS — txn  // user task does this
reverted; staged records revert to ACQUIRED on broker
     consumer.acknowledgeAsync(record, ACCEPT)
        .thenAccept(_ -> markComplete(record))   // Records will be re-delivered on next poll(); retry naturally.
    }
}