KAFKA-18900: Implement share.acknowledgement.mode to choose acknowledgement mode (#19417)

Choose the acknowledgement mode based on the config
(`share.acknowledgement.mode`) and not on the basis of how the user
designs the application.
- The default value of the config is `IMPLICIT`, so if any
empty/null/invalid value is configured, then the mode defaults to
`IMPLICIT`.
- Removed AcknowledgementModes `UNKNOWN` and `PENDING` as they are no
longer required.
- Added code to ensure if the application has any unacknowledged records
in a batch in "`explicit`" mode, then it will throw an
`IllegalStateException`. The expectation is if the mode is "explicit",
all the records received in that `poll()` would be acknowledged before
the next call to `poll()`.
- Modified the `ConsoleShareConsumer` to configure the mode to
"explicit" as it was using the explicit mode of acknowledging records.

Reviewers: Andrew Schofield <[email protected]>

Author: ShivsundarR

Diff for: clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerTest.java

@@ -20,6 +20,7 @@
 import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.MetadataRecoveryStrategy;
 import org.apache.kafka.clients.consumer.internals.AutoOffsetResetStrategy;
+import org.apache.kafka.clients.consumer.internals.ShareAcknowledgementMode;
 import org.apache.kafka.common.IsolationLevel;
 import org.apache.kafka.common.config.AbstractConfig;
 import org.apache.kafka.common.config.ConfigDef;
@@ -381,13 +382,10 @@ public class ConsumerConfig extends AbstractConfig {
     private static final String SECURITY_PROVIDERS_DOC = SecurityConfig.SECURITY_PROVIDERS_DOC;
 
     /**
-     * <code>share.acknowledgement.mode</code> is being evaluated as a new configuration to control the acknowledgement mode
-     * for share consumers. It will be removed or converted to a proper configuration before release.
-     * An alternative being considered is <code>enable.explicit.share.acknowledgement</code> as a boolean configuration.
+     * <code>share.acknowledgement.mode</code>
      */
-    public static final String INTERNAL_SHARE_ACKNOWLEDGEMENT_MODE_CONFIG = "internal.share.acknowledgement.mode";
-    private static final String INTERNAL_SHARE_ACKNOWLEDGEMENT_MODE_DOC = "Controls the acknowledgement mode for a share consumer." +
-            " If unset, the acknowledgement mode of the consumer is decided by the method calls it uses to fetch and commit." +
+    public static final String SHARE_ACKNOWLEDGEMENT_MODE_CONFIG = "share.acknowledgement.mode";
+    private static final String SHARE_ACKNOWLEDGEMENT_MODE_DOC = "Controls the acknowledgement mode for a share consumer." +
             " If set to <code>implicit</code>, the acknowledgement mode of the consumer is implicit and it must not" +
             " use <code>org.apache.kafka.clients.consumer.ShareConsumer.acknowledge()</code> to acknowledge delivery of records. Instead," +
             " delivery is acknowledged implicitly on the next call to poll or commit." +
@@ -401,7 +399,7 @@ public class ConsumerConfig extends AbstractConfig {
      */
     private static final List<String> CLASSIC_PROTOCOL_UNSUPPORTED_CONFIGS = List.of(
             GROUP_REMOTE_ASSIGNOR_CONFIG,
-            INTERNAL_SHARE_ACKNOWLEDGEMENT_MODE_CONFIG
+            SHARE_ACKNOWLEDGEMENT_MODE_CONFIG
     );
 
     /**
@@ -411,7 +409,7 @@ public class ConsumerConfig extends AbstractConfig {
             PARTITION_ASSIGNMENT_STRATEGY_CONFIG, 
             HEARTBEAT_INTERVAL_MS_CONFIG, 
             SESSION_TIMEOUT_MS_CONFIG,
-            INTERNAL_SHARE_ACKNOWLEDGEMENT_MODE_CONFIG
+            SHARE_ACKNOWLEDGEMENT_MODE_CONFIG
     );
 
     static {
@@ -695,12 +693,12 @@ public class ConsumerConfig extends AbstractConfig {
                                         atLeast(0),
                                         Importance.LOW,
                                         CommonClientConfigs.METADATA_RECOVERY_REBOOTSTRAP_TRIGGER_MS_DOC)
-                                .define(ConsumerConfig.INTERNAL_SHARE_ACKNOWLEDGEMENT_MODE_CONFIG,
+                                .define(ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG,
                                         Type.STRING,
-                                        null,
-                                        in(null, "implicit", "explicit"),
+                                        ShareAcknowledgementMode.IMPLICIT.name(),
+                                        new ShareAcknowledgementMode.Validator(),
                                         Importance.MEDIUM,
-                                        ConsumerConfig.INTERNAL_SHARE_ACKNOWLEDGEMENT_MODE_DOC);
+                                        ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_DOC);
     }
 
     @Override

Diff for: clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java

@@ -116,33 +116,33 @@
  * {@code group.share.record.lock.partition.limit}. By limiting the duration of the acquisition lock and automatically
  * releasing the locks, the broker ensures delivery progresses even in the presence of consumer failures.
  * <p>
- * The consumer can choose to use implicit or explicit acknowledgement of the records it processes.
- * <p>If the application calls {@link #acknowledge(ConsumerRecord, AcknowledgeType)} for any record in the batch,
- * it is using <em>explicit acknowledgement</em>. In this case:
- * <ul>
- *     <li>The application calls {@link #commitSync()} or {@link #commitAsync()} which commits the acknowledgements to Kafka.
- *     If any records in the batch were not acknowledged, they remain acquired and will be presented to the application
- *     in response to a future poll.</li>
- *     <li>The application calls {@link #poll(Duration)} without committing first, which commits the acknowledgements to
- *     Kafka asynchronously. In this case, no exception is thrown by a failure to commit the acknowledgement.
- *     If any records in the batch were not acknowledged, they remain acquired and will be presented to the application
- *     in response to a future poll.</li>
- *     <li>The application calls {@link #close()} which attempts to commit any pending acknowledgements and
- *     releases any remaining acquired records.</li>
- * </ul>
- * If the application does not call {@link #acknowledge(ConsumerRecord, AcknowledgeType)} for any record in the batch,
- * it is using <em>implicit acknowledgement</em>. In this case:
+ * The consumer can choose to use implicit or explicit acknowledgement of the records it processes by configuring the
+ * consumer {@code share.acknowledgement.mode} property.
+ * <p>
+ * If the application sets the property to "implicit" or does not set it at all, then the consumer is using
+ * <em>implicit acknowledgement</em>. In this mode, the application acknowledges delivery by:
  * <ul>
- *     <li>The application calls {@link #commitSync()} or {@link #commitAsync()} which implicitly acknowledges all of
- *     the delivered records as processed successfully and commits the acknowledgements to Kafka.</li>
- *     <li>The application calls {@link #poll(Duration)} without committing, which also implicitly acknowledges all of
+ *     <li>Calling {@link #poll(Duration)} without committing, which also implicitly acknowledges all
  *     the delivered records and commits the acknowledgements to Kafka asynchronously. In this case, no exception is
  *     thrown by a failure to commit the acknowledgements.</li>
- *     <li>The application calls {@link #close()}  which releases any acquired records without acknowledgement.</li>
+ *     <li>Calling {@link #commitSync()} or {@link #commitAsync()} which implicitly acknowledges all
+ *     the delivered records as processed successfully and commits the acknowledgements to Kafka.</li>
+ *     <li>Calling {@link #close()} which releases any acquired records without acknowledgement.</li>
+ * </ul>
+ * If the application sets the property to "explicit", then the consumer is using <em>explicit acknowledgment</em>.
+ * The application must acknowledge all records returned from {@link #poll(Duration)} using
+ * {@link #acknowledge(ConsumerRecord, AcknowledgeType)} before its next call to {@link #poll(Duration)}.
+ * If the application calls {@link #poll(Duration)} without having acknowledged all records, an
+ * {@link IllegalStateException} is thrown. The remaining unacknowledged records can still be acknowledged.
+ * In this mode, the application acknowledges delivery by:
+ * <ul>
+ *     <li>Calling {@link #poll(Duration)} after it has acknowledged all records, which commits the acknowledgements
+ *     to Kafka asynchronously. In this case, no exception is thrown by a failure to commit the acknowledgements.</li>
+ *     <li>Calling {@link #commitSync()} or {@link #commitAsync()} which commits any pending
+ *     acknowledgements to Kafka.</li>
+ *     <li>Calling {@link #close()} which attempts to commit any pending acknowledgements and releases
+ *     any remaining acquired records.</li>
  * </ul>
- * <p>The consumer can optionally use the {@code internal.share.acknowledgement.mode} configuration property to choose
- * between implicit and explicit acknowledgement, specifying <code>"implicit"</code> or <code>"explicit"</code> as required.
- * <p>
  * The consumer guarantees that the records returned in the {@code ConsumerRecords} object for a specific topic-partition
  * are in order of increasing offset. For each topic-partition, Kafka guarantees that acknowledgements for the records
  * in a batch are performed atomically. This makes error handling significantly more straightforward because there can be
@@ -195,12 +195,14 @@
  *
  * <h4>Per-record acknowledgement (explicit acknowledgement)</h4>
  * This example demonstrates using different acknowledgement types depending on the outcome of processing the records.
+ * Here the {@code share.acknowledgement.mode} property is set to "explicit" so the consumer must explicitly acknowledge each record.
  * <pre>
  *     Properties props = new Properties();
  *     props.setProperty(&quot;bootstrap.servers&quot;, &quot;localhost:9092&quot;);
  *     props.setProperty(&quot;group.id&quot;, &quot;test&quot;);
  *     props.setProperty(&quot;key.deserializer&quot;, &quot;org.apache.kafka.common.serialization.StringDeserializer&quot;);
  *     props.setProperty(&quot;value.deserializer&quot;, &quot;org.apache.kafka.common.serialization.StringDeserializer&quot;);
+ *     props.setProperty(&quot;share.acknowledgement.mode&quot;, &quot;explicit&quot;);
  *     KafkaShareConsumer&lt;String, String&gt; consumer = new KafkaShareConsumer&lt;&gt;(props);
  *     consumer.subscribe(Arrays.asList(&quot;foo&quot;));
  *     while (true) {
@@ -227,42 +229,6 @@
  * It is only once {@link #commitSync()} is called that the acknowledgements are committed by sending the new state
  * information to Kafka.
  *
- * <h4>Per-record acknowledgement, ending processing of the batch on an error (explicit acknowledgement)</h4>
- * This example demonstrates ending processing of a batch of records on the first error.
- * <pre>
- *     Properties props = new Properties();
- *     props.setProperty(&quot;bootstrap.servers&quot;, &quot;localhost:9092&quot;);
- *     props.setProperty(&quot;group.id&quot;, &quot;test&quot;);
- *     props.setProperty(&quot;key.deserializer&quot;, &quot;org.apache.kafka.common.serialization.StringDeserializer&quot;);
- *     props.setProperty(&quot;value.deserializer&quot;, &quot;org.apache.kafka.common.serialization.StringDeserializer&quot;);
- *     KafkaShareConsumer&lt;String, String&gt; consumer = new KafkaShareConsumer&lt;&gt;(props);
- *     consumer.subscribe(Arrays.asList(&quot;foo&quot;));
- *     while (true) {
- *         ConsumerRecords&lt;String, String&gt; records = consumer.poll(Duration.ofMillis(100));
- *         for (ConsumerRecord&lt;String, String&gt; record : records) {
- *             try {
- *                 doProcessing(record);
- *                 consumer.acknowledge(record, AcknowledgeType.ACCEPT);
- *             } catch (Exception e) {
- *                 consumer.acknowledge(record, AcknowledgeType.REJECT);
- *                 break;
- *             }
- *         }
- *         consumer.commitSync();
- *     }
- * </pre>
- * There are the following cases in this example:
- * <ol>
- *     <li>The batch contains no records, in which case the application just polls again. The call to {@link #commitSync()}
- *     just does nothing because the batch was empty.</li>
- *     <li>All of the records in the batch are processed successfully. The calls to {@link #acknowledge(ConsumerRecord, AcknowledgeType)}
- *     specifying {@code AcknowledgeType.ACCEPT} mark all records in the batch as successfully processed.</li>
- *     <li>One of the records encounters an exception. The call to {@link #acknowledge(ConsumerRecord, AcknowledgeType)} specifying
- *     {@code AcknowledgeType.REJECT} rejects that record. Earlier records in the batch have already been marked as successfully
- *     processed. The call to {@link #commitSync()} commits the acknowledgements, but the records after the failed record
- *     remain acquired as part of the same delivery attempt and will be presented to the application in response to another poll.</li>
- * </ol>
- *
  * <h3>Reading Transactional Records</h3>
  * The way that share groups handle transactional records is controlled by the {@code group.share.isolation.level}</code>
  * configuration property. In a share group, the isolation level applies to the entire share group, not just individual

Diff for: clients/src/main/java/org/apache/kafka/clients/consumer/KafkaShareConsumer.java

@@ -0,0 +1,116 @@
+/*
+ * 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;
+
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.common.utils.Utils;
+
+import java.util.Arrays;
+import java.util.Locale;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+public class ShareAcknowledgementMode {
+    public enum AcknowledgementMode {
+        IMPLICIT, EXPLICIT;
+
+        @Override
+        public String toString() {
+            return super.toString().toLowerCase(Locale.ROOT);
+        }
+    }
+
+    private final AcknowledgementMode acknowledgementMode;
+
+    public static final ShareAcknowledgementMode IMPLICIT = new ShareAcknowledgementMode(AcknowledgementMode.IMPLICIT);
+    public static final ShareAcknowledgementMode EXPLICIT = new ShareAcknowledgementMode(AcknowledgementMode.EXPLICIT);
+
+    private ShareAcknowledgementMode(AcknowledgementMode acknowledgementMode) {
+        this.acknowledgementMode = acknowledgementMode;
+    }
+
+    /**
+     * Returns the ShareAcknowledgementMode from the given string.
+     */
+    public static ShareAcknowledgementMode fromString(String acknowledgementMode) {
+        if (acknowledgementMode == null) {
+            throw new IllegalArgumentException("Acknowledgement mode is null");
+        }
+
+        if (Arrays.asList(Utils.enumOptions(AcknowledgementMode.class)).contains(acknowledgementMode)) {
+            AcknowledgementMode mode = AcknowledgementMode.valueOf(acknowledgementMode.toUpperCase(Locale.ROOT));
+            switch (mode) {
+                case IMPLICIT:
+                    return IMPLICIT;
+                case EXPLICIT:
+                    return EXPLICIT;
+                default:
+                    throw new IllegalArgumentException("Invalid acknowledgement mode: " + acknowledgementMode);
+            }
+        } else {
+            throw new IllegalArgumentException("Invalid acknowledgement mode: " + acknowledgementMode);
+        }
+    }
+
+    /**
+     * Returns the name of the acknowledgement mode.
+     */
+    public String name() {
+        return acknowledgementMode.toString();
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        ShareAcknowledgementMode that = (ShareAcknowledgementMode) o;
+        return acknowledgementMode == that.acknowledgementMode;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(acknowledgementMode);
+    }
+
+    @Override
+    public String toString() {
+        return "ShareAcknowledgementMode{" +
+                "mode=" + acknowledgementMode +
+                '}';
+    }
+
+    public static class Validator implements ConfigDef.Validator {
+        @Override
+        public void ensureValid(String name, Object value) {
+            String acknowledgementMode = (String) value;
+            try {
+                fromString(acknowledgementMode);
+            } catch (Exception e) {
+                throw new ConfigException(name, value, "Invalid value `" + acknowledgementMode + "` for configuration " +
+                        name + ". The value must either be 'implicit' or 'explicit'.");
+            }
+        }
+
+        @Override
+        public String toString() {
+            String values = Arrays.stream(ShareAcknowledgementMode.AcknowledgementMode.values())
+                    .map(ShareAcknowledgementMode.AcknowledgementMode::toString).collect(Collectors.joining(", "));
+            return "[" + values + "]";
+        }
+    }
+}