From ef51e45335824f3f1973eef6d2974d09e3619f5c Mon Sep 17 00:00:00 2001 From: Sanskar Jhajharia Date: Wed, 8 Jul 2026 10:39:59 +0530 Subject: [PATCH 1/5] Add DLQ Single Partition Tests --- tests/kafkatest/services/kafka/kafka.py | 71 +++++- .../services/verifiable_share_consumer.py | 41 +++- .../tests/client/share_consumer_dlq_test.py | 202 ++++++++++++++++++ .../kafka/tools/VerifiableShareConsumer.java | 90 +++++++- 4 files changed, 392 insertions(+), 12 deletions(-) create mode 100644 tests/kafkatest/tests/client/share_consumer_dlq_test.py diff --git a/tests/kafkatest/services/kafka/kafka.py b/tests/kafkatest/services/kafka/kafka.py index 8d98b96e2830b..2252e3d98d0b1 100644 --- a/tests/kafkatest/services/kafka/kafka.py +++ b/tests/kafkatest/services/kafka/kafka.py @@ -207,7 +207,8 @@ def __init__(self, context, num_nodes, zk, security_protocol=SecurityConfig.PLAI dynamicRaftQuorum=False, use_transactions_v2=False, use_streams_groups=False, - enable_assignment_batching=None + enable_assignment_batching=None, + share_version=None ): """ :param context: test context @@ -273,6 +274,7 @@ def __init__(self, context, num_nodes, zk, security_protocol=SecurityConfig.PLAI :param use_transactions_v2: When true, uses transaction.version=2 which utilizes the new transaction protocol introduced in KIP-890 :param use_streams_groups: When true, enables the use of streams groups introduced in KIP-1071 :param enable_assignment_batching: When true, enables assignment batching introduced in KIP-1263. If not specified, defaults to True. + :param share_version: When set, bootstraps the cluster with --feature share.version= (KIP-932/KIP-1191). """ self.zk = zk @@ -288,6 +290,7 @@ def __init__(self, context, num_nodes, zk, security_protocol=SecurityConfig.PLAI self.use_transactions_v2 = use_transactions_v2 self.use_streams_groups = use_streams_groups + self.share_version = share_version # Set consumer_group_migration_policy based on context and arguments. if consumer_group_migration_policy is None: @@ -361,7 +364,8 @@ def __init__(self, context, num_nodes, zk, security_protocol=SecurityConfig.PLAI server_prop_overrides=server_prop_overrides, dynamicRaftQuorum=self.dynamicRaftQuorum, use_transactions_v2=self.use_transactions_v2, use_streams_groups=self.use_streams_groups, - enable_assignment_batching=self.enable_assignment_batching + enable_assignment_batching=self.enable_assignment_batching, + share_version=self.share_version ) self.controller_quorum = self.isolated_controller_quorum @@ -786,6 +790,12 @@ def prop_file(self, node): override_configs[config_property.UNSTABLE_API_VERSIONS_ENABLE] = str(True) override_configs[config_property.UNSTABLE_FEATURE_VERSIONS_ENABLE] = str(True) + if self.share_version is not None and int(self.share_version) > 1: + # share.version=2 (KIP-1191 DLQ) is above ShareVersion.LATEST_PRODUCTION (SV_1), so both + # the storage format tool and the broker need unstable feature versions enabled to + # accept it -- kafka-storage.sh format reads this same properties file via --config. + override_configs[config_property.UNSTABLE_FEATURE_VERSIONS_ENABLE] = str(True) + if self.enable_assignment_batching: # Assignment batching is enabled by default in Kafka pass @@ -907,6 +917,12 @@ def start_node(self, node, timeout_sec=60, **kwargs): else: if get_version(node).supports_feature_command(): cmd += " --feature transaction.version=0" + if self.share_version is not None: + # share.version=2 (KIP-1191 DLQ) declares a bootstrap metadata.version of 4.4-IV0, + # but that isn't yet a valid --release-version for kafka-storage.sh format in this + # build (max supported is 4.3-IV0), and Feature.validateVersion() does not enforce + # the dependency, so the feature can be bootstrapped on its own. + cmd += " --feature share.version=%s" % self.share_version self.logger.info("Running log directory format command...\n%s" % cmd) node.account.ssh(cmd) @@ -1748,6 +1764,57 @@ def set_share_group_offset_reset_strategy(self, group, strategy=None, node=None, command_config) return "Completed" in self.run_cli_tool(node, cmd) + def set_share_group_dlq_config(self, group, topic_name=None, copy_record_enable=None, node=None, command_config=None): + """ Set the DLQ configs (errors.deadletterqueue.topic.name / errors.deadletterqueue.copy.record.enable) + for the given share group (KIP-1191). + """ + if topic_name is None and copy_record_enable is None: + return + if node is None: + node = self.nodes[0] + config_script = self.path.script("kafka-configs.sh", node) + + if command_config is None: + command_config = "" + else: + command_config = "--command-config " + command_config + + configs = [] + if topic_name is not None: + configs.append("errors.deadletterqueue.topic.name=%s" % topic_name) + if copy_record_enable is not None: + configs.append("errors.deadletterqueue.copy.record.enable=%s" % str(copy_record_enable).lower()) + + cmd = fix_opts_for_new_jvm(node) + cmd += "%s --bootstrap-server %s --group %s --alter --add-config \"%s\" %s" % \ + (config_script, + self.bootstrap_servers(self.security_protocol), + group, + ",".join(configs), + command_config) + return "Completed" in self.run_cli_tool(node, cmd) + + def set_share_group_delivery_count_limit(self, group, limit, node=None, command_config=None): + """ Set the share.delivery.count.limit config (GroupConfig, per-group override) for the given share group. + """ + if node is None: + node = self.nodes[0] + config_script = self.path.script("kafka-configs.sh", node) + + if command_config is None: + command_config = "" + else: + command_config = "--command-config " + command_config + + cmd = fix_opts_for_new_jvm(node) + cmd += "%s --bootstrap-server %s --group %s --alter --add-config \"share.delivery.count.limit=%s\" %s" % \ + (config_script, + self.bootstrap_servers(self.security_protocol), + group, + limit, + command_config) + return "Completed" in self.run_cli_tool(node, cmd) + def list_consumer_groups(self, node=None, command_config=None, state=None, type=None): """ Get list of consumer groups. """ diff --git a/tests/kafkatest/services/verifiable_share_consumer.py b/tests/kafkatest/services/verifiable_share_consumer.py index 65f8696623771..f061beeb1fed6 100644 --- a/tests/kafkatest/services/verifiable_share_consumer.py +++ b/tests/kafkatest/services/verifiable_share_consumer.py @@ -36,6 +36,10 @@ def __init__(self, node, idx, state=ShareConsumerState.Dead): self.total_consumed = 0 self.total_acknowledged_successfully = 0 self.total_acknowledged_failed = 0 + self.total_accepted = 0 + self.total_released = 0 + self.total_rejected = 0 + self.total_renewed = 0 self.consumed_per_partition = {} self.acknowledged_per_partition = {} self.acknowledged_per_partition_failed = {} @@ -56,6 +60,12 @@ def handle_offsets_acknowledged(self, event, node, logger): for share_partition_data in event["partitions"]: topic_partition = TopicPartition(share_partition_data["topic"], share_partition_data["partition"]) self.acknowledged_per_partition[topic_partition] = self.acknowledged_per_partition.get(topic_partition, 0) + share_partition_data["count"] + ack_type_counts = event.get("ackTypeCounts") + if ack_type_counts: + self.total_accepted += ack_type_counts.get("ACCEPT", 0) + self.total_released += ack_type_counts.get("RELEASE", 0) + self.total_rejected += ack_type_counts.get("REJECT", 0) + self.total_renewed += ack_type_counts.get("RENEW", 0) logger.debug("Offsets acknowledged for %s" % (node.account.hostname)) else: self.total_acknowledged_failed += event["count"] @@ -107,11 +117,16 @@ class VerifiableShareConsumer(KafkaPathResolverMixin, VerifiableClientMixin, Bac "collect_default": True} } - def __init__(self, context, num_nodes, kafka, topic, group_id, max_messages=-1, - acknowledgement_mode="auto", version=DEV_BRANCH, stop_timeout_sec=60, - log_level="INFO", jaas_override_variables=None, on_record_consumed=None): + def __init__(self, context, num_nodes, kafka, topic, group_id, max_messages=-1, + acknowledgement_mode="auto", version=DEV_BRANCH, stop_timeout_sec=60, + log_level="INFO", jaas_override_variables=None, on_record_consumed=None, + ack_pattern=None): """ :param jaas_override_variables: A dict of variables to be used in the jaas.conf template file + :param ack_pattern: A list of acknowledge types (e.g. ["reject", "release", "accept"]) cycled + through per record via (offset % len(ack_pattern)). Requires acknowledgement_mode to be + "sync" or "async". When None/empty (the default), acknowledgement stays implicit (unchanged + behavior). """ super(VerifiableShareConsumer, self).__init__(context, num_nodes) self.log_level = log_level @@ -120,6 +135,7 @@ def __init__(self, context, num_nodes, kafka, topic, group_id, max_messages=-1, self.group_id = group_id self.max_messages = max_messages self.acknowledgement_mode = acknowledgement_mode + self.ack_pattern = ack_pattern self.prop_file = "" self.stop_timeout_sec = stop_timeout_sec self.on_record_consumed = on_record_consumed @@ -221,6 +237,9 @@ def start_cmd(self, node): cmd += " --acknowledgement-mode %s" % self.acknowledgement_mode + if self.ack_pattern: + cmd += " --ack-pattern %s" % ",".join(self.ack_pattern) + cmd += " --bootstrap-server %s" % self.kafka.bootstrap_servers(self.security_config.security_protocol) cmd += " --group-id %s --topic %s" % (self.group_id, self.topic) @@ -291,6 +310,22 @@ def total_failed_acknowledged(self): with self.lock: return self.total_records_acknowledged_failed + def total_accepted(self): + with self.lock: + return sum(handler.total_accepted for handler in self.event_handlers.values()) + + def total_released(self): + with self.lock: + return sum(handler.total_released for handler in self.event_handlers.values()) + + def total_rejected(self): + with self.lock: + return sum(handler.total_rejected for handler in self.event_handlers.values()) + + def total_renewed(self): + with self.lock: + return sum(handler.total_renewed for handler in self.event_handlers.values()) + def total_consumed_for_a_share_consumer(self, node): with self.lock: return self.event_handlers[node].total_consumed diff --git a/tests/kafkatest/tests/client/share_consumer_dlq_test.py b/tests/kafkatest/tests/client/share_consumer_dlq_test.py new file mode 100644 index 0000000000000..3c9edf6548a2c --- /dev/null +++ b/tests/kafkatest/tests/client/share_consumer_dlq_test.py @@ -0,0 +1,202 @@ +# 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. + +from ducktape.mark import matrix +from ducktape.mark.resource import cluster +from ducktape.utils.util import wait_until + +from kafkatest.services.kafka import KafkaService, quorum +from kafkatest.services.verifiable_consumer import VerifiableConsumer +from kafkatest.tests.verifiable_share_consumer_test import VerifiableShareConsumerTest + + +class ShareConsumerDLQTest(VerifiableShareConsumerTest): + """System tests for share group dead-letter queues (KIP-1191). + + These tests exercise the real distributed cluster/consumer topology that the JUnit + integration test suite (ShareConsumerDLQTest, clients-integration-tests) cannot: multiple + brokers, real consumer processes, and end-to-end DLQ topic content verification. + """ + + TOPIC = {"name": "dlq-source-topic", "partitions": 1, "replication_factor": 1} + + num_consumers = 1 + num_producers = 1 + num_brokers = 3 + + share_group_id = "dlq-test-group" + total_messages = 300 + + default_timeout_sec = 180 + + def __init__(self, test_context): + super(ShareConsumerDLQTest, self).__init__(test_context, num_consumers=self.num_consumers, + num_producers=self.num_producers, num_zk=0, num_brokers=self.num_brokers, + topics={self.TOPIC["name"]: {"partitions": self.TOPIC["partitions"], + "replication-factor": self.TOPIC["replication_factor"]}}) + + # DLQ (KIP-1191) is gated behind share.version=2, which is not yet the production + # default (LATEST_PRODUCTION=SV_1) -- bootstrap the cluster with it explicitly. KafkaTest's + # constructor already built a throwaway self.kafka above; nothing has started yet, but + # ducktape allocates cluster nodes at Service.__init__ time (not at start()), so the + # throwaway service's nodes (and its isolated controller quorum's, if any) must be freed + # before building the replacement, or the cluster runs out of nodes. + self.kafka.free() + if self.kafka.isolated_controller_quorum: + self.kafka.isolated_controller_quorum.free() + self.kafka = KafkaService(test_context, self.num_brokers, self.zk, topics=self.topics, + controller_num_nodes_override=self.num_zk, share_version="2") + + def create_dlq_topic(self, name): + self.kafka.create_topic({ + "topic": name, + "partitions": 1, + "replication-factor": 1, + "configs": {"errors.deadletterqueue.group.enable": "true"} + }) + + def setup_dlq_group_config(self, dlq_topic, copy_record_enable=None): + wait_until(lambda: self.kafka.set_share_group_offset_reset_strategy(group=self.share_group_id, strategy="earliest"), + timeout_sec=20, backoff_sec=2, err_msg="share.auto.offset.reset not set to earliest") + wait_until(lambda: self.kafka.set_share_group_dlq_config(group=self.share_group_id, topic_name=dlq_topic, + copy_record_enable=copy_record_enable), + timeout_sec=20, backoff_sec=2, err_msg="DLQ config not applied to share group") + + def read_dlq_topic(self, dlq_topic, min_messages, timeout_sec=None): + """Consume dlq_topic with a plain (non-share) VerifiableConsumer and return the collected record_data events.""" + timeout_sec = timeout_sec or self.default_timeout_sec + dlq_records = [] + + def on_dlq_record(event, node): + dlq_records.append(event) + + dlq_consumer = VerifiableConsumer(self.test_context, 1, self.kafka, dlq_topic, + group_id="%s-dlq-verifier" % self.share_group_id, + on_record_consumed=on_dlq_record) + dlq_consumer.start() + wait_until(lambda: dlq_consumer.total_consumed() >= min_messages, timeout_sec=timeout_sec, + err_msg="Timed out waiting to consume %d records from DLQ topic %s" % (min_messages, dlq_topic)) + dlq_consumer.stop_all() + return dlq_records + + @cluster(num_nodes=10) + @matrix(metadata_quorum=[quorum.isolated_kraft]) + def test_single_partition_dlq_reject(self, metadata_quorum=quorum.isolated_kraft): + """Every record is REJECTed and must be written to the DLQ topic exactly once, headers-only + (copy-record left at its default of false).""" + dlq_topic = "dlq.reject-single" + self.create_dlq_topic(dlq_topic) + self.setup_dlq_group_config(dlq_topic) + + producer = self.setup_producer(self.TOPIC["name"], max_messages=self.total_messages) + consumer = self.setup_share_group(self.TOPIC["name"], group_id=self.share_group_id, + acknowledgement_mode="sync", ack_pattern=["reject"]) + + producer.start() + self.await_produced_messages(producer, min_messages=self.total_messages, timeout_sec=self.default_timeout_sec) + + consumer.start() + self.await_all_members(consumer, timeout_sec=self.default_timeout_sec) + + wait_until(lambda: consumer.total_rejected() >= self.total_messages, timeout_sec=self.default_timeout_sec, + err_msg="Timed out waiting for all records to be rejected") + + producer.stop() + consumer.stop_all() + + dlq_records = self.read_dlq_topic(dlq_topic, self.total_messages) + assert len(dlq_records) == self.total_messages + assert all(record["value"] is None for record in dlq_records), \ + "Expected DLQ records to carry no value when copy-record is disabled" + + @cluster(num_nodes=10) + @matrix(metadata_quorum=[quorum.isolated_kraft]) + def test_single_partition_dlq_release(self, metadata_quorum=quorum.isolated_kraft): + """Every record is RELEASEd repeatedly; once its delivery count exceeds the (lowered) limit + it must be written to the DLQ topic.""" + dlq_topic = "dlq.release-single" + delivery_count_limit = 2 + self.create_dlq_topic(dlq_topic) + self.setup_dlq_group_config(dlq_topic) + wait_until(lambda: self.kafka.set_share_group_delivery_count_limit(group=self.share_group_id, limit=delivery_count_limit), + timeout_sec=20, backoff_sec=2, err_msg="group.share.delivery.count.limit not set") + + producer = self.setup_producer(self.TOPIC["name"], max_messages=self.total_messages) + consumer = self.setup_share_group(self.TOPIC["name"], group_id=self.share_group_id, + acknowledgement_mode="sync", ack_pattern=["release"]) + + producer.start() + self.await_produced_messages(producer, min_messages=self.total_messages, timeout_sec=self.default_timeout_sec) + + consumer.start() + self.await_all_members(consumer, timeout_sec=self.default_timeout_sec) + + producer.stop() + + # Assert via DLQ topic content, not consumer-side release counts, since RELEASE keeps + # redelivering (and re-releasing) the same offsets until the delivery count is exceeded. + # The consumer must stay running while we wait: it's what drives the redelivery cycles + # that eventually push each record over the delivery count limit and into the DLQ. + dlq_records = self.read_dlq_topic(dlq_topic, self.total_messages, timeout_sec=self.default_timeout_sec * 2) + assert len(dlq_records) == self.total_messages + + consumer.stop_all() + + @cluster(num_nodes=10) + @matrix(metadata_quorum=[quorum.isolated_kraft]) + def test_single_partition_dlq_mixed(self, metadata_quorum=quorum.isolated_kraft): + """Records are cycled reject/release/accept by (offset % 3); reject+release offsets must + eventually land in the DLQ topic with the original value copied (copy-record enabled), + while accept offsets never reach the DLQ.""" + dlq_topic = "dlq.mixed-single" + delivery_count_limit = 2 + self.create_dlq_topic(dlq_topic) + self.setup_dlq_group_config(dlq_topic, copy_record_enable=True) + wait_until(lambda: self.kafka.set_share_group_delivery_count_limit(group=self.share_group_id, limit=delivery_count_limit), + timeout_sec=20, backoff_sec=2, err_msg="group.share.delivery.count.limit not set") + + producer = self.setup_producer(self.TOPIC["name"], max_messages=self.total_messages) + consumer = self.setup_share_group(self.TOPIC["name"], group_id=self.share_group_id, + acknowledgement_mode="sync", ack_pattern=["reject", "release", "accept"]) + + producer.start() + self.await_produced_messages(producer, min_messages=self.total_messages, timeout_sec=self.default_timeout_sec) + + consumer.start() + self.await_all_members(consumer, timeout_sec=self.default_timeout_sec) + + expected_dlq_count = sum(1 for offset in range(self.total_messages) if offset % 3 != 2) + expected_accepted_count = self.total_messages - expected_dlq_count + + wait_until(lambda: consumer.total_accepted() >= expected_accepted_count, timeout_sec=self.default_timeout_sec, + err_msg="Timed out waiting for all accept-pattern records to be accepted") + + producer.stop() + + # The consumer must stay running while we wait: released offsets only reach the DLQ after + # repeated redelivery, which the accepted-count check above does not guarantee has finished. + dlq_records = self.read_dlq_topic(dlq_topic, expected_dlq_count, timeout_sec=self.default_timeout_sec * 2) + assert len(dlq_records) == expected_dlq_count + + # With copy-record enabled, verify DLQ records carry the actual original values, not just + # some non-null value: since offsets map 1:1 to VerifiableProducer's sequential int values on + # this single-partition topic, cross-check against the real values the producer sent rather + # than trusting the DLQ content blindly. + expected_dlq_values = {value for value in producer.acked_values if value % 3 != 2} + actual_dlq_values = {int(record["value"]) for record in dlq_records} + assert actual_dlq_values == expected_dlq_values, \ + "DLQ record values did not match the original produced values (copy-record enabled)" + + consumer.stop_all() diff --git a/tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java b/tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java index 5b897cfb643a6..15867e2adb233 100644 --- a/tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java +++ b/tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java @@ -21,6 +21,7 @@ import org.apache.kafka.clients.admin.AlterConfigOp; import org.apache.kafka.clients.admin.AlterConfigsOptions; import org.apache.kafka.clients.admin.ConfigEntry; +import org.apache.kafka.clients.consumer.AcknowledgeType; import org.apache.kafka.clients.consumer.AcknowledgementCommitCallback; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecord; @@ -88,7 +89,8 @@ * See {@link VerifiableShareConsumer.OffsetResetStrategySet}. *
  • records_consumed: contains a summary of records consumed in a single call to {@link KafkaShareConsumer#poll(Duration)}. * See {@link VerifiableShareConsumer.RecordsConsumed}.
    • - *
  • offsets_acknowledged: contains the result of acknowledging offsets. + *
  • offsets_acknowledged: contains the result of acknowledging offsets. When {@code --ack-pattern} is + * used, also includes a per-{@link AcknowledgeType} count breakdown. * See {@link VerifiableShareConsumer.OffsetsAcknowledged}.
    • *
  • record_data: contains the key, value, and offset of an individual consumed record (only included if verbose * * output is enabled). See {@link VerifiableShareConsumer.RecordData}.
    • @@ -114,6 +116,8 @@ public class VerifiableShareConsumer implements Closeable, AcknowledgementCommit private Integer totalAcknowledged = 0; private final String groupId; private final CountDownLatch shutdownLatch = new CountDownLatch(1); + private final List ackPattern; + private final Map> pendingAckTypes = new HashMap<>(); public static class PartitionData { private final String topic; @@ -261,19 +265,22 @@ public List partitions() { } } - @JsonPropertyOrder({ "timestamp", "name", "count", "partitions", "success", "error" }) + @JsonPropertyOrder({ "timestamp", "name", "count", "partitions", "success", "error", "ackTypeCounts" }) protected static class OffsetsAcknowledged extends ShareConsumerEvent { private final long count; private final List partitions; private final String error; private final boolean success; + private final Map ackTypeCounts; - public OffsetsAcknowledged(long count, List partitions, String error, boolean success) { + public OffsetsAcknowledged(long count, List partitions, String error, boolean success, + Map ackTypeCounts) { this.count = count; this.partitions = partitions; this.error = error; this.success = success; + this.ackTypeCounts = ackTypeCounts; } @Override @@ -302,6 +309,12 @@ public boolean success() { return success; } + @JsonProperty + @JsonInclude(JsonInclude.Include.NON_NULL) + public Map ackTypeCounts() { + return ackTypeCounts; + } + } @JsonPropertyOrder({ "timestamp", "name", "key", "value", "topic", "partition", "offset" }) @@ -353,7 +366,8 @@ public VerifiableShareConsumer(KafkaShareConsumer consumer, AcknowledgementMode acknowledgementMode, String offsetResetStrategy, String groupId, - Boolean verbose) { + Boolean verbose, + List ackPattern) { this.out = out; this.consumer = consumer; this.adminClient = adminClient; @@ -363,6 +377,7 @@ public VerifiableShareConsumer(KafkaShareConsumer consumer, this.verbose = verbose; this.maxMessages = maxMessages; this.groupId = groupId; + this.ackPattern = ackPattern; addKafkaSerializerModule(); } @@ -380,7 +395,8 @@ public void serialize(TopicPartition tp, JsonGenerator gen, SerializerProvider s mapper.registerModule(kafka); } - private void onRecordsReceived(ConsumerRecords records) { + // package-private for testing + void onRecordsReceived(ConsumerRecords records) { List summaries = new ArrayList<>(); for (TopicPartition tp : records.partitions()) { List> partitionRecords = records.records(tp); @@ -392,6 +408,12 @@ private void onRecordsReceived(ConsumerRecords records) { for (ConsumerRecord record : partitionRecords) { partitionOffsets.add(record.offset()); + + if (!ackPattern.isEmpty()) { + AcknowledgeType type = ackPattern.get((int) (record.offset() % ackPattern.size())); + consumer.acknowledge(record, type); + pendingAckTypes.computeIfAbsent(tp, k -> new HashMap<>()).put(record.offset(), type); + } } summaries.add(new RecordSetSummary(tp.topic(), tp.partition(), partitionOffsets)); @@ -410,10 +432,27 @@ private void onRecordsReceived(ConsumerRecords records) { public void onComplete(Map> offsetsMap, Exception exception) { List acknowledgedOffsets = new ArrayList<>(); int totalAcknowledged = 0; + Map ackTypeCounts = ackPattern.isEmpty() ? null : new HashMap<>(); for (Map.Entry> offsetEntry : offsetsMap.entrySet()) { TopicIdPartition tp = offsetEntry.getKey(); acknowledgedOffsets.add(new AcknowledgedData(tp.topic(), tp.partition(), offsetEntry.getValue())); totalAcknowledged += offsetEntry.getValue().size(); + + if (ackTypeCounts != null) { + TopicPartition topicPartition = new TopicPartition(tp.topic(), tp.partition()); + Map offsetTypes = pendingAckTypes.get(topicPartition); + if (offsetTypes != null) { + for (Long offset : offsetEntry.getValue()) { + AcknowledgeType type = offsetTypes.remove(offset); + if (type != null) { + ackTypeCounts.merge(type.name(), 1L, Long::sum); + } + } + if (offsetTypes.isEmpty()) { + pendingAckTypes.remove(topicPartition); + } + } + } } boolean success = true; @@ -422,7 +461,7 @@ public void onComplete(Map> offsetsMap, Exception ex success = false; error = exception.getMessage(); } - printJson(new OffsetsAcknowledged(totalAcknowledged, acknowledgedOffsets, error, success)); + printJson(new OffsetsAcknowledged(totalAcknowledged, acknowledgedOffsets, error, success, ackTypeCounts)); if (success) { this.totalAcknowledged += totalAcknowledged; } @@ -573,6 +612,19 @@ private static ArgumentParser argParser() { .metavar("ACKNOWLEDGEMENT-MODE") .help("Acknowledgement mode for the share consumers (must be either 'auto', 'sync' or 'async')"); + parser.addArgument("--ack-pattern") + .action(store()) + .required(false) + .setDefault("") + .type(String.class) + .dest("ackPattern") + .metavar("ACK-PATTERN") + .help("Comma-separated cycle of accept|release|reject|renew applied to each consumed record " + + "via AcknowledgeType, selected by (record offset % pattern length). When set, the share " + + "consumer is switched to explicit acknowledgement mode and requires --acknowledgement-mode " + + "to be 'sync' or 'async'. When empty (the default), acknowledgement is implicit via commit " + + "(existing behavior, unchanged)."); + parser.addArgument("--offset-reset-strategy") .action(store()) .required(false) @@ -593,6 +645,17 @@ private static ArgumentParser argParser() { return parser; } + // package-private for testing + static List parseAckPattern(String ackPatternArg) { + List ackPattern = new ArrayList<>(); + if (ackPatternArg != null && !ackPatternArg.isEmpty()) { + for (String type : ackPatternArg.split(",")) { + ackPattern.add(AcknowledgeType.valueOf(type.trim().toUpperCase(Locale.ROOT))); + } + } + return ackPattern; + } + public static VerifiableShareConsumer createFromArgs(ArgumentParser parser, String[] args) throws ArgumentParserException { Namespace res = parser.parseArgs(args); @@ -602,6 +665,14 @@ public static VerifiableShareConsumer createFromArgs(ArgumentParser parser, Stri String configFile = res.getString("commandConfig"); String brokerHostAndPort = res.getString("bootstrapServer"); + List ackPattern = parseAckPattern(res.getString("ackPattern")); + if (!ackPattern.isEmpty()) { + if (acknowledgementMode == AcknowledgementMode.AUTO) { + throw new ArgumentParserException( + "--ack-pattern requires --acknowledgement-mode to be 'sync' or 'async', not 'auto'", parser); + } + } + Properties consumerProps = new Properties(); if (configFile != null) { try { @@ -617,6 +688,10 @@ public static VerifiableShareConsumer createFromArgs(ArgumentParser parser, Stri consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerHostAndPort); + if (!ackPattern.isEmpty()) { + consumerProps.put(ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG, "explicit"); + } + String topic = res.getString("topic"); int maxMessages = res.getInt("maxMessages"); boolean verbose = res.getBoolean("verbose"); @@ -645,7 +720,8 @@ public static VerifiableShareConsumer createFromArgs(ArgumentParser parser, Stri acknowledgementMode, offsetResetStrategy, groupId, - verbose); + verbose, + ackPattern); } public static void main(String[] args) { From e5927fbaf11c11017c6b31f544298f6acd3c81fc Mon Sep 17 00:00:00 2001 From: Sanskar Jhajharia Date: Wed, 8 Jul 2026 13:31:13 +0530 Subject: [PATCH 2/5] Add combined_kraft mode --- tests/kafkatest/tests/client/share_consumer_dlq_test.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/tests/kafkatest/tests/client/share_consumer_dlq_test.py b/tests/kafkatest/tests/client/share_consumer_dlq_test.py index 3c9edf6548a2c..76cfe3a45b8d3 100644 --- a/tests/kafkatest/tests/client/share_consumer_dlq_test.py +++ b/tests/kafkatest/tests/client/share_consumer_dlq_test.py @@ -92,7 +92,7 @@ def on_dlq_record(event, node): return dlq_records @cluster(num_nodes=10) - @matrix(metadata_quorum=[quorum.isolated_kraft]) + @matrix(metadata_quorum=[quorum.isolated_kraft, quorum.combined_kraft]) def test_single_partition_dlq_reject(self, metadata_quorum=quorum.isolated_kraft): """Every record is REJECTed and must be written to the DLQ topic exactly once, headers-only (copy-record left at its default of false).""" @@ -122,7 +122,7 @@ def test_single_partition_dlq_reject(self, metadata_quorum=quorum.isolated_kraft "Expected DLQ records to carry no value when copy-record is disabled" @cluster(num_nodes=10) - @matrix(metadata_quorum=[quorum.isolated_kraft]) + @matrix(metadata_quorum=[quorum.isolated_kraft, quorum.combined_kraft]) def test_single_partition_dlq_release(self, metadata_quorum=quorum.isolated_kraft): """Every record is RELEASEd repeatedly; once its delivery count exceeds the (lowered) limit it must be written to the DLQ topic.""" @@ -155,7 +155,7 @@ def test_single_partition_dlq_release(self, metadata_quorum=quorum.isolated_kraf consumer.stop_all() @cluster(num_nodes=10) - @matrix(metadata_quorum=[quorum.isolated_kraft]) + @matrix(metadata_quorum=[quorum.isolated_kraft, quorum.combined_kraft]) def test_single_partition_dlq_mixed(self, metadata_quorum=quorum.isolated_kraft): """Records are cycled reject/release/accept by (offset % 3); reject+release offsets must eventually land in the DLQ topic with the original value copied (copy-record enabled), From be8b9850dd48e0e59cd67f0135bd845f519a258e Mon Sep 17 00:00:00 2001 From: Sanskar Jhajharia Date: Thu, 9 Jul 2026 11:23:01 +0530 Subject: [PATCH 3/5] Address comment --- .../java/org/apache/kafka/tools/VerifiableShareConsumer.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java b/tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java index 15867e2adb233..73bfa191ad672 100644 --- a/tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java +++ b/tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java @@ -432,7 +432,7 @@ void onRecordsReceived(ConsumerRecords records) { public void onComplete(Map> offsetsMap, Exception exception) { List acknowledgedOffsets = new ArrayList<>(); int totalAcknowledged = 0; - Map ackTypeCounts = ackPattern.isEmpty() ? null : new HashMap<>(); + Map ackTypeCounts = (ackPattern.isEmpty() || exception != null) ? null : new HashMap<>(); for (Map.Entry> offsetEntry : offsetsMap.entrySet()) { TopicIdPartition tp = offsetEntry.getKey(); acknowledgedOffsets.add(new AcknowledgedData(tp.topic(), tp.partition(), offsetEntry.getValue())); From 74361bd31ae563962ce7f6277c0845d50879d821 Mon Sep 17 00:00:00 2001 From: Sanskar Jhajharia Date: Thu, 9 Jul 2026 14:35:56 +0530 Subject: [PATCH 4/5] Empty Commit From 43994141e4d2befb4327e653725a53bd2dab2534 Mon Sep 17 00:00:00 2001 From: Sanskar Jhajharia Date: Fri, 10 Jul 2026 09:37:28 +0530 Subject: [PATCH 5/5] Address comment-1 --- .../kafka/tools/VerifiableShareConsumer.java | 43 ++++++++++++++++--- 1 file changed, 37 insertions(+), 6 deletions(-) diff --git a/tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java b/tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java index 73bfa191ad672..3917718c93e67 100644 --- a/tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java +++ b/tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java @@ -118,6 +118,10 @@ public class VerifiableShareConsumer implements Closeable, AcknowledgementCommit private final CountDownLatch shutdownLatch = new CountDownLatch(1); private final List ackPattern; private final Map> pendingAckTypes = new HashMap<>(); + // Tracks, per offset, which ackPattern index to apply next. Only advances past a RENEW (RENEW + // re-delivers the same record), so non-renewed offsets keep resolving to their original + // (offset % ackPattern.size()) entry, e.g. a RELEASE keeps being re-applied on every redelivery. + private final Map> ackPatternIndex = new HashMap<>(); public static class PartitionData { private final String topic; @@ -410,9 +414,18 @@ void onRecordsReceived(ConsumerRecords records) { partitionOffsets.add(record.offset()); if (!ackPattern.isEmpty()) { - AcknowledgeType type = ackPattern.get((int) (record.offset() % ackPattern.size())); + Map partitionIndex = ackPatternIndex.computeIfAbsent(tp, k -> new HashMap<>()); + int index = partitionIndex.computeIfAbsent(record.offset(), + offset -> (int) (offset % ackPattern.size())); + AcknowledgeType type = ackPattern.get(index); consumer.acknowledge(record, type); pendingAckTypes.computeIfAbsent(tp, k -> new HashMap<>()).put(record.offset(), type); + if (type == AcknowledgeType.RENEW) { + // RENEW re-delivers the same record; advance to the next pattern entry so a + // subsequent delivery resolves to a different (eventually terminal) type + // instead of renewing the same offset indefinitely. + partitionIndex.put(record.offset(), (index + 1) % ackPattern.size()); + } } } @@ -442,10 +455,21 @@ public void onComplete(Map> offsetsMap, Exception ex TopicPartition topicPartition = new TopicPartition(tp.topic(), tp.partition()); Map offsetTypes = pendingAckTypes.get(topicPartition); if (offsetTypes != null) { - for (Long offset : offsetEntry.getValue()) { + for (long offset : offsetEntry.getValue()) { AcknowledgeType type = offsetTypes.remove(offset); if (type != null) { ackTypeCounts.merge(type.name(), 1L, Long::sum); + if (type != AcknowledgeType.RENEW) { + // Terminal (or about-to-be-redelivered-unchanged, e.g. RELEASE) outcome: + // no future delivery of this offset needs an advanced pattern index. + Map partitionIndex = ackPatternIndex.get(topicPartition); + if (partitionIndex != null) { + partitionIndex.remove(offset); + if (partitionIndex.isEmpty()) { + ackPatternIndex.remove(topicPartition); + } + } + } } } if (offsetTypes.isEmpty()) { @@ -620,10 +644,12 @@ private static ArgumentParser argParser() { .dest("ackPattern") .metavar("ACK-PATTERN") .help("Comma-separated cycle of accept|release|reject|renew applied to each consumed record " + - "via AcknowledgeType, selected by (record offset % pattern length). When set, the share " + - "consumer is switched to explicit acknowledgement mode and requires --acknowledgement-mode " + - "to be 'sync' or 'async'. When empty (the default), acknowledgement is implicit via commit " + - "(existing behavior, unchanged)."); + "via AcknowledgeType, selected by (record offset % pattern length). RENEW re-delivers the " + + "same record, so its next delivery advances to the following pattern entry instead of " + + "renewing again (a pattern of only 'renew' is rejected, since it could never resolve). " + + "When set, the share consumer is switched to explicit acknowledgement mode and requires " + + "--acknowledgement-mode to be 'sync' or 'async'. When empty (the default), acknowledgement " + + "is implicit via commit (existing behavior, unchanged)."); parser.addArgument("--offset-reset-strategy") .action(store()) @@ -671,6 +697,11 @@ public static VerifiableShareConsumer createFromArgs(ArgumentParser parser, Stri throw new ArgumentParserException( "--ack-pattern requires --acknowledgement-mode to be 'sync' or 'async', not 'auto'", parser); } + if (ackPattern.stream().allMatch(type -> type == AcknowledgeType.RENEW)) { + throw new ArgumentParserException( + "--ack-pattern must contain at least one non-RENEW type; a pattern of only 'renew' " + + "would re-deliver every record indefinitely without ever resolving it", parser); + } } Properties consumerProps = new Properties();