KAFKA-20772 (2/n): Add multi-partition and multi-topic DLQ system tests (#22851)

### Summary
Extends the share-group DLQ system test suite (`ShareConsumerDLQTest`,
added in the prior PR covering single-partition scenarios) to cover two
more topologies from the KAFKA-20772 test plan:
- Multi-partition (single topic, 3 partitions):
`test_multi_partition_dlq_reject` (parametrized over
`num_dlq_partitions=[3, 1]` to cover both 1:1 and modulus-collapsing DLQ
routing), `test_multi_partition_dlq_release`,
`test_multi_partition_dlq_mixed`.
- Multi-topic (two source topics sharing one share group and one DLQ
topic): `test_multi_topic_dlq_reject`, `test_multi_topic_dlq_release`. A
single `VerifiableShareConsumer` member subscribes to both topics at
once via a comma-separated `--topic` list (new capability added here).
DLQ records are asserted to carry the correct `__dlq.errors.topic`
header and route to `source_partition % num_dlq_partitions` per topic.

### Changes
- Multi-topic membership: two independent VerifiableShareConsumer
members join the same share group, each subscribed to just one source
topic, instead of one member subscribed to both.
- Routing/attribution verification: instead of reading per-record
__dlq.errors.* headers, routing correctness is verified from the
observable per-DLQ-partition record counts (via the existing
kafka.get_offset_shell helper), computed from
producer.acked_values_by_partition and the source_partition %
num_dlq_partitions rule.

Reviewers: Sushant Mahajan <smahajan@confluent.io>, Andrew Schofield
 <aschofield@confluent.io>

Author: sjhajharia

tests/kafkatest/tests/client/share_consumer_dlq_test.py

@@ -31,21 +31,28 @@ class ShareConsumerDLQTest(VerifiableShareConsumerTest):
     """
 
     TOPIC = {"name": "dlq-source-topic", "partitions": 1, "replication_factor": 1}
+    TOPIC_MULTI = {"name": "dlq-source-multi", "partitions": 3, "replication_factor": 3}
+    TOPIC_MULTI_A = {"name": "dlq-source-multi-a", "partitions": 3, "replication_factor": 3}
+    TOPIC_MULTI_B = {"name": "dlq-source-multi-b", "partitions": 2, "replication_factor": 3}
 
     num_consumers = 1
     num_producers = 1
     num_brokers = 3
 
     share_group_id = "dlq-test-group"
-    total_messages = 300
+    total_messages = 50
 
     default_timeout_sec = 180
 
     def __init__(self, test_context):
+        topics = {}
+        for topic in (self.TOPIC, self.TOPIC_MULTI, self.TOPIC_MULTI_A, self.TOPIC_MULTI_B):
+            topics[topic["name"]] = {"partitions": topic["partitions"],
+                                      "replication-factor": topic["replication_factor"]}
+
         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"]}})
+            topics=topics)
 
         # 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
@@ -59,14 +66,75 @@ def __init__(self, test_context):
         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):
+    def create_dlq_topic(self, name, partitions=1, replication_factor=1):
         self.kafka.create_topic({
             "topic": name,
-            "partitions": 1,
-            "replication-factor": 1,
+            "partitions": partitions,
+            "replication-factor": replication_factor,
             "configs": {"errors.deadletterqueue.group.enable": "true"}
         })
 
+    def dlq_partition_counts(self, dlq_topic):
+        """Return {partition: record_count} for dlq_topic, from each partition's latest offset
+        (these DLQ topics are freshly created for each test, so offset == record count).
+        """
+        counts = {}
+        for line in self.kafka.get_offset_shell(topic=dlq_topic, time="latest").strip().split("\n"):
+            if not line:
+                continue
+            _, partition, offset = line.rsplit(":", 2)
+            counts[int(partition)] = int(offset)
+        return counts
+
+    def expected_dlq_partition_counts(self, dlq_counts_by_source_partition, num_dlq_partitions):
+        """Compute the expected number of DLQ records per DLQ partition from the number of records
+        actually DLQ'd on each source partition and the source_partition % num_dlq_partitions
+        routing rule.
+        """
+        counts = {}
+        for source_partition, dlq_count in dlq_counts_by_source_partition.items():
+            dlq_partition = source_partition % num_dlq_partitions
+            counts[dlq_partition] = counts.get(dlq_partition, 0) + dlq_count
+        return counts
+
+    def assert_dlq_partition_counts(self, dlq_topic, expected_counts_by_dlq_partition):
+        """Assert the number of records that actually landed in each DLQ partition matches
+        expected_counts_by_dlq_partition, validating source_partition % num_dlq_partitions routing
+        purely from observable per-partition record counts (no per-record header access needed).
+        """
+        actual_counts = self.dlq_partition_counts(dlq_topic)
+        for partition in set(expected_counts_by_dlq_partition) | set(actual_counts):
+            expected = expected_counts_by_dlq_partition.get(partition, 0)
+            actual = actual_counts.get(partition, 0)
+            assert actual == expected, \
+                "DLQ topic %s partition %d has %d records, expected %d" % \
+                (dlq_topic, partition, actual, expected)
+
+    def merge_partition_counts(self, *count_dicts):
+        """Sum multiple {partition: count} dicts (e.g. one per source topic sharing a DLQ topic)
+        into a single combined {partition: count} dict."""
+        merged = {}
+        for counts in count_dicts:
+            for partition, count in counts.items():
+                merged[partition] = merged.get(partition, 0) + count
+        return merged
+
+    def expected_mixed_dlq_outcome(self, producer):
+        """For a producer whose consumer uses the reject/release/accept-cycling ack pattern, compute
+        the expected DLQ'd value set and accepted count 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_values = set()
+        accepted_count = 0
+        for values in producer.acked_values_by_partition.values():
+            for offset, value in enumerate(values):
+                if offset % 3 != 2:
+                    dlq_values.add(value)
+                else:
+                    accepted_count += 1
+        return dlq_values, accepted_count
+
     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")
@@ -200,3 +268,214 @@ 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
+
+        dlq_counts_by_source_partition = {tp.partition: len(values) for tp, values in producer.acked_values_by_partition.items()}
+        expected_counts = self.expected_dlq_partition_counts(dlq_counts_by_source_partition, num_dlq_partitions)
+        self.assert_dlq_partition_counts(dlq_topic, expected_counts)
+
+    @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
+
+        dlq_counts_by_source_partition = {tp.partition: len(values) for tp, values in producer.acked_values_by_partition.items()}
+        expected_counts = self.expected_dlq_partition_counts(dlq_counts_by_source_partition, num_dlq_partitions)
+        self.assert_dlq_partition_counts(dlq_topic, expected_counts)
+
+        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)"
+
+        dlq_counts_by_source_partition = {
+            tp.partition: sum(1 for offset in range(len(values)) if offset % 3 != 2)
+            for tp, values in producer.acked_values_by_partition.items()
+        }
+        expected_counts = self.expected_dlq_partition_counts(dlq_counts_by_source_partition, num_dlq_partitions)
+        self.assert_dlq_partition_counts(dlq_topic, expected_counts)
+
+        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. Two separate VerifiableShareConsumer members join the group, each subscribed to
+        just one of the two topics. 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
+        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_a = self.setup_share_group(self.TOPIC_MULTI_A["name"], group_id=self.share_group_id,
+                                             acknowledgement_mode="sync", ack_pattern=["reject"])
+        consumer_b = self.setup_share_group(self.TOPIC_MULTI_B["name"], 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_a.start()
+        consumer_b.start()
+        self.await_all_members(consumer_a, timeout_sec=self.default_timeout_sec)
+        self.await_all_members(consumer_b, timeout_sec=self.default_timeout_sec)
+
+        wait_until(lambda: consumer_a.total_rejected() >= self.total_messages,
+                   timeout_sec=self.default_timeout_sec, err_msg="Timed out waiting for all records on topic A to be rejected")
+        wait_until(lambda: consumer_b.total_rejected() >= self.total_messages,
+                   timeout_sec=self.default_timeout_sec, err_msg="Timed out waiting for all records on topic B to be rejected")
+
+        producer_a.stop()
+        producer_b.stop()
+        consumer_a.stop_all()
+        consumer_b.stop_all()
+
+        dlq_records = self.read_dlq_topic(dlq_topic, 2 * self.total_messages)
+        assert len(dlq_records) == 2 * self.total_messages
+
+        counts_a = {tp.partition: len(values) for tp, values in producer_a.acked_values_by_partition.items()}
+        counts_b = {tp.partition: len(values) for tp, values in producer_b.acked_values_by_partition.items()}
+        expected_counts = self.merge_partition_counts(
+            self.expected_dlq_partition_counts(counts_a, num_dlq_partitions),
+            self.expected_dlq_partition_counts(counts_b, num_dlq_partitions))
+        self.assert_dlq_partition_counts(dlq_topic, expected_counts)
+
+    @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
+        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_a = self.setup_share_group(self.TOPIC_MULTI_A["name"], group_id=self.share_group_id,
+                                             acknowledgement_mode="sync", ack_pattern=["release"])
+        consumer_b = self.setup_share_group(self.TOPIC_MULTI_B["name"], 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_a.start()
+        consumer_b.start()
+        self.await_all_members(consumer_a, timeout_sec=self.default_timeout_sec)
+        self.await_all_members(consumer_b, timeout_sec=self.default_timeout_sec)
+
+        producer_a.stop()
+        producer_b.stop()
+
+        # Both consumers 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, 2 * self.total_messages, timeout_sec=self.default_timeout_sec * 2)
+        assert len(dlq_records) == 2 * self.total_messages
+
+        counts_a = {tp.partition: len(values) for tp, values in producer_a.acked_values_by_partition.items()}
+        counts_b = {tp.partition: len(values) for tp, values in producer_b.acked_values_by_partition.items()}
+        expected_counts = self.merge_partition_counts(
+            self.expected_dlq_partition_counts(counts_a, num_dlq_partitions),
+            self.expected_dlq_partition_counts(counts_b, num_dlq_partitions))
+        self.assert_dlq_partition_counts(dlq_topic, expected_counts)
+
+        consumer_a.stop_all()
+        consumer_b.stop_all()