This is an automated email from the ASF dual-hosted git repository.
AndrewJSchofield pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 931fde96ffd KAFKA-20772 (1/n): Add DLQ system tests for
single-partition share group scenarios (#22786)
931fde96ffd is described below
commit 931fde96ffd449161661c96fbf89c705aa349603
Author: Sanskar Jhajharia <[email protected]>
AuthorDate: Thu Jul 16 19:00:24 2026 +0530
KAFKA-20772 (1/n): Add DLQ system tests for single-partition share group
scenarios (#22786)
This adds the first system-test coverage for share group dead-letter
queues (KIP-1191, part of KAFKA-19469), plus the client-side tooling
changes needed to drive it — there was previously no way for any
ducktape test client to explicitly REJECT or RELEASE a share-consumed
record; VerifiableShareConsumer only ever implicitly accepted.
`VerifiableShareConsumer.java`
- New `--ack-pattern` CLI flag: a comma-separated cycle of
`accept|release|reject|renew`, applied per record via `(offset %
pattern.length)` — keyed off the immutable record offset so the
assignment is stable across redeliveries.
- When set, the consumer is switched to
`share.acknowledgement.mode=explicit` and requires
`--acknowledgement-mode sync|async` (rejects auto at arg-parse time,
since explicit per-record acks still need a commit to ship).
- `offsets_acknowledged` events now optionally include an
`ackTypeCounts` breakdown `({"ACCEPT": n, "RELEASE": n, ...})`, omitted
entirely when `--ack-pattern` isn't used — no change to the existing
JSON schema for any current caller.
`verifiable_share_consumer.py` (ducktape service)
- Plumbs `ack_pattern` through to the CLI, and adds
`total_accepted()/total_released()/total_rejected()/total_renewed()`
accessors backed by the new `ackTypeCounts` data.
`kafka.py`
- `set_share_group_dlq_config(group, topic_name, copy_record_enable)` —
sets the GroupConfig DLQ configs via kafka-configs.sh.
- `set_share_group_delivery_count_limit(group, limit)` — sets
`share.delivery.count.limit` (the per-group override).
- New share_version constructor param to bootstrap a cluster with
`--feature share.version=<n>`. Since DLQ (share.version=2) is above
`ShareVersion.LATEST_PRODUCTION`, this also flips on
`unstable.feature.versions.enable` in the broker properties.
New: `tests/kafkatest/tests/client/share_consumer_dlq_test.py`
Three tests on a real 3-broker isolated-KRaft cluster, verifying actual
DLQ topic content (not just consumer-side counters):
- `test_single_partition_dlq_reject` — every record REJECTed, DLQ'd
exactly once, values null (copy-record disabled, the default).
- `test_single_partition_dlq_release` — every record RELEASEd
repeatedly; once delivery count exceeds a lowered
share.delivery.count.limit, it lands in the DLQ.
- `test_single_partition_dlq_mixed` — records cycled
reject/release/accept by offset % 3, copy-record enabled; asserts the
DLQ'd record set and values exactly match the subset of produced values
that should have been rejected/released (cross-checked against
VerifiableProducer.acked_values, not just "value is non-null").
All three verified passing end-to-end against a live ducker cluster
(tests/docker/run_tests.sh).
```
> TC_PATHS="tests/kafkatest/tests/client/share_consumer_dlq_test.py"
bash tests/docker/run_tests.sh
================================================================================
SESSION REPORT (ALL TESTS)
ducktape version: 0.14.0
session_id: 2026-07-08--005
run time: 5 minutes 27.339 seconds
tests run: 6
passed: 6
flaky: 0
failed: 0
ignored: 0
================================================================================
test_id:
kafkatest.tests.client.share_consumer_dlq_test.ShareConsumerDLQTest.test_single_partition_dlq_mixed.metadata_quorum=COMBINED_KRAFT
status: PASS
run time: 45.786 seconds
--------------------------------------------------------------------------------
test_id:
kafkatest.tests.client.share_consumer_dlq_test.ShareConsumerDLQTest.test_single_partition_dlq_mixed.metadata_quorum=ISOLATED_KRAFT
status: PASS
run time: 1 minute 5.269 seconds
--------------------------------------------------------------------------------
test_id:
kafkatest.tests.client.share_consumer_dlq_test.ShareConsumerDLQTest.test_single_partition_dlq_reject.metadata_quorum=COMBINED_KRAFT
status: PASS
run time: 43.421 seconds
--------------------------------------------------------------------------------
test_id:
kafkatest.tests.client.share_consumer_dlq_test.ShareConsumerDLQTest.test_single_partition_dlq_reject.metadata_quorum=ISOLATED_KRAFT
status: PASS
run time: 1 minute 2.872 seconds
--------------------------------------------------------------------------------
test_id:
kafkatest.tests.client.share_consumer_dlq_test.ShareConsumerDLQTest.test_single_partition_dlq_release.metadata_quorum=COMBINED_KRAFT
status: PASS
run time: 43.429 seconds
--------------------------------------------------------------------------------
test_id:
kafkatest.tests.client.share_consumer_dlq_test.ShareConsumerDLQTest.test_single_partition_dlq_release.metadata_quorum=ISOLATED_KRAFT
status: PASS
run time: 1 minute 5.794 seconds
--------------------------------------------------------------------------------
```
Reviewers: Sushant Mahajan <[email protected]>, Chia-Ping Tsai
<[email protected]>, Andrew Schofield <[email protected]>
---
tests/kafkatest/services/kafka/kafka.py | 65 ++++++-
.../services/verifiable_share_consumer.py | 41 ++++-
.../tests/client/share_consumer_dlq_test.py | 202 +++++++++++++++++++++
.../kafka/tools/VerifiableShareConsumer.java | 120 +++++++++++-
4 files changed, 416 insertions(+), 12 deletions(-)
diff --git a/tests/kafkatest/services/kafka/kafka.py
b/tests/kafkatest/services/kafka/kafka.py
index 8d98b96e283..82193813e96 100644
--- a/tests/kafkatest/services/kafka/kafka.py
+++ b/tests/kafkatest/services/kafka/kafka.py
@@ -207,7 +207,8 @@ class KafkaService(KafkaPathResolverMixin, JmxMixin,
Service):
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 @@ class KafkaService(KafkaPathResolverMixin, JmxMixin,
Service):
: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=<value> (KIP-932/KIP-1191).
"""
self.zk = zk
@@ -288,6 +290,7 @@ class KafkaService(KafkaPathResolverMixin, JmxMixin,
Service):
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 @@ class KafkaService(KafkaPathResolverMixin, JmxMixin,
Service):
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
@@ -907,6 +911,12 @@ class KafkaService(KafkaPathResolverMixin, JmxMixin,
Service):
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 +1758,57 @@ class KafkaService(KafkaPathResolverMixin, JmxMixin,
Service):
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 65f86966237..f061beeb1fe 100644
--- a/tests/kafkatest/services/verifiable_share_consumer.py
+++ b/tests/kafkatest/services/verifiable_share_consumer.py
@@ -36,6 +36,10 @@ class ShareConsumerEventHandler(object):
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 @@ class ShareConsumerEventHandler(object):
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 @@ class VerifiableShareConsumer(KafkaPathResolverMixin,
VerifiableClientMixin, Bac
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 @@ class VerifiableShareConsumer(KafkaPathResolverMixin,
VerifiableClientMixin, Bac
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 @@ class VerifiableShareConsumer(KafkaPathResolverMixin,
VerifiableClientMixin, Bac
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 00000000000..76cfe3a45b8
--- /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, 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)."""
+ 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, 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."""
+ 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, 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),
+ 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 5b897cfb643..da176c3a689 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.Admin;
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 @@ import static
net.sourceforge.argparse4j.impl.Arguments.storeTrue;
* See {@link VerifiableShareConsumer.OffsetResetStrategySet}.</li>
* <li>records_consumed: contains a summary of records consumed in a single
call to {@link KafkaShareConsumer#poll(Duration)}.
* See {@link VerifiableShareConsumer.RecordsConsumed}.</li>
- * <li>offsets_acknowledged: contains the result of acknowledging offsets.
+ * <li>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}.</li>
* <li>record_data: contains the key, value, and offset of an individual
consumed record (only included if verbose
* * output is enabled). See {@link
VerifiableShareConsumer.RecordData}.</li>
@@ -114,6 +116,12 @@ public class VerifiableShareConsumer implements Closeable,
AcknowledgementCommit
private Integer totalAcknowledged = 0;
private final String groupId;
private final CountDownLatch shutdownLatch = new CountDownLatch(1);
+ private final List<AcknowledgeType> ackPattern;
+ private final Map<TopicPartition, Map<Long, AcknowledgeType>>
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<TopicPartition, Map<Long, Integer>> ackPatternIndex =
new HashMap<>();
public static class PartitionData {
private final String topic;
@@ -261,19 +269,22 @@ public class VerifiableShareConsumer implements
Closeable, AcknowledgementCommit
}
}
- @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<AcknowledgedData> partitions;
private final String error;
private final boolean success;
+ private final Map<String, Long> ackTypeCounts;
- public OffsetsAcknowledged(long count, List<AcknowledgedData>
partitions, String error, boolean success) {
+ public OffsetsAcknowledged(long count, List<AcknowledgedData>
partitions, String error, boolean success,
+ Map<String, Long> ackTypeCounts) {
this.count = count;
this.partitions = partitions;
this.error = error;
this.success = success;
+ this.ackTypeCounts = ackTypeCounts;
}
@Override
@@ -302,6 +313,12 @@ public class VerifiableShareConsumer implements Closeable,
AcknowledgementCommit
return success;
}
+ @JsonProperty
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ public Map<String, Long> ackTypeCounts() {
+ return ackTypeCounts;
+ }
+
}
@JsonPropertyOrder({ "timestamp", "name", "key", "value", "topic",
"partition", "offset" })
@@ -353,7 +370,8 @@ public class VerifiableShareConsumer implements Closeable,
AcknowledgementCommit
AcknowledgementMode acknowledgementMode,
String offsetResetStrategy,
String groupId,
- Boolean verbose) {
+ Boolean verbose,
+ List<AcknowledgeType> ackPattern) {
this.out = out;
this.consumer = consumer;
this.adminClient = adminClient;
@@ -363,6 +381,7 @@ public class VerifiableShareConsumer implements Closeable,
AcknowledgementCommit
this.verbose = verbose;
this.maxMessages = maxMessages;
this.groupId = groupId;
+ this.ackPattern = ackPattern;
addKafkaSerializerModule();
}
@@ -380,7 +399,8 @@ public class VerifiableShareConsumer implements Closeable,
AcknowledgementCommit
mapper.registerModule(kafka);
}
- private void onRecordsReceived(ConsumerRecords<String, String> records) {
+ // package-private for testing
+ void onRecordsReceived(ConsumerRecords<String, String> records) {
List<RecordSetSummary> summaries = new ArrayList<>();
for (TopicPartition tp : records.partitions()) {
List<ConsumerRecord<String, String>> partitionRecords =
records.records(tp);
@@ -392,6 +412,21 @@ public class VerifiableShareConsumer implements Closeable,
AcknowledgementCommit
for (ConsumerRecord<String, String> record : partitionRecords) {
partitionOffsets.add(record.offset());
+
+ if (!ackPattern.isEmpty()) {
+ Map<Long, Integer> 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());
+ }
+ }
}
summaries.add(new RecordSetSummary(tp.topic(), tp.partition(),
partitionOffsets));
@@ -410,10 +445,38 @@ public class VerifiableShareConsumer implements
Closeable, AcknowledgementCommit
public void onComplete(Map<TopicIdPartition, Set<Long>> offsetsMap,
Exception exception) {
List<AcknowledgedData> acknowledgedOffsets = new ArrayList<>();
int totalAcknowledged = 0;
+ Map<String, Long> ackTypeCounts = (ackPattern.isEmpty() || exception
!= null) ? null : new HashMap<>();
for (Map.Entry<TopicIdPartition, Set<Long>> 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<Long, AcknowledgeType> 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 (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<Long, Integer> partitionIndex =
ackPatternIndex.get(topicPartition);
+ if (partitionIndex != null) {
+ partitionIndex.remove(offset);
+ if (partitionIndex.isEmpty()) {
+ ackPatternIndex.remove(topicPartition);
+ }
+ }
+ }
+ }
+ }
+ if (offsetTypes.isEmpty()) {
+ pendingAckTypes.remove(topicPartition);
+ }
+ }
+ }
}
boolean success = true;
@@ -422,7 +485,7 @@ public class VerifiableShareConsumer implements Closeable,
AcknowledgementCommit
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 +636,20 @@ public class VerifiableShareConsumer implements Closeable,
AcknowledgementCommit
.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). 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 uses explicit acknowledgement
and requires " +
+ "--acknowledgement-mode to be 'sync' or 'async'.");
+
parser.addArgument("--offset-reset-strategy")
.action(store())
.required(false)
@@ -593,6 +670,17 @@ public class VerifiableShareConsumer implements Closeable,
AcknowledgementCommit
return parser;
}
+ // package-private for testing
+ static List<AcknowledgeType> parseAckPattern(String ackPatternArg) {
+ List<AcknowledgeType> 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 +690,19 @@ public class VerifiableShareConsumer implements Closeable,
AcknowledgementCommit
String configFile = res.getString("commandConfig");
String brokerHostAndPort = res.getString("bootstrapServer");
+ List<AcknowledgeType> 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);
+ }
+ 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();
if (configFile != null) {
try {
@@ -617,6 +718,10 @@ public class VerifiableShareConsumer implements Closeable,
AcknowledgementCommit
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 +750,8 @@ public class VerifiableShareConsumer implements Closeable,
AcknowledgementCommit
acknowledgementMode,
offsetResetStrategy,
groupId,
- verbose);
+ verbose,
+ ackPattern);
}
public static void main(String[] args) {
