From ec51537f3c6a1cf3955ce184e87dbc091ac67007 Mon Sep 17 00:00:00 2001 From: Sushant Mahajan Date: Mon, 19 Oct 2026 20:34:50 +0530 Subject: [PATCH 1/3] KAFKA-21307: Add share group DLQ setup and record acknowledger to ShareConsumeBenchSpec Add two optional fields to ShareConsumeBenchSpec: - shareGroupDLQConfig: the DLQ topic name, its PartitionsSpec and copyRecordEnable. Before the consumers start, ShareConsumeBenchWorker creates the DLQ topic, enables it for use as a DLQ, and sets the DLQ topic name and copy record configs on the share group. - recordAcknowledger: a RecordAcknowledger, following the RecordProcessor pattern, that returns the AcknowledgeType for each consumed record. RandomRecordAcknowledger is the first implementation. Its counts are reported in the worker status. recordAcknowledger makes the consumers use explicit acknowledgement mode and cannot be combined with acknowledgeType. Co-Authored-By: Claude Sonnet 5.5 --- .../workload/RandomRecordAcknowledger.java | 97 +++++++++++++++ .../trogdor/workload/RecordAcknowledger.java | 40 +++++++ .../workload/ShareConsumeBenchSpec.java | 44 ++++++- .../workload/ShareConsumeBenchWorker.java | 110 +++++++++++++++--- .../trogdor/workload/ShareGroupDLQConfig.java | 69 +++++++++++ .../workload/ShareConsumeBenchSpecTest.java | 86 +++++++++++++- 6 files changed, 425 insertions(+), 21 deletions(-) create mode 100644 trogdor/src/main/java/org/apache/kafka/trogdor/workload/RandomRecordAcknowledger.java create mode 100644 trogdor/src/main/java/org/apache/kafka/trogdor/workload/RecordAcknowledger.java create mode 100644 trogdor/src/main/java/org/apache/kafka/trogdor/workload/ShareGroupDLQConfig.java diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/RandomRecordAcknowledger.java b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/RandomRecordAcknowledger.java new file mode 100644 index 0000000000000..09522d2f93888 --- /dev/null +++ b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/RandomRecordAcknowledger.java @@ -0,0 +1,97 @@ +/* + * 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.trogdor.workload; + +import org.apache.kafka.clients.consumer.AcknowledgeType; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.trogdor.common.JsonUtil; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.databind.JsonNode; + +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.atomic.AtomicLong; + +/** + * This class acknowledges each record with a randomly chosen type. The percentage of records that are rejected and + * released is configurable, and all the remaining records are accepted. The number of records acknowledged with each + * type is present in the status from the `ShareConsumeBenchWorker` class. + * + * Example spec: + * { + * "type": "random", + * "rejectPercent": 20, + * "releasePercent": 10 + * } + * + * This will reject 20% of the records, release 10% of them and accept the remaining 70%. + */ +public class RandomRecordAcknowledger implements RecordAcknowledger { + private final int rejectPercent; + private final int releasePercent; + private final AtomicLong accepted = new AtomicLong(); + private final AtomicLong rejected = new AtomicLong(); + private final AtomicLong released = new AtomicLong(); + + @JsonCreator + public RandomRecordAcknowledger(@JsonProperty("rejectPercent") int rejectPercent, + @JsonProperty("releasePercent") int releasePercent) { + if (rejectPercent < 0 || releasePercent < 0 || rejectPercent + releasePercent > 100) { + throw new IllegalArgumentException(String.format( + "rejectPercent and releasePercent must be non-negative and sum to at most 100, but were %d and %d", + rejectPercent, releasePercent)); + } + this.rejectPercent = rejectPercent; + this.releasePercent = releasePercent; + } + + @JsonProperty + public int rejectPercent() { + return rejectPercent; + } + + @JsonProperty + public int releasePercent() { + return releasePercent; + } + + @Override + public AcknowledgeType acknowledge(ConsumerRecord record) { + int roll = ThreadLocalRandom.current().nextInt(100); + if (roll < rejectPercent) { + rejected.incrementAndGet(); + return AcknowledgeType.REJECT; + } else if (roll < rejectPercent + releasePercent) { + released.incrementAndGet(); + return AcknowledgeType.RELEASE; + } + accepted.incrementAndGet(); + return AcknowledgeType.ACCEPT; + } + + @Override + public JsonNode acknowledgerStatus() { + Map counts = new LinkedHashMap<>(); + counts.put("accepted", accepted.get()); + counts.put("rejected", rejected.get()); + counts.put("released", released.get()); + return JsonUtil.JSON_SERDE.valueToTree(counts); + } +} diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/RecordAcknowledger.java b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/RecordAcknowledger.java new file mode 100644 index 0000000000000..d5a92323ee01f --- /dev/null +++ b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/RecordAcknowledger.java @@ -0,0 +1,40 @@ +/* + * 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.trogdor.workload; + +import org.apache.kafka.clients.consumer.AcknowledgeType; +import org.apache.kafka.clients.consumer.ConsumerRecord; + +import com.fasterxml.jackson.annotation.JsonSubTypes; +import com.fasterxml.jackson.annotation.JsonTypeInfo; +import com.fasterxml.jackson.databind.JsonNode; + +/** + * RecordAcknowledger decides how each record consumed by a ShareConsumeBench workload is acknowledged. + * A single instance is shared by all the share consumers of the worker, so implementations must be thread safe. + */ +@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type") +@JsonSubTypes(value = { + @JsonSubTypes.Type(value = RandomRecordAcknowledger.class, name = "random"), +}) +public interface RecordAcknowledger { + /** + * Returns the acknowledge type to apply to the given record. It must be one of ACCEPT, REJECT or RELEASE. + */ + AcknowledgeType acknowledge(ConsumerRecord record); + JsonNode acknowledgerStatus(); +} diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchSpec.java b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchSpec.java index 62e68bb9c28d2..fdd2953850925 100644 --- a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchSpec.java +++ b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchSpec.java @@ -66,6 +66,14 @@ * 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. * + * The optional "recordAcknowledger" field is like "acknowledgeType", but lets the acknowledge type be decided per record + * by a `RecordAcknowledger`. It also makes the share consumers use explicit acknowledgement mode, and cannot be + * combined with "acknowledgeType". + * + * The optional "shareGroupDLQConfig" field sets up a dead-letter queue for the share group before the consumers start. + * See `ShareGroupDLQConfig`. Records rejected, or released until their delivery attempts are exhausted, are then written + * to the DLQ topic. + * * 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 @@ -96,6 +104,8 @@ public final class ShareConsumeBenchSpec extends TaskSpec { private final int threadsPerWorker; private final Optional recordProcessor; private final Optional acknowledgeType; + private final Optional recordAcknowledger; + private final Optional shareGroupDLQConfig; @JsonCreator public ShareConsumeBenchSpec(@JsonProperty("startMs") long startMs, @@ -111,7 +121,9 @@ public ShareConsumeBenchSpec(@JsonProperty("startMs") long startMs, @JsonProperty("threadsPerWorker") Integer threadsPerWorker, @JsonProperty("recordProcessor") Optional recordProcessor, @JsonProperty("activeTopics") List activeTopics, - @JsonProperty("acknowledgeType") AcknowledgeType acknowledgeType) { + @JsonProperty("acknowledgeType") Optional acknowledgeType, + @JsonProperty("recordAcknowledger") Optional recordAcknowledger, + @JsonProperty("shareGroupDLQConfig") Optional shareGroupDLQConfig) { super(startMs, durationMs); this.consumerNode = (consumerNode == null) ? "" : consumerNode; this.bootstrapServers = (bootstrapServers == null) ? "" : bootstrapServers; @@ -124,10 +136,24 @@ public ShareConsumeBenchSpec(@JsonProperty("startMs") long startMs, 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 = orEmpty(acknowledgeType); + this.recordAcknowledger = orEmpty(recordAcknowledger); + this.shareGroupDLQConfig = orEmpty(shareGroupDLQConfig); + validateAcknowledgement(this.acknowledgeType, this.recordAcknowledger); + } + + private static Optional orEmpty(Optional optional) { + return optional == null ? Optional.empty() : optional; + } + + private static void validateAcknowledgement(Optional acknowledgeType, + Optional recordAcknowledger) { + if (acknowledgeType.filter(type -> type == AcknowledgeType.RENEW).isPresent()) { + throw new IllegalArgumentException("acknowledgeType must be ACCEPT, REJECT or RELEASE, but was " + AcknowledgeType.RENEW); + } + if (acknowledgeType.isPresent() && recordAcknowledger.isPresent()) { + throw new IllegalArgumentException("acknowledgeType and recordAcknowledger cannot both be set"); } - this.acknowledgeType = Optional.ofNullable(acknowledgeType); } @JsonProperty @@ -170,6 +196,16 @@ public Optional acknowledgeType() { return acknowledgeType; } + @JsonProperty + public Optional recordAcknowledger() { + return recordAcknowledger; + } + + @JsonProperty + public Optional shareGroupDLQConfig() { + return shareGroupDLQConfig; + } + @JsonProperty public Map consumerConf() { return consumerConf; diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchWorker.java b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchWorker.java index 54d951af3be54..b07d78c25d2bd 100644 --- a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchWorker.java +++ b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchWorker.java @@ -17,16 +17,23 @@ package org.apache.kafka.trogdor.workload; +import org.apache.kafka.clients.admin.Admin; +import org.apache.kafka.clients.admin.AlterConfigOp; +import org.apache.kafka.clients.admin.ConfigEntry; +import org.apache.kafka.clients.admin.NewTopic; 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; import org.apache.kafka.clients.consumer.KafkaShareConsumer; +import org.apache.kafka.common.config.ConfigResource; +import org.apache.kafka.common.config.TopicConfig; import org.apache.kafka.common.internals.KafkaFutureImpl; import org.apache.kafka.common.serialization.ByteArrayDeserializer; import org.apache.kafka.common.utils.Time; import org.apache.kafka.common.utils.Utils; import org.apache.kafka.common.utils.internals.ThreadUtils; +import org.apache.kafka.coordinator.group.GroupConfig; import org.apache.kafka.trogdor.common.JsonUtil; import org.apache.kafka.trogdor.common.Platform; import org.apache.kafka.trogdor.common.WorkerUtils; @@ -42,6 +49,7 @@ import java.time.Duration; import java.util.ArrayList; +import java.util.Collection; import java.util.HashMap; import java.util.HashSet; import java.util.List; @@ -86,6 +94,17 @@ public void start(Platform platform, WorkerStatusTracker status, throw new IllegalStateException("ShareConsumeBenchWorker is already running."); } log.info("{}: Activating ShareConsumeBenchWorker with {}", id, spec); + if (spec.shareGroupDLQConfig().isPresent()) { + try { + setupDLQ(spec.shareGroupDLQConfig().get()); + } catch (Throwable e) { + running.set(false); + if (e instanceof Exception) { + throw (Exception) e; + } + throw new RuntimeException("Failed to set up the DLQ for share group " + shareGroup(), e); + } + } this.statusUpdater = new StatusUpdater(); this.executor = Executors.newScheduledThreadPool( spec.threadsPerWorker() + 2, // 1 thread for all the ConsumeStatusUpdater and 1 for the StatusUpdater @@ -96,6 +115,61 @@ public void start(Platform platform, WorkerStatusTracker status, executor.submit(new Prepare()); } + /** + * Creates the DLQ topic, enables it for use as a DLQ and configures the share group to use it. + */ + private void setupDLQ(ShareGroupDLQConfig dlqConfig) throws Throwable { + String dlqTopic = dlqConfig.topicName(); + NewTopic newTopic = dlqConfig.topicSpec().newTopic(dlqTopic); + WorkerUtils.createTopics(log, spec.bootstrapServers(), spec.commonClientConf(), spec.adminClientConf(), + Map.of(dlqTopic, newTopic), false); + + try (Admin adminClient = WorkerUtils.createAdminClient(spec.bootstrapServers(), spec.commonClientConf(), spec.adminClientConf())) { + // Also done for a pre-existing topic, which would not have been created with the DLQ config. + alterConfigs(adminClient, new ConfigResource(ConfigResource.Type.TOPIC, dlqTopic), List.of( + setOp(TopicConfig.ERRORS_DEADLETTERQUEUE_GROUP_ENABLE_CONFIG, "true"))); + alterConfigs(adminClient, new ConfigResource(ConfigResource.Type.GROUP, shareGroup()), List.of( + setOp(GroupConfig.ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG, dlqTopic), + setOp(GroupConfig.ERRORS_DEADLETTERQUEUE_COPY_RECORD_ENABLE_CONFIG, + String.valueOf(dlqConfig.copyRecordEnable())))); + } + log.info("{}: Share group {} will use {} as its DLQ topic.", id, shareGroup(), dlqTopic); + } + + private static AlterConfigOp setOp(String name, String value) { + return new AlterConfigOp(new ConfigEntry(name, value), AlterConfigOp.OpType.SET); + } + + private static void alterConfigs(Admin adminClient, ConfigResource resource, Collection ops) throws Exception { + try { + adminClient.incrementalAlterConfigs(Map.of(resource, ops)).all().get(60, TimeUnit.SECONDS); + } catch (Exception e) { + throw new RuntimeException("Failed to alter configs " + ops + " of " + resource, e); + } + } + + private String shareGroup() { + return spec.shareGroup(); + } + + private boolean explicitAcknowledgement() { + return spec.acknowledgeType().isPresent() || spec.recordAcknowledger().isPresent(); + } + + /** + * Returns the acknowledge type for the record, or empty if the consumers use implicit acknowledgement. + */ + private Optional acknowledgeTypeFor(ConsumerRecord record) { + if (spec.recordAcknowledger().isPresent()) { + AcknowledgeType type = spec.recordAcknowledger().get().acknowledge(record); + if (type == null || type == AcknowledgeType.RENEW) { + throw new IllegalStateException("RecordAcknowledger must return ACCEPT, REJECT or RELEASE, but returned " + type); + } + return Optional.of(type); + } + return spec.acknowledgeType(); + } + public class Prepare implements Runnable { @Override public void run() { @@ -140,21 +214,18 @@ private ThreadSafeShareConsumer consumer(String shareGroup, String 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")); + if (explicitAcknowledgement()) { + 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() + if (explicitAcknowledgement() && "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)); + throw new IllegalArgumentException(String.format("acknowledgeType or recordAcknowledger requires %s=explicit", + ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG)); } return new ThreadSafeShareConsumer(new KafkaShareConsumer<>(props, new ByteArrayDeserializer(), new ByteArrayDeserializer()), clientId); } - - private String shareGroup() { - return spec.shareGroup(); - } } public class ConsumeMessages implements Callable { @@ -213,7 +284,7 @@ public Void call() throws Exception { for (ConsumerRecord record : records) { // Explicit acknowledgement requires acknowledging each record individually. - spec.acknowledgeType().ifPresent(type -> consumer.acknowledge(record, type)); + acknowledgeTypeFor(record).ifPresent(type -> consumer.acknowledge(record, type)); messagesConsumed++; long messageBytes = 0; if (record.key() != null) { @@ -234,7 +305,9 @@ public Void call() throws Exception { } // 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()); + if (explicitAcknowledgement()) { + consumer.commitSync(); + } } catch (Exception e) { WorkerUtils.abort(log, "ConsumeRecords", e, doneFuture); } finally { @@ -336,6 +409,8 @@ StatusData update() { recordProcessorStatus = Optional.of(recordProcessor.get().processorStatus()); } + Optional recordAcknowledgerStatus = spec.recordAcknowledger().map(RecordAcknowledger::acknowledgerStatus); + StatusData statusData = new StatusData( consumer.subscription(), latSummary.numSamples(), @@ -345,7 +420,8 @@ StatusData update() { latSummary.percentiles().get(0).value(), latSummary.percentiles().get(1).value(), latSummary.percentiles().get(2).value(), - recordProcessorStatus); + recordProcessorStatus, + recordAcknowledgerStatus); statusUpdater.updateConsumeStatus(consumer.clientId(), statusData); log.info("Status={}", JsonUtil.toJsonString(statusData)); return statusData; @@ -362,6 +438,7 @@ public static class StatusData { private final int p95LatencyMs; private final int p99LatencyMs; private final Optional recordProcessorStatus; + private final Optional recordAcknowledgerStatus; /** * The percentiles to use when calculating the histogram data. @@ -377,7 +454,8 @@ public static class StatusData { @JsonProperty("p50LatencyMs") int p50latencyMs, @JsonProperty("p95LatencyMs") int p95latencyMs, @JsonProperty("p99LatencyMs") int p99latencyMs, - @JsonProperty("recordProcessorStatus") Optional recordProcessorStatus) { + @JsonProperty("recordProcessorStatus") Optional recordProcessorStatus, + @JsonProperty("recordAcknowledgerStatus") Optional recordAcknowledgerStatus) { this.subscription = subscription; this.totalMessagesReceived = totalMessagesReceived; this.totalBytesReceived = totalBytesReceived; @@ -387,6 +465,7 @@ public static class StatusData { this.p95LatencyMs = p95latencyMs; this.p99LatencyMs = p99latencyMs; this.recordProcessorStatus = recordProcessorStatus; + this.recordAcknowledgerStatus = recordAcknowledgerStatus; } @JsonProperty @@ -433,6 +512,11 @@ public int p99LatencyMs() { public JsonNode recordProcessorStatus() { return recordProcessorStatus.orElse(null); } + + @JsonProperty + public JsonNode recordAcknowledgerStatus() { + return recordAcknowledgerStatus.orElse(null); + } } @Override diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ShareGroupDLQConfig.java b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ShareGroupDLQConfig.java new file mode 100644 index 0000000000000..090a4ae5a0686 --- /dev/null +++ b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ShareGroupDLQConfig.java @@ -0,0 +1,69 @@ +/* + * 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.trogdor.workload; + +import org.apache.kafka.trogdor.rest.Message; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; + +/** + * Describes the dead-letter queue (DLQ) setup of the share group used by a ShareConsumeBench workload. + * + * Before the consumers start, the worker creates the DLQ topic as described by "topicSpec" (or with the default + * partitions spec if absent), enables it for use as a DLQ, and points the share group at it. If the topic already + * exists it is reused and only enabled as a DLQ. + * + * Example: + * { + * "topicName": "sg-dlq", + * "topicSpec": {"numPartitions": 1, "replicationFactor": 3}, + * "copyRecordEnable": true + * } + */ +public class ShareGroupDLQConfig extends Message { + private final String topicName; + private final PartitionsSpec topicSpec; + private final boolean copyRecordEnable; + + @JsonCreator + public ShareGroupDLQConfig(@JsonProperty("topicName") String topicName, + @JsonProperty("topicSpec") PartitionsSpec topicSpec, + @JsonProperty("copyRecordEnable") Boolean copyRecordEnable) { + if (topicName == null || topicName.isEmpty() || topicName.startsWith("__")) { + throw new IllegalArgumentException("topicName must be non-empty and must not start with \"__\", but was " + topicName); + } + this.topicName = topicName; + this.topicSpec = topicSpec == null ? new PartitionsSpec(0, (short) 0, null, null) : topicSpec; + this.copyRecordEnable = copyRecordEnable != null && copyRecordEnable; + } + + @JsonProperty + public String topicName() { + return topicName; + } + + @JsonProperty + public PartitionsSpec topicSpec() { + return topicSpec; + } + + @JsonProperty + public boolean copyRecordEnable() { + return copyRecordEnable; + } +} diff --git a/trogdor/src/test/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchSpecTest.java b/trogdor/src/test/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchSpecTest.java index 44d8709a06683..f24e1f6103882 100644 --- a/trogdor/src/test/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchSpecTest.java +++ b/trogdor/src/test/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchSpecTest.java @@ -17,6 +17,7 @@ package org.apache.kafka.trogdor.workload; import org.apache.kafka.clients.consumer.AcknowledgeType; +import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.trogdor.common.JsonUtil; import com.fasterxml.jackson.databind.JsonMappingException; @@ -30,7 +31,9 @@ import java.util.Set; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; class ShareConsumeBenchSpecTest { @@ -58,7 +61,7 @@ public void testInvalidNameRaisesException() { 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"), null); + Optional.empty(), List.of("abc"), Optional.empty(), Optional.empty(), Optional.empty()); assertEquals("share", shareConsumeBenchSpec.shareGroup()); } @@ -82,14 +85,89 @@ public void testUnsupportedAcknowledgeTypeRaisesException() { } 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)); + Optional.empty(), List.of("abc"), Optional.of(AcknowledgeType.RENEW), Optional.empty(), Optional.empty())); + } + + @Test + public void testRecordAcknowledgerAndDLQConfigAbsentByDefault() throws Exception { + ShareConsumeBenchSpec spec = parseSpec(null); + assertEquals(Optional.empty(), spec.recordAcknowledger()); + assertEquals(Optional.empty(), spec.shareGroupDLQConfig()); + } + + @Test + public void testRecordAcknowledgerParsedFromJson() throws Exception { + ShareConsumeBenchSpec spec = parseSpec(null, ", \"recordAcknowledger\": " + + "{\"type\": \"random\", \"rejectPercent\": 20, \"releasePercent\": 10}"); + RandomRecordAcknowledger acknowledger = (RandomRecordAcknowledger) spec.recordAcknowledger().get(); + assertEquals(20, acknowledger.rejectPercent()); + assertEquals(10, acknowledger.releasePercent()); + } + + @Test + public void testInvalidRecordAcknowledgerRaisesException() { + assertThrows(JsonMappingException.class, () -> parseSpec(null, ", \"recordAcknowledger\": " + + "{\"type\": \"random\", \"rejectPercent\": 80, \"releasePercent\": 30}")); + assertThrows(JsonMappingException.class, () -> parseSpec(null, ", \"recordAcknowledger\": " + + "{\"type\": \"random\", \"rejectPercent\": -1}")); + } + + @Test + public void testAcknowledgeTypeAndRecordAcknowledgerAreMutuallyExclusive() { + assertThrows(JsonMappingException.class, () -> parseSpec("REJECT", ", \"recordAcknowledger\": " + + "{\"type\": \"random\", \"rejectPercent\": 20}")); + } + + @Test + public void testRandomRecordAcknowledger() { + RandomRecordAcknowledger rejectAll = new RandomRecordAcknowledger(100, 0); + RandomRecordAcknowledger releaseAll = new RandomRecordAcknowledger(0, 100); + RandomRecordAcknowledger acceptAll = new RandomRecordAcknowledger(0, 0); + ConsumerRecord record = new ConsumerRecord<>("topic", 0, 0L, new byte[0], new byte[0]); + for (int i = 0; i < 20; i++) { + assertEquals(AcknowledgeType.REJECT, rejectAll.acknowledge(record)); + assertEquals(AcknowledgeType.RELEASE, releaseAll.acknowledge(record)); + assertEquals(AcknowledgeType.ACCEPT, acceptAll.acknowledge(record)); + } + assertEquals(20, rejectAll.acknowledgerStatus().get("rejected").asLong()); + assertEquals(0, rejectAll.acknowledgerStatus().get("accepted").asLong()); + } + + @Test + public void testShareGroupDLQConfigParsedFromJson() throws Exception { + ShareConsumeBenchSpec spec = parseSpec(null, ", \"shareGroupDLQConfig\": {\"topicName\": \"dlq\", " + + "\"topicSpec\": {\"numPartitions\": 2, \"replicationFactor\": 1}, \"copyRecordEnable\": true}"); + ShareGroupDLQConfig dlqConfig = spec.shareGroupDLQConfig().get(); + assertEquals("dlq", dlqConfig.topicName()); + assertEquals(2, dlqConfig.topicSpec().numPartitions()); + assertEquals(1, dlqConfig.topicSpec().replicationFactor()); + assertTrue(dlqConfig.copyRecordEnable()); + } + + @Test + public void testShareGroupDLQConfigDefaults() throws Exception { + ShareGroupDLQConfig dlqConfig = parseSpec(null, ", \"shareGroupDLQConfig\": {\"topicName\": \"dlq\"}") + .shareGroupDLQConfig().get(); + assertFalse(dlqConfig.copyRecordEnable()); + assertEquals(List.of(0), dlqConfig.topicSpec().partitionNumbers()); + } + + @Test + public void testInvalidShareGroupDLQTopicNameRaisesException() { + for (String json : List.of("{}", "{\"topicName\": \"\"}", "{\"topicName\": \"__dlq\"}")) { + assertThrows(JsonMappingException.class, () -> parseSpec(null, ", \"shareGroupDLQConfig\": " + json)); + } } private ShareConsumeBenchSpec parseSpec(String acknowledgeType) throws Exception { + return parseSpec(acknowledgeType, ""); + } + + private ShareConsumeBenchSpec parseSpec(String acknowledgeType, String extraJson) 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 + "}"; + "\"maxMessages\": 10, \"activeTopics\": [\"abc\"]" + ackJson + extraJson + "}"; return JsonUtil.JSON_SERDE.readValue(json, ShareConsumeBenchSpec.class); } @@ -97,7 +175,7 @@ private ShareConsumeBenchSpec shareConsumeBenchSpec(List activeTopics) { return new ShareConsumeBenchSpec(0, 0, "node", "localhost", 123, 1234, "sg-1", Map.of(), Map.of(), Map.of(), 1, - Optional.empty(), activeTopics, null); + Optional.empty(), activeTopics, Optional.empty(), Optional.empty(), Optional.empty()); } } \ No newline at end of file From 194ceab8c2b358657aa4d0043f7f45e8aa48c3c6 Mon Sep 17 00:00:00 2001 From: Sushant Mahajan Date: Mon, 19 Oct 2026 20:39:12 +0530 Subject: [PATCH 2/3] MINOR: Use plain constructor params for acknowledgeType and shareGroupDLQConfig Follow the existing spec convention: only pluggable components such as recordProcessor and recordAcknowledger take Optional constructor parameters. Simple values keep a nullable parameter that is wrapped in an Optional, which the getter returns. Co-Authored-By: Claude Sonnet 5.5 --- .../trogdor/workload/ShareConsumeBenchSpec.java | 14 +++++--------- .../workload/ShareConsumeBenchSpecTest.java | 6 +++--- 2 files changed, 8 insertions(+), 12 deletions(-) diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchSpec.java b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchSpec.java index fdd2953850925..1174963d22048 100644 --- a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchSpec.java +++ b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchSpec.java @@ -121,9 +121,9 @@ public ShareConsumeBenchSpec(@JsonProperty("startMs") long startMs, @JsonProperty("threadsPerWorker") Integer threadsPerWorker, @JsonProperty("recordProcessor") Optional recordProcessor, @JsonProperty("activeTopics") List activeTopics, - @JsonProperty("acknowledgeType") Optional acknowledgeType, + @JsonProperty("acknowledgeType") AcknowledgeType acknowledgeType, @JsonProperty("recordAcknowledger") Optional recordAcknowledger, - @JsonProperty("shareGroupDLQConfig") Optional shareGroupDLQConfig) { + @JsonProperty("shareGroupDLQConfig") ShareGroupDLQConfig shareGroupDLQConfig) { super(startMs, durationMs); this.consumerNode = (consumerNode == null) ? "" : consumerNode; this.bootstrapServers = (bootstrapServers == null) ? "" : bootstrapServers; @@ -136,16 +136,12 @@ public ShareConsumeBenchSpec(@JsonProperty("startMs") long startMs, this.shareGroup = shareGroup == null ? DEFAULT_SHARE_GROUP_NAME : shareGroup; this.threadsPerWorker = threadsPerWorker == null ? 1 : threadsPerWorker; this.recordProcessor = recordProcessor; - this.acknowledgeType = orEmpty(acknowledgeType); - this.recordAcknowledger = orEmpty(recordAcknowledger); - this.shareGroupDLQConfig = orEmpty(shareGroupDLQConfig); + this.acknowledgeType = Optional.ofNullable(acknowledgeType); + this.recordAcknowledger = recordAcknowledger == null ? Optional.empty() : recordAcknowledger; + this.shareGroupDLQConfig = Optional.ofNullable(shareGroupDLQConfig); validateAcknowledgement(this.acknowledgeType, this.recordAcknowledger); } - private static Optional orEmpty(Optional optional) { - return optional == null ? Optional.empty() : optional; - } - private static void validateAcknowledgement(Optional acknowledgeType, Optional recordAcknowledger) { if (acknowledgeType.filter(type -> type == AcknowledgeType.RENEW).isPresent()) { diff --git a/trogdor/src/test/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchSpecTest.java b/trogdor/src/test/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchSpecTest.java index f24e1f6103882..caf4cb7783b46 100644 --- a/trogdor/src/test/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchSpecTest.java +++ b/trogdor/src/test/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchSpecTest.java @@ -61,7 +61,7 @@ public void testInvalidNameRaisesException() { 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(), Optional.empty(), Optional.empty()); + Optional.empty(), List.of("abc"), null, Optional.empty(), null); assertEquals("share", shareConsumeBenchSpec.shareGroup()); } @@ -85,7 +85,7 @@ public void testUnsupportedAcknowledgeTypeRaisesException() { } 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"), Optional.of(AcknowledgeType.RENEW), Optional.empty(), Optional.empty())); + Optional.empty(), List.of("abc"), AcknowledgeType.RENEW, Optional.empty(), null)); } @Test @@ -175,7 +175,7 @@ private ShareConsumeBenchSpec shareConsumeBenchSpec(List activeTopics) { return new ShareConsumeBenchSpec(0, 0, "node", "localhost", 123, 1234, "sg-1", Map.of(), Map.of(), Map.of(), 1, - Optional.empty(), activeTopics, Optional.empty(), Optional.empty(), Optional.empty()); + Optional.empty(), activeTopics, null, Optional.empty(), null); } } \ No newline at end of file From 2b560be60c90297a2624452d21f571c33b6a9a36 Mon Sep 17 00:00:00 2001 From: Sushant Mahajan Date: Mon, 19 Oct 2026 20:41:10 +0530 Subject: [PATCH 3/3] MINOR: Allow RENEW from the ShareConsumeBench RecordAcknowledger Co-Authored-By: Claude Sonnet 5.5 --- .../org/apache/kafka/trogdor/workload/RecordAcknowledger.java | 2 +- .../kafka/trogdor/workload/ShareConsumeBenchWorker.java | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/RecordAcknowledger.java b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/RecordAcknowledger.java index d5a92323ee01f..f3332a3f31339 100644 --- a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/RecordAcknowledger.java +++ b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/RecordAcknowledger.java @@ -33,7 +33,7 @@ }) public interface RecordAcknowledger { /** - * Returns the acknowledge type to apply to the given record. It must be one of ACCEPT, REJECT or RELEASE. + * Returns the acknowledge type to apply to the given record. It must not be null. */ AcknowledgeType acknowledge(ConsumerRecord record); JsonNode acknowledgerStatus(); diff --git a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchWorker.java b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchWorker.java index b07d78c25d2bd..cbdfe70743e73 100644 --- a/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchWorker.java +++ b/trogdor/src/main/java/org/apache/kafka/trogdor/workload/ShareConsumeBenchWorker.java @@ -162,8 +162,8 @@ private boolean explicitAcknowledgement() { private Optional acknowledgeTypeFor(ConsumerRecord record) { if (spec.recordAcknowledger().isPresent()) { AcknowledgeType type = spec.recordAcknowledger().get().acknowledge(record); - if (type == null || type == AcknowledgeType.RENEW) { - throw new IllegalStateException("RecordAcknowledger must return ACCEPT, REJECT or RELEASE, but returned " + type); + if (type == null) { + throw new IllegalStateException("RecordAcknowledger must return an acknowledge type, but returned null"); } return Optional.of(type); }