MINOR: Add optional acknowledgeType to ShareConsumeBenchSpec (#23802)

Add an optional "acknowledgeType" property (ACCEPT, REJECT or RELEASE)
to the Trogdor ShareConsumeBenchSpec. When set, the share consumers use
explicit acknowledgement mode and acknowledge every polled record with
that type. RENEW is not supported. When absent, behaviour is unchanged
(implicit acknowledgement), so existing specs and callers keep working.

The worker fails fast if acknowledgeType is combined with
share.acknowledgement.mode=implicit in the client configs. Also add an
acknowledge_type argument to the ducktape ShareConsumeBenchWorkloadSpec.
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

Reviewers: Andrew Schofield <aschofield@confluent.io>

---------

Co-authored-by: Claude Sonnet 5.5 <noreply@anthropic.com>

tests/kafkatest/services/trogdor/share_consume_bench_workload.py

@@ -21,7 +21,8 @@ from kafkatest.services.trogdor.task_spec import TaskSpec
 class ShareConsumeBenchWorkloadSpec(TaskSpec):
     def __init__(self, start_ms, duration_ms, consumer_node, bootstrap_servers,
                  target_messages_per_sec, max_messages, active_topics,
-                 consumer_conf, common_client_conf, admin_client_conf, share_group=None, threads_per_worker=1):
+                 consumer_conf, common_client_conf, admin_client_conf, share_group=None, threads_per_worker=1,
+                 acknowledge_type=None):
         super(ShareConsumeBenchWorkloadSpec, self).__init__(start_ms, duration_ms)
         self.message["class"] = "org.apache.kafka.trogdor.workload.ShareConsumeBenchSpec"
         self.message["consumerNode"] = consumer_node
@@ -35,6 +36,8 @@ class ShareConsumeBenchWorkloadSpec(TaskSpec):
         self.message["threadsPerWorker"] = threads_per_worker
         if share_group is not None:
             self.message["shareGroup"] = share_group
+        if acknowledge_type is not None:
+            self.message["acknowledgeType"] = acknowledge_type
 
 
 class ShareConsumeBenchWorkloadService(Service):

trogdor/src/main/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchSpec.java

@@ -17,6 +17,7 @@
 
 package org.apache.kafka.trogdor.workload;
 
+import org.apache.kafka.clients.consumer.AcknowledgeType;
 import org.apache.kafka.trogdor.common.StringExpander;
 import org.apache.kafka.trogdor.task.TaskController;
 import org.apache.kafka.trogdor.task.TaskSpec;
@@ -60,6 +61,11 @@ import java.util.Set;
  * The "recordProcessor" field allows the specification of tasks to run on records that are consumed.  This is run
  * immediately after the messages are polled.  See the `RecordProcessor` interface for more information.
  *
+ * The optional "acknowledgeType" field makes the share consumers use explicit acknowledgement mode and acknowledge
+ * every consumed record with the given type. Supported values are "ACCEPT", "REJECT" and "RELEASE". If it is not given, the
+ * share consumers use implicit acknowledgement. Note that with "RELEASE" the records are redelivered, so the
+ * "maxMessages" field bounds the total number of deliveries rather than the number of distinct records.
+ *
  * An example JSON representation which will result in a share consumer that is part of the share group "sg" and
  * subscribed to topics foo1, foo2, foo3 and bar.
  * #{@code
