Copilot commented on code in PR #22851:
URL: https://github.com/apache/kafka/pull/22851#discussion_r3596504885
##########
tools/src/main/java/org/apache/kafka/tools/VerifiableConsumer.java:
##########
@@ -418,6 +421,15 @@ public long offset() {
return record.offset();
}
+ @JsonProperty
+ public Map<String, String> headers() {
+ Map<String, String> result = new LinkedHashMap<>();
+ for (Header header : record.headers()) {
+ result.put(header.key(), header.value() == null ? null : new
String(header.value(), StandardCharsets.UTF_8));
+ }
Review Comment:
`headers()` exposes Kafka headers as a `Map<String, String>`, but Kafka
headers allow repeated keys and arbitrary binary values. Using a `Map` silently
drops duplicates (last-write-wins), and forcing UTF-8 decoding can corrupt
non-text header values. Consider serializing headers as an ordered list of
key/value entries (and/or base64-encoding values) to preserve semantics for
downstream consumers.
##########
tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java:
##########
@@ -513,7 +513,11 @@ public void run() {
printJson(new
OffsetResetStrategySet(offsetResetStrategy.type().toString()));
}
- consumer.subscribe(Set.of(this.topic));
+ // --topic accepts a comma-separated list so a single share
consumer can subscribe to
+ // multiple topics with one process, keeping the group's
subscription homogeneous
+ // across members (every member subscribed to the same topic set)
instead of splitting
+ // topics across separate consumer processes/members with
different subscriptions.
+ consumer.subscribe(Set.of(this.topic.split(",")));
consumer.setAcknowledgementCommitCallback(this);
Review Comment:
`--topic` is now documented as a comma-separated list, but
`this.topic.split(",")` is used directly. This will include leading/trailing
whitespace and can produce empty topic names (e.g., trailing comma), and
`Set.of(...)` will also throw if a topic is repeated. It’d be more robust to
trim/filter and validate the parsed topic set before subscribing.
##########
tests/kafkatest/tests/client/share_consumer_dlq_test.py:
##########
@@ -200,3 +246,197 @@ def test_single_partition_dlq_mixed(self,
metadata_quorum=quorum.isolated_kraft)
"DLQ record values did not match the original produced values
(copy-record enabled)"
consumer.stop_all()
+
+ @cluster(num_nodes=10)
+ @matrix(metadata_quorum=[quorum.isolated_kraft, quorum.combined_kraft],
num_dlq_partitions=[3, 1])
+ def test_multi_partition_dlq_reject(self,
metadata_quorum=quorum.isolated_kraft, num_dlq_partitions=3):
+ """Every record on a 3-partition source topic is REJECTed. Verifies
both DLQ topic content
+ and per-record routing (source_partition % num_dlq_partitions) --
num_dlq_partitions=3 covers
+ 1:1 routing, num_dlq_partitions=1 covers the modulus-collapsing
case."""
+ dlq_topic = "dlq.reject-multi-partition-%d" % num_dlq_partitions
+ self.create_dlq_topic(dlq_topic, partitions=num_dlq_partitions,
replication_factor=3)
+ self.setup_dlq_group_config(dlq_topic)
+
+ producer = self.setup_producer(self.TOPIC_MULTI["name"],
max_messages=self.total_messages)
+ consumer = self.setup_share_group(self.TOPIC_MULTI["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
+ self.assert_dlq_routing(dlq_records, num_dlq_partitions,
source_topic=self.TOPIC_MULTI["name"])
+
+ @cluster(num_nodes=10)
+ @matrix(metadata_quorum=[quorum.isolated_kraft, quorum.combined_kraft])
+ def test_multi_partition_dlq_release(self,
metadata_quorum=quorum.isolated_kraft):
+ """Same as test_single_partition_dlq_release but on a 3-partition
source topic."""
+ dlq_topic = "dlq.release-multi-partition"
+ num_dlq_partitions = 3
+ delivery_count_limit = 2
+ self.create_dlq_topic(dlq_topic, partitions=num_dlq_partitions,
replication_factor=3)
+ 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="share.delivery.count.limit not set")
+
+ producer = self.setup_producer(self.TOPIC_MULTI["name"],
max_messages=self.total_messages)
+ consumer = self.setup_share_group(self.TOPIC_MULTI["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()
+
+ 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
+ self.assert_dlq_routing(dlq_records, num_dlq_partitions,
source_topic=self.TOPIC_MULTI["name"])
+
+ consumer.stop_all()
+
+ @cluster(num_nodes=10)
+ @matrix(metadata_quorum=[quorum.isolated_kraft, quorum.combined_kraft])
+ def test_multi_partition_dlq_mixed(self,
metadata_quorum=quorum.isolated_kraft):
+ """Same as test_single_partition_dlq_mixed but on a 3-partition source
topic. The expected
+ DLQ set/count is computed per-partition from the producer's actual
per-partition ack order
+ (offset % 3 is evaluated per-partition, not globally, since each
partition has its own
+ independent offset sequence starting at 0)."""
+ dlq_topic = "dlq.mixed-multi-partition"
+ num_dlq_partitions = 3
+ delivery_count_limit = 2
+ self.create_dlq_topic(dlq_topic, partitions=num_dlq_partitions,
replication_factor=3)
+ 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="share.delivery.count.limit not set")
+
+ producer = self.setup_producer(self.TOPIC_MULTI["name"],
max_messages=self.total_messages)
+ consumer = self.setup_share_group(self.TOPIC_MULTI["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_values, expected_accepted_count =
self.expected_mixed_dlq_outcome(producer)
+
+ 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()
+
+ dlq_records = self.read_dlq_topic(dlq_topic, len(expected_dlq_values),
timeout_sec=self.default_timeout_sec * 2)
+ assert len(dlq_records) == len(expected_dlq_values)
+
+ 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)"
+ self.assert_dlq_routing(dlq_records, num_dlq_partitions,
source_topic=self.TOPIC_MULTI["name"])
+
+ consumer.stop_all()
+
+ @cluster(num_nodes=11)
+ @matrix(metadata_quorum=[quorum.isolated_kraft, quorum.combined_kraft])
+ def test_multi_topic_dlq_reject(self,
metadata_quorum=quorum.isolated_kraft):
+ """Two source topics share one share group and one DLQ topic; every
record on both is
+ REJECTed. A single VerifiableShareConsumer member subscribes to BOTH
topics at once
+ (comma-separated --topic). Verifies DLQ content is correctly
attributed to its source
+ topic via the __dlq.errors.topic header."""
+ dlq_topic = "dlq.reject-multi-topic"
+ num_dlq_partitions = 3
+ both_topics = "%s,%s" % (self.TOPIC_MULTI_A["name"],
self.TOPIC_MULTI_B["name"])
+ self.create_dlq_topic(dlq_topic, partitions=num_dlq_partitions,
replication_factor=3)
+ self.setup_dlq_group_config(dlq_topic)
+
+ producer_a = self.setup_producer(self.TOPIC_MULTI_A["name"],
max_messages=self.total_messages)
+ producer_b = self.setup_producer(self.TOPIC_MULTI_B["name"],
max_messages=self.total_messages)
+ consumer = self.setup_share_group(both_topics,
group_id=self.share_group_id,
+ acknowledgement_mode="sync",
ack_pattern=["reject"])
+
+ producer_a.start()
+ producer_b.start()
+ self.await_produced_messages(producer_a,
min_messages=self.total_messages, timeout_sec=self.default_timeout_sec)
+ self.await_produced_messages(producer_b,
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() >= 2 *
self.total_messages,
+ timeout_sec=self.default_timeout_sec, err_msg="Timed out
waiting for all records to be rejected")
+
+ producer_a.stop()
+ producer_b.stop()
+ consumer.stop_all()
+
+ dlq_records = self.read_dlq_topic(dlq_topic, 2 * self.total_messages)
+ assert len(dlq_records) == 2 * self.total_messages
+
+ records_by_topic = self.group_dlq_records_by_topic(dlq_records)
+ assert set(records_by_topic.keys()) == {self.TOPIC_MULTI_A["name"],
self.TOPIC_MULTI_B["name"]}
+ assert len(records_by_topic[self.TOPIC_MULTI_A["name"]]) ==
self.total_messages
+ assert len(records_by_topic[self.TOPIC_MULTI_B["name"]]) ==
self.total_messages
+
+ self.assert_dlq_routing(records_by_topic[self.TOPIC_MULTI_A["name"]],
num_dlq_partitions,
+ source_topic=self.TOPIC_MULTI_A["name"])
+ self.assert_dlq_routing(records_by_topic[self.TOPIC_MULTI_B["name"]],
num_dlq_partitions,
+ source_topic=self.TOPIC_MULTI_B["name"])
+
+ @cluster(num_nodes=11)
+ @matrix(metadata_quorum=[quorum.isolated_kraft, quorum.combined_kraft])
+ def test_multi_topic_dlq_release(self,
metadata_quorum=quorum.isolated_kraft):
+ """Same as test_multi_topic_dlq_reject but with every record RELEASEd
repeatedly on both
+ topics until it exceeds the (lowered) delivery count limit."""
+ dlq_topic = "dlq.release-multi-topic"
+ num_dlq_partitions = 3
+ delivery_count_limit = 2
+ both_topics = "%s,%s" % (self.TOPIC_MULTI_A["name"],
self.TOPIC_MULTI_B["name"])
+ self.create_dlq_topic(dlq_topic, partitions=num_dlq_partitions,
replication_factor=3)
+ 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="share.delivery.count.limit not set")
+
+ producer_a = self.setup_producer(self.TOPIC_MULTI_A["name"],
max_messages=self.total_messages)
+ producer_b = self.setup_producer(self.TOPIC_MULTI_B["name"],
max_messages=self.total_messages)
+ consumer = self.setup_share_group(both_topics,
group_id=self.share_group_id,
+ acknowledgement_mode="sync",
ack_pattern=["release"])
+
+ producer_a.start()
+ producer_b.start()
+ self.await_produced_messages(producer_a,
min_messages=self.total_messages, timeout_sec=self.default_timeout_sec)
+ self.await_produced_messages(producer_b,
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_a.stop()
+ producer_b.stop()
+
+ dlq_records = self.read_dlq_topic(dlq_topic, 2 * self.total_messages,
timeout_sec=self.default_timeout_sec * 2)
+ assert len(dlq_records) == 2 * self.total_messages
+
+ records_by_topic = self.group_dlq_records_by_topic(dlq_records)
+ assert len(records_by_topic.get(self.TOPIC_MULTI_A["name"], [])) ==
self.total_messages
+ assert len(records_by_topic.get(self.TOPIC_MULTI_B["name"], [])) ==
self.total_messages
+
+ self.assert_dlq_routing(records_by_topic[self.TOPIC_MULTI_A["name"]],
num_dlq_partitions,
+ source_topic=self.TOPIC_MULTI_A["name"])
+ self.assert_dlq_routing(records_by_topic[self.TOPIC_MULTI_B["name"]],
num_dlq_partitions,
+ source_topic=self.TOPIC_MULTI_B["name"])
+
+ consumer.stop_all()
+
+ consumer.stop_all()
Review Comment:
`consumer.stop_all()` is called twice back-to-back, which is redundant and
makes the test harder to read/debug.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
