Producer (clients module)

public interface Producer<K, V> {

    /**
   * Stages share-group acknowledgements for atomic commit with the records produced in this
   * transaction. On the broker the records move to TX_PENDING until the transaction resolves:
   *   COMMIT: ACCEPT -> ACKNOWLEDGED; REJECT -> ARCHIVING (DLQ) or ARCHIVED (no DLQ).
   *   ABORT : all staged records -> AVAILABLE (lock released, memberId cleared, redeliverable
   *           to ANY member on next fetch).
   * Only ACCEPT and REJECT are valid in a transaction; RELEASE/RENEW/GAP -> InvalidRecordStateException.
   *
   * Threading: enqueues the request for the producer Sender thread and returns immediately;
   *   the broker has not processed it yet. Errors surface at commitTransaction()/abortTransaction().
   *
   * @throws IllegalStateException      no transaction in progress / non-transactional producer
   * @throws ProducerFencedException / InvalidProducerEpochException  stale producer
   * @throws UnsupportedVersionException  cluster does not advertise apiKey 94 (TxnShareAcknowledge)
   * @throws GroupAuthorizationException  principal lacks READ on the share group   // correction: READ, not WRITE
   * @throws KafkaException             other abortable errors (abort + retry the CTP loop)
   */
  void sendShareAcknowledgementsToTransaction(ShareAcknowledgements acknowledgements,
                                              ShareGroupMetadata groupMetadata) throws ProducerFencedException;
}

ShareConsumer (clients module)

public interface ShareConsumer<K, V> {

  /** Immutable snapshot of group identity (groupId, memberId, memberEpoch). Thread-safe. */
  ShareGroupMetadata shareGroupMetadata();
  
  /** Returns AND CLEARS the acks staged during processing, for the producer txn.
   *  Explicit acknowledgement mode only. */
  ShareAcknowledgements acknowledgementsForTransaction();

}

ShareGroupMetadata (new class in clients module, package o.a.k.clients.consumer)

/* Thread-safe, Immutable, Concurrent with poll/acknowledge */
public final class ShareGroupMetadata {
    public ShareGroupMetadata(String groupId, String memberId, int memberEpoch);
    public final String groupId();
    public final String memberId();
    public final int memberEpoch();
    @Override public boolean equals(Object other);
    @Override public int hashCode();
    @Override public String toString();
}

Request — TransactionalId, GroupId, ProducerId, ProducerEpoch, MemberId, MemberEpoch, then Topics[] → Partitions[] → AcknowledgementBatches[] {FirstOffset, LastOffset, AcknowledgeTypes[]}. Only
  ACCEPT(1)/REJECT(3) valid.

Response — ThrottleTimeMs, top-level ErrorCode, Responses[] → Partitions[] {PartitionIndex, ErrorCode, ErrorMessage}. (correction: no CurrentLeader/NodeEndpoints fields — the schema does not carry 
  leader-redirect fields.)
  - Top-level errors: GROUP/TOPIC/TRANSACTIONAL_ID_AUTHORIZATION_FAILED, TRANSACTIONAL_ID_NOT_FOUND, INVALID_PRODUCER_EPOCH/PRODUCER_FENCED, INVALID_PRODUCER_ID_MAPPING, INVALID_TXN_STATE, UNKNOWN_MEMBER_ID, STALE_MEMBER_EPOCH, TRANSACTION_ABORTABLE.
  - Per-partition: UNKNOWN_TOPIC_OR_PARTITION/UNKNOWN_TOPIC_ID, INVALID_RECORD_STATE, INVALID_REQUEST, KAFKA_STORAGE_ERROR, NOT_LEADER_OR_FOLLOWER.

Existing RPCs — behavior:
  - AddPartitionsToTxn (24): no schema change, but used broker-internally (correction: the KIP earlier claimed share partitions are NOT registered — they are). The TxnShareAcknowledge handler calls
  AddPartitionsToTxnManager.addOrVerifyTransaction(..., verifyOnly=false) to register __share_group_state-N as a real transaction participant.
  - WriteTxnMarkers (27): no schema change. On the __share_group_state partition the broker calls shareCoordinator.completeTransaction(...), which returns the affected Set<SharePartitionKey>; the broker then
  invalidates only those caches. (correction: it does not broadcast to all SharePartitions with per-record fencing — a SharePartitionManager.applyTxnMarker broadcast method exists but is unused on this path; 
  tests assert never().) 
  - TxnOffsetCommit (28), ApiVersions (18): no schema change. Broker advertises apiKey 94 only when share.version >= 3 (below).

New RPC: TxnShareAcknowledgeResponse (apiKey 94, v0)

Top-level supported errors:
- GROUP_AUTHORIZATION_FAILED
- TOPIC_AUTHORIZATION_FAILED
- TRANSACTIONAL_ID_AUTHORIZATION_FAILED
- TRANSACTIONAL_ID_NOT_FOUND
- INVALID_PRODUCER_EPOCH / PRODUCER_FENCED
- INVALID_PRODUCER_ID_MAPPING
- INVALID_TXN_STATE
- UNKNOWN_MEMBER_ID
- STALE_MEMBER_EPOCH
- TRANSACTION_ABORTABLE 
- UNKNOWN_SERVER_ERROR

