KAFKA-20772 (1/n): Add DLQ system tests for single-partition share group scenarios

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=<value> (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.
         """

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