@@ -89,6 +95,7 @@ public final class ShareConsumeBenchSpec extends TaskSpec {
     private final String shareGroup;
     private final int threadsPerWorker;
     private final Optional<RecordProcessor> recordProcessor;
+    private final Optional<AcknowledgeType> acknowledgeType;
 
     @JsonCreator
     public ShareConsumeBenchSpec(@JsonProperty("startMs") long startMs,
@@ -103,7 +110,8 @@ public final class ShareConsumeBenchSpec extends TaskSpec {
                             @JsonProperty("adminClientConf") Map<String, String> adminClientConf,
                             @JsonProperty("threadsPerWorker") Integer threadsPerWorker,
                             @JsonProperty("recordProcessor") Optional<RecordProcessor> recordProcessor,
-                            @JsonProperty("activeTopics") List<String> activeTopics) {
+                            @JsonProperty("activeTopics") List<String> activeTopics,
+                            @JsonProperty("acknowledgeType") AcknowledgeType acknowledgeType) {
         super(startMs, durationMs);
         this.consumerNode = (consumerNode == null) ? "" : consumerNode;
         this.bootstrapServers = (bootstrapServers == null) ? "" : bootstrapServers;
@@ -116,6 +124,10 @@ public final class ShareConsumeBenchSpec extends TaskSpec {
         this.shareGroup = shareGroup == null ? DEFAULT_SHARE_GROUP_NAME : shareGroup;
         this.threadsPerWorker = threadsPerWorker == null ? 1 : threadsPerWorker;
         this.recordProcessor = recordProcessor;
+        if (acknowledgeType == AcknowledgeType.RENEW) {
+            throw new IllegalArgumentException("acknowledgeType must be ACCEPT, REJECT or RELEASE, but was " + acknowledgeType);
+        }
+        this.acknowledgeType = Optional.ofNullable(acknowledgeType);
     }
 
     @JsonProperty
@@ -153,6 +165,11 @@ public final class ShareConsumeBenchSpec extends TaskSpec {
         return this.recordProcessor;
     }
 
+    @JsonProperty
+    public Optional<AcknowledgeType> acknowledgeType() {
+        return acknowledgeType;
+    }
+
     @JsonProperty
     public Map<String, String> consumerConf() {
         return consumerConf;

trogdor/src/main/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchWorker.java

@@ -17,6 +17,7 @@
 
 package org.apache.kafka.trogdor.workload;
 
+import org.apache.kafka.clients.consumer.AcknowledgeType;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
@@ -138,8 +139,16 @@ public class ShareConsumeBenchWorker implements TaskWorker {
             props.put(ConsumerConfig.CLIENT_ID_CONFIG, clientId);
             props.put(ConsumerConfig.GROUP_ID_CONFIG, shareGroup);
             props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 100000);
+            // an explicit acknowledge type requires the consumer to acknowledge each record itself
+            spec.acknowledgeType().ifPresent(type ->
+                    props.put(ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG, "explicit"));
             // these defaults maybe over-written by the user-specified commonClientConf or consumerConf
             WorkerUtils.addConfigsToProperties(props, spec.commonClientConf(), spec.consumerConf());
+            if (spec.acknowledgeType().isPresent()
+                    && "implicit".equalsIgnoreCase(String.valueOf(props.get(ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG)))) {
+                throw new IllegalArgumentException(String.format("acknowledgeType %s requires %s=explicit",
+                        spec.acknowledgeType().get(), ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG));
+            }
             return new ThreadSafeShareConsumer(new KafkaShareConsumer<>(props, new ByteArrayDeserializer(), new ByteArrayDeserializer()), clientId);
         }
 
@@ -203,6 +212,8 @@ public class ShareConsumeBenchWorker implements TaskWorker {
                     recordProcessor.ifPresent(processor -> processor.processRecords(records));
 
                     for (ConsumerRecord<byte[], byte[]> record : records) {
+                        // Explicit acknowledgement requires acknowledging each record individually.
+                        spec.acknowledgeType().ifPresent(type -> consumer.acknowledge(record, type));
                         messagesConsumed++;
                         long messageBytes = 0;
                         if (record.key() != null) {
@@ -221,6 +232,9 @@ public class ShareConsumeBenchWorker implements TaskWorker {
                     }
                     startBatchMs = Time.SYSTEM.milliseconds();
                 }
+                // Send the pending acknowledgements while the consumer is still a member of the share group.
+                // Records after the maxMessages break are deliberately left unacknowledged.
+                spec.acknowledgeType().ifPresent(__ -> consumer.commitSync());
             } catch (Exception e) {
                 WorkerUtils.abort(log, "ConsumeRecords", e, doneFuture);
             } finally {
@@ -465,6 +479,25 @@ public class ShareConsumeBenchWorker implements TaskWorker {
             }
         }
 
+        void acknowledge(ConsumerRecord<byte[], byte[]> record, AcknowledgeType type) {
+            this.consumerLock.lock();
+            try {
+                consumer.acknowledge(record, type);
+            } finally {
+                this.consumerLock.unlock();
+            }
+        }
+
+        void commitSync() {
+            this.consumerLock.lock();
+            try {
+                consumer.commitSync().forEach((partition, exception) ->
+                        exception.ifPresent(e -> log.warn("{} failed to acknowledge records of {}", clientId, partition, e)));
+            } finally {
+                this.consumerLock.unlock();
+            }
+        }
+
         void close() {
             if (closed)
                 return;

trogdor/src/test/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchSpecTest.java

@@ -16,6 +16,11 @@
  */
 package org.apache.kafka.trogdor.workload;
 
+import org.apache.kafka.clients.consumer.AcknowledgeType;
+import org.apache.kafka.trogdor.common.JsonUtil;
+
+import com.fasterxml.jackson.databind.JsonMappingException;
+
 import org.junit.jupiter.api.Test;
 
 import java.util.HashSet;
@@ -53,15 +58,46 @@ class ShareConsumeBenchSpecTest {
     public void testDefaultShareGroupName() {
         ShareConsumeBenchSpec shareConsumeBenchSpec = new ShareConsumeBenchSpec(0, 0, "node", "localhost",
                 123, 1234, null, Map.of(), Map.of(), Map.of(), 1,
-                Optional.empty(), List.of("abc"));
+                Optional.empty(), List.of("abc"), null);
         assertEquals("share", shareConsumeBenchSpec.shareGroup());
     }
 
+    @Test
+    public void testAcknowledgeTypeAbsentByDefault() throws Exception {
+        ShareConsumeBenchSpec spec = parseSpec(null);
+        assertEquals(Optional.empty(), spec.acknowledgeType());
+    }
+
+    @Test
+    public void testAcknowledgeTypeParsedFromJson() throws Exception {
+        assertEquals(Optional.of(AcknowledgeType.ACCEPT), parseSpec("ACCEPT").acknowledgeType());
+        assertEquals(Optional.of(AcknowledgeType.REJECT), parseSpec("REJECT").acknowledgeType());
+        assertEquals(Optional.of(AcknowledgeType.RELEASE), parseSpec("RELEASE").acknowledgeType());
+    }
+
+    @Test
+    public void testUnsupportedAcknowledgeTypeRaisesException() {
+        for (String unsupported : List.of("RENEW", "bogus")) {
+            assertThrows(JsonMappingException.class, () -> parseSpec(unsupported));
+        }
+        assertThrows(IllegalArgumentException.class, () -> new ShareConsumeBenchSpec(0, 0, "node", "localhost",
+                123, 1234, "sg-1", Map.of(), Map.of(), Map.of(), 1,
+                Optional.empty(), List.of("abc"), AcknowledgeType.RENEW));
+    }
+
+    private ShareConsumeBenchSpec parseSpec(String acknowledgeType) throws Exception {
+        String ackJson = acknowledgeType == null ? "" : String.format(", \"acknowledgeType\": \"%s\"", acknowledgeType);
+        String json = "{\"class\": \"org.apache.kafka.trogdor.workload.ShareConsumeBenchSpec\", " +
+                "\"startMs\": 0, \"durationMs\": 0, \"consumerNode\": \"node\", \"bootstrapServers\": \"localhost\", " +
+                "\"maxMessages\": 10, \"activeTopics\": [\"abc\"]" + ackJson + "}";
+        return JsonUtil.JSON_SERDE.readValue(json, ShareConsumeBenchSpec.class);
+    }
+
     private ShareConsumeBenchSpec shareConsumeBenchSpec(List<String> activeTopics) {
         return new ShareConsumeBenchSpec(0, 0, "node", "localhost",
                 123, 1234, "sg-1",
                 Map.of(), Map.of(), Map.of(), 1,
-                Optional.empty(), activeTopics);
+                Optional.empty(), activeTopics, null);
     }
 
 }