KAFKA-19883: [DRAFT] KIP-1289 — Transactional acknowledgments for share groups (state machine + client API + wire schema)

clients/src/main/java/org/apache/kafka/clients/consumer/KafkaShareConsumer.java

@@ -837,6 +837,16 @@ public void wakeup() {
         delegate.wakeup();
     }
 
+    /**
+     * Returns a snapshot of the current share group membership metadata for this consumer.
+     * Pass to {@link org.apache.kafka.clients.producer.Producer#sendShareAcknowledgementsToTransaction}
+     * to atomically bind acknowledgments to a producer transaction (KIP-1289).
+     */
+    @Override
+    public ShareGroupMetadata shareGroupMetadata() {
+        return delegate.shareGroupMetadata();
+    }
+
     // Functions below are for testing only
     String clientId() {
         return delegate.clientId();

clients/src/main/java/org/apache/kafka/clients/consumer/MockShareConsumer.java

@@ -175,6 +175,11 @@ public synchronized void wakeup() {
         wakeup.set(true);
     }
 
+    @Override
+    public synchronized ShareGroupMetadata shareGroupMetadata() {
+        throw new UnsupportedOperationException("MockShareConsumer does not support shareGroupMetadata()");
+    }
+
     public synchronized void addRecord(ConsumerRecord<K, V> record) {
         ensureNotClosed();
         TopicPartition tp = new TopicPartition(record.topic(), record.partition());

clients/src/main/java/org/apache/kafka/clients/consumer/ShareConsumer.java

@@ -132,4 +132,9 @@ public interface ShareConsumer<K, V> extends Closeable {
      * @see KafkaShareConsumer#wakeup()
      */
     void wakeup();
+
+    /**
+     * @see KafkaShareConsumer#shareGroupMetadata()
+     */
+    ShareGroupMetadata shareGroupMetadata();
 }

clients/src/main/java/org/apache/kafka/clients/consumer/ShareGroupMetadata.java

@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer;
+
+import java.util.Objects;
+
+/**
+ * Metadata for the share group member obtained via {@link ShareConsumer#shareGroupMetadata()}.
+ * Pass this to {@link org.apache.kafka.clients.producer.Producer#sendShareAcknowledgementsToTransaction}
+ * to atomically bind acknowledgments to a producer transaction (KIP-1289).
+ */
+public class ShareGroupMetadata {
+
+    private final String groupId;
+    private final String memberId;
+    private final int memberEpoch;
+
+    public ShareGroupMetadata(String groupId, String memberId, int memberEpoch) {
+        this.groupId = Objects.requireNonNull(groupId, "groupId cannot be null");
+        this.memberId = Objects.requireNonNull(memberId, "memberId cannot be null");
+        this.memberEpoch = memberEpoch;
+    }
+
+    public String groupId() {
+        return groupId;
+    }
+
+    public String memberId() {
+        return memberId;
+    }
+
+    public int memberEpoch() {
+        return memberEpoch;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        ShareGroupMetadata that = (ShareGroupMetadata) o;
+        return memberEpoch == that.memberEpoch
+            && Objects.equals(groupId, that.groupId)
+            && Objects.equals(memberId, that.memberId);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(groupId, memberId, memberEpoch);
+    }
+
+    @Override
+    public String toString() {
+        return "ShareGroupMetadata(groupId=" + groupId
+            + ", memberId=" + memberId
+            + ", memberEpoch=" + memberEpoch
+            + ")";
+    }
+}

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java

@@ -22,6 +22,7 @@
 import org.apache.kafka.clients.GroupRebalanceConfig;
 import org.apache.kafka.clients.KafkaClient;
 import org.apache.kafka.clients.consumer.AcknowledgeType;
+import org.apache.kafka.clients.consumer.ShareGroupMetadata;
 import org.apache.kafka.clients.consumer.AcknowledgementCommitCallback;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -45,6 +46,7 @@
 import org.apache.kafka.clients.consumer.internals.events.ShareAcknowledgementEventHandler;
 import org.apache.kafka.clients.consumer.internals.events.ShareFetchEvent;
 import org.apache.kafka.clients.consumer.internals.events.SharePollEvent;
+import org.apache.kafka.clients.consumer.internals.events.ShareGroupMetadataEvent;
 import org.apache.kafka.clients.consumer.internals.events.ShareSubscriptionChangeEvent;
 import org.apache.kafka.clients.consumer.internals.events.ShareUnsubscribeEvent;
 import org.apache.kafka.clients.consumer.internals.events.StopFindCoordinatorOnCloseEvent;
@@ -1114,6 +1116,21 @@ public void wakeup() {
         wakeupTrigger.wakeup();
     }
 
+    @Override
+    public ShareGroupMetadata shareGroupMetadata() {
+        acquireAndEnsureOpen();
+        try {
+            ShareGroupMetadataEvent event = new ShareGroupMetadataEvent(
+                calculateDeadlineMs(time.timer(defaultApiTimeoutMs)));
+            applicationEventHandler.add(event);
+            return processBackgroundEvents(event.future(),
+                time.timer(defaultApiTimeoutMs),
+                e -> false);
+        } finally {
+            release();
+        }
+    }
+
     /**
      * Acquire the light lock and ensure that the consumer hasn't been closed.
      *

clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEvent.java

@@ -39,6 +39,7 @@ public enum Type {
         SHARE_SUBSCRIPTION_CHANGE, SHARE_UNSUBSCRIBE,
         SHARE_ACKNOWLEDGE_ON_CLOSE,
         SHARE_ACKNOWLEDGEMENT_COMMIT_CALLBACK_REGISTRATION,
+        SHARE_GROUP_METADATA,
         SEEK_UNVALIDATED,
         STREAMS_ON_TASKS_ASSIGNED_CALLBACK_COMPLETED,
         STREAMS_ON_TASKS_REVOKED_CALLBACK_COMPLETED,

clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java

@@ -91,6 +91,10 @@ public void process(ApplicationEvent event) {
                 process((SharePollEvent) event);
                 return;
 
+            case SHARE_GROUP_METADATA:
+                process((ShareGroupMetadataEvent) event);
+                return;
+
             case COMMIT_ASYNC:
                 process((AsyncCommitEvent) event);
                 return;
@@ -619,6 +623,16 @@ private void process(final ShareAcknowledgementCommitCallbackRegistrationEvent e
         manager.setAcknowledgementCommitCallbackRegistered(event.isCallbackRegistered());
     }
 
+    private void process(final ShareGroupMetadataEvent event) {
+        requestManagers.shareMembershipManager.ifPresentOrElse(
+            mgr -> event.future().complete(
+                new org.apache.kafka.clients.consumer.ShareGroupMetadata(
+                    mgr.groupId(), mgr.memberId(), mgr.memberEpoch())),
+            () -> event.future().completeExceptionally(
+                new IllegalStateException("No share group membership manager available"))
+        );
+    }
+
     private void process(final SeekUnvalidatedEvent event) {
         try {
             event.offsetEpoch().ifPresent(epoch -> metadata.updateLastSeenEpochIfNewer(event.partition(), epoch));

clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ShareGroupMetadataEvent.java

@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer.internals.events;
+
+import org.apache.kafka.clients.consumer.ShareGroupMetadata;
+
+/**
+ * Event requesting the current share group membership metadata snapshot from the background thread.
+ * Used by {@link org.apache.kafka.clients.consumer.internals.ShareConsumerImpl#shareGroupMetadata()}.
+ */
+public class ShareGroupMetadataEvent extends CompletableApplicationEvent<ShareGroupMetadata> {
+
+    public ShareGroupMetadataEvent(long deadlineMs) {
+        super(Type.SHARE_GROUP_METADATA, deadlineMs);
+    }
+}

clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java

@@ -26,6 +26,9 @@
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.clients.consumer.OffsetCommitCallback;
+import org.apache.kafka.clients.consumer.ShareGroupMetadata;
+import org.apache.kafka.clients.consumer.internals.AcknowledgementBatch;
+import org.apache.kafka.common.TopicIdPartition;
 import org.apache.kafka.clients.producer.internals.BufferPool;
 import org.apache.kafka.clients.producer.internals.BuiltInPartitioner;
 import org.apache.kafka.clients.producer.internals.KafkaProducerMetrics;
@@ -780,6 +783,22 @@ public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offs
         }
     }
 