Per-partition supported errors:
- UNKNOWN_TOPIC_OR_PARTITION / UNKNOWN_TOPIC_ID
- NOT_LEADER_OR_FOLLOWER (with CurrentLeader populated)
- INVALID_RECORD_STATE
- INVALID_REQUEST
- KAFKA_STORAGE_ERROR

  TxnShareAcknowledgeRequest.json

  {
    "apiKey": 94,
    "type": "request",
    "listeners": ["broker"],
    "name": "TxnShareAcknowledgeRequest",
    "validVersions": "0",
    "flexibleVersions": "0+",
    "fields": [
      { "name": "TransactionalId", "type": "string", "versions": "0+",
        "nullableVersions": "0+", "entityType": "transactionalId",
        "about": "The transactional ID of the producer." },
      { "name": "GroupId", "type": "string", "versions": "0+",
        "entityType": "groupId", "about": "The share group ID." },
      { "name": "ProducerId", "type": "int64", "versions": "0+",
        "entityType": "producerId", "about": "The producer ID." },
      { "name": "ProducerEpoch", "type": "int16", "versions": "0+",
        "about": "The producer epoch." },
      { "name": "MemberId", "type": "string", "versions": "0+",
        "about": "The share group member ID." },
      { "name": "MemberEpoch", "type": "int32", "versions": "0+",
        "about": "The share group member epoch for fencing." },
      { "name": "Topics", "type": "[]TxnShareAcknowledgeTopic", "versions": "0+",
        "about": "The topics and partitions to acknowledge.", "fields": [
        { "name": "TopicId", "type": "uuid", "versions": "0+",
          "about": "The unique topic ID." },
        { "name": "Partitions", "type": "[]TxnShareAcknowledgePartition", "versions": "0+",
          "fields": [
          { "name": "PartitionIndex", "type": "int32", "versions": "0+",
            "about": "The partition index." },
          { "name": "AcknowledgementBatches", "type": "[]TxnShareAcknowledgeBatch", "versions": "0+",
            "about": "Record ranges to acknowledge. Only ACCEPT (1) and REJECT (3) are valid inside a transaction.",
            "fields": [
            { "name": "FirstOffset", "type": "int64", "versions": "0+",
              "about": "First offset of the batch (inclusive)." },
            { "name": "LastOffset", "type": "int64", "versions": "0+",
              "about": "Last offset of the batch (inclusive)." },
            { "name": "AcknowledgeTypes", "type": "[]int8", "versions": "0+",
              "about": "Per-offset acknowledge types: 1=ACCEPT, 3=REJECT. Size 1 = uniform type for whole range." }
          ]}
        ]}
      ]}
    ]
  }

TxnShareAcknowledgeResponse.json

  {
    "apiKey": 94,
    "type": "response",
    "name": "TxnShareAcknowledgeResponse",
    "validVersions": "0",
    "flexibleVersions": "0+",
    "fields": [
      { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
        "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
      { "name": "ErrorCode", "type": "int16", "versions": "0+",
        "about": "The top-level error code, or 0 if there was no error." },
      { "name": "Responses", "type": "[]TxnShareAcknowledgeTopicResponse", "versions": "0+",
        "about": "The results for each topic.", "fields": [
        { "name": "TopicId", "type": "uuid", "versions": "0+",
          "about": "The unique topic ID." },
        { "name": "Partitions", "type": "[]TxnShareAcknowledgePartitionResponse", "versions": "0+",
          "fields": [
          { "name": "PartitionIndex", "type": "int32", "versions": "0+",
            "about": "The partition index." },
          { "name": "ErrorCode", "type": "int16", "versions": "0+",
            "about": "The error code, or 0 if there was no error." },
          { "name": "ErrorMessage", "type": "string", "versions": "0+",
            "nullableVersions": "0+", "default": "null",
            "about": "The error message, or null if there was no error." }
        ]}
      ]}
    ]
  }

shareConsumer.subscribe(List.of("source-topic"));   // explicit acknowledgement mode
producer.initTransactions();
  while (running) {
    var records = shareConsumer.poll(Duration.ofSeconds(5));
    if (records.isEmpty()) continue;
    try {
      producer.beginTransaction();
      for (var rec : records) {
        producer.send(new ProducerRecord<>("destination-topic", process(rec)));
        shareConsumer.acknowledge(rec, AcknowledgeType.ACCEPT);          // accumulate acks
      }
      producer.sendShareAcknowledgementsToTransaction(
          shareConsumer.acknowledgementsForTransaction(),                // harvests + clears
          shareConsumer.shareGroupMetadata());
      producer.commitTransaction();                                     // atomic: produced AND acked
    } catch (ProducerFencedException | UnsupportedVersionException fatal) {
      throw fatal;
    } catch (KafkaException abortable) {
      producer.abortTransaction();                                      // staged records -> AVAILABLE, redelivered
    }
  }