+    @Override
+    public void sendShareAcknowledgementsToTransaction(
+            Map<TopicIdPartition, List<AcknowledgementBatch>> acknowledgements,
+            ShareGroupMetadata groupMetadata) throws ProducerFencedException {
+        throwIfNoTransactionManager();
+        throwIfProducerClosed();
+        throwIfInPreparedState();
+
+        if (!acknowledgements.isEmpty()) {
+            TransactionalRequestResult result =
+                transactionManager.sendShareAcknowledgementsToTransaction(acknowledgements, groupMetadata);
+            sender.wakeup();
+            result.await(maxBlockTimeMs, TimeUnit.MILLISECONDS, SEND_OFFSETS_TIMEOUT_MSG);
+        }
+    }
+
     /**
      * Prepares the current transaction for a two-phase commit. This method will flush all pending messages
      * and transition the producer into a mode where only {@link #commitTransaction()}, {@link #abortTransaction()},

clients/src/main/java/org/apache/kafka/clients/producer/MockProducer.java

@@ -18,6 +18,9 @@
 
 import org.apache.kafka.clients.consumer.ConsumerGroupMetadata;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.clients.consumer.ShareGroupMetadata;
+import org.apache.kafka.clients.consumer.internals.AcknowledgementBatch;
+import org.apache.kafka.common.TopicIdPartition;
 import org.apache.kafka.clients.producer.internals.FutureRecordMetadata;
 import org.apache.kafka.clients.producer.internals.ProduceRequestResult;
 import org.apache.kafka.common.Cluster;
@@ -209,6 +212,16 @@ public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offs
         this.sentOffsets = true;
     }
 
+    @Override
+    public void sendShareAcknowledgementsToTransaction(
+            Map<TopicIdPartition, List<AcknowledgementBatch>> acknowledgements,
+            ShareGroupMetadata groupMetadata) throws ProducerFencedException {
+        verifyNotClosed();
+        verifyNotFenced();
+        verifyTransactionsInitialized();
+        verifyTransactionInFlight();
+    }
+
     @Override
     public PreparedTxnState prepareTransaction() throws ProducerFencedException {
         verifyNotClosed();

clients/src/main/java/org/apache/kafka/clients/producer/Producer.java

@@ -18,9 +18,12 @@
 
 import org.apache.kafka.clients.consumer.ConsumerGroupMetadata;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.clients.consumer.ShareGroupMetadata;
+import org.apache.kafka.clients.consumer.internals.AcknowledgementBatch;
 import org.apache.kafka.common.Metric;
 import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicIdPartition;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.errors.ProducerFencedException;
@@ -62,6 +65,12 @@ default void initTransactions() {
     void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
                                   ConsumerGroupMetadata groupMetadata) throws ProducerFencedException;
 
+    /**
+     * See {@link KafkaProducer#sendShareAcknowledgementsToTransaction(Map, ShareGroupMetadata)}
+     */
+    void sendShareAcknowledgementsToTransaction(Map<TopicIdPartition, List<AcknowledgementBatch>> acknowledgements,
+                                               ShareGroupMetadata groupMetadata) throws ProducerFencedException;
+
     /**
      * See {@link KafkaProducer#prepareTransaction()}
      */