Skip to content

Add cluster mirroring secure system test #3

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
fvaleri merged 1 commit into from
May 18, 2026
Merged

Add cluster mirroring secure system test #3

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
102 changes: 70 additions & 32 deletions tests/kafkatest/services/kafka/kafka.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -600,8 +600,7 @@ def close_port(self, listener_name):
self.port_mappings[listener_name].open = False

def start_minikdc_if_necessary(self, add_principals=""):
has_sasl = self.security_config.has_sasl
if has_sasl:
if self.security_config.has_sasl_kerberos:
if self.minikdc is None:
other_service = self.isolated_kafka if self.isolated_kafka else self.controller_quorum if self.quorum_info.using_kraft else None
if not other_service or not other_service.minikdc:
Expand Down Expand Up @@ -947,18 +946,22 @@ def downgrade_metadata_version(self, new_version):
self.run_metadata_features_command("downgrade", new_version)

def run_metadata_features_command(self, op, new_version):
cmd = self.path.script("kafka-features.sh ")
cmd += "--bootstrap-server %s " % self.bootstrap_servers()
cmd += "%s --metadata %s" % (op, new_version)
node = self.nodes[0]
env_prefix, cmd_suffix = self._cmd_security_opts(node)
cmd = "%s%s " % (env_prefix, self.path.script("kafka-features.sh "))
cmd += "--bootstrap-server %s" % self.bootstrap_servers(self.security_protocol)
cmd += "%s %s --metadata %s" % (cmd_suffix, op, new_version)
self.logger.info("Running %s command...\n%s" % (op, cmd))
self.nodes[0].account.ssh(cmd)
node.account.ssh(cmd)

def run_features_command(self, op, feature, new_version):
cmd = self.path.script("kafka-features.sh ")
cmd += "--bootstrap-server %s " % self.bootstrap_servers()
cmd += "%s --feature %s=%s" % (op, feature, new_version)
node = self.nodes[0]
env_prefix, cmd_suffix = self._cmd_security_opts(node)
cmd = "%s%s " % (env_prefix, self.path.script("kafka-features.sh "))
cmd += "--bootstrap-server %s" % self.bootstrap_servers(self.security_protocol)
cmd += "%s %s --feature %s=%s" % (cmd_suffix, op, feature, new_version)
self.logger.info("Running %s command...\n%s" % (op, cmd))
self.nodes[0].account.ssh(cmd)
node.account.ssh(cmd)

def pids(self, node):
"""Return process ids associated with running processes on the given node."""
Expand Down Expand Up @@ -1402,6 +1405,28 @@ def kafka_acls_cmd_with_optional_security_settings(self, node, kafka_security_pr
(optional_jass_krb_system_props_prefix, kafka_acls_script,
bootstrap_server, optional_command_config_suffix)

def _cmd_security_opts(self, node):
"""Returns (env_prefix, cmd_suffix) for wiring security into CLI commands.
When security is enabled, env_prefix sets KAFKA_OPTS with JAAS and KRB5
system properties, and cmd_suffix appends --command-config with the client
security properties (protocol, SSL stores, SASL mechanism).
Uses broker credentials (user "kafka") when client and inter-broker
protocols match, otherwise uses regular client credentials (user "client").
Returns empty strings for PLAINTEXT (no security needed)."""
security_protocol_to_use = self.security_protocol
if security_protocol_to_use == SecurityConfig.PLAINTEXT:
return ("", "")
if security_protocol_to_use == self.interbroker_security_protocol:
jaas_conf_prop = KafkaService.ADMIN_CLIENT_AS_BROKER_JAAS_CONF_PROPERTY
use_inter_broker_mechanism_for_client = True
else:
jaas_conf_prop = KafkaService.JAAS_CONF_PROPERTY
use_inter_broker_mechanism_for_client = False
using_sasl = security_protocol_to_use != "SSL"
env_prefix = "KAFKA_OPTS='-D%s -D%s' " % (jaas_conf_prop, KafkaService.KRB5_CONF) if using_sasl else ""
cmd_suffix = " --command-config <(echo '%s')" % (self.security_config.client_config(use_inter_broker_mechanism_for_client = use_inter_broker_mechanism_for_client))
return (env_prefix, cmd_suffix)

def run_cli_tool(self, node, cmd):
output = ""
self.logger.debug(cmd)
Expand Down Expand Up @@ -1441,7 +1466,6 @@ def parse_describe_topic(self, topic_description):

return {"partitions": partitions}


def _connect_setting_reassign_partitions(self, node):
if self.all_nodes_reassign_partitions_command_supports_bootstrap_server():
return "--bootstrap-server %s " % self.bootstrap_servers(self.security_protocol)
Expand Down Expand Up @@ -2052,24 +2076,30 @@ def create_cluster_mirror(self, node, mirror_name, mirror_config):
node.account.ssh("mkdir -p /mnt/cluster_mirroring", allow_fail=False)
node.account.create_file(mirror_config_file, str(mirror_config))

env_prefix, cmd_suffix = self._cmd_security_opts(node)
cluster_mirror_script = self.path.script("kafka-cluster-mirrors.sh", node)

cmd = fix_opts_for_new_jvm(node)
cmd += "%s --bootstrap-server %s --create --mirror %s --mirror-config %s" % \
(cluster_mirror_script,
cmd += "%s%s --bootstrap-server %s --create --mirror %s --mirror-config %s%s" % \
(env_prefix,
cluster_mirror_script,
self.bootstrap_servers(self.security_protocol),
mirror_name,
mirror_config_file)
mirror_config_file,
cmd_suffix)
return self.run_cli_tool(node, cmd)

def delete_cluster_mirror(self, node, mirror_name):
env_prefix, cmd_suffix = self._cmd_security_opts(node)
cluster_mirror_script = self.path.script("kafka-cluster-mirrors.sh", node)

cmd = fix_opts_for_new_jvm(node)
cmd += "%s --bootstrap-server %s --delete --mirror %s" % \
(cluster_mirror_script,
cmd += "%s%s --bootstrap-server %s --delete --mirror %s%s" % \
(env_prefix,
cluster_mirror_script,
self.bootstrap_servers(self.security_protocol),
mirror_name)
mirror_name,
cmd_suffix)
output = ""
self.logger.debug(cmd)
for line in node.account.ssh_capture(cmd, allow_fail=True):
Expand All @@ -2079,37 +2109,45 @@ def delete_cluster_mirror(self, node, mirror_name):
return output

def list_cluster_mirror(self, node):
env_prefix, cmd_suffix = self._cmd_security_opts(node)
cluster_mirror_script = self.path.script("kafka-cluster-mirrors.sh", node)

cmd = fix_opts_for_new_jvm(node)
cmd += "%s --bootstrap-server %s --list" % \
(cluster_mirror_script,
self.bootstrap_servers(self.security_protocol))
cmd += "%s%s --bootstrap-server %s --list%s" % \
(env_prefix,
cluster_mirror_script,
self.bootstrap_servers(self.security_protocol),
cmd_suffix)
return self.run_cli_tool(node, cmd)

def describe_cluster_mirror(self, node):
env_prefix, cmd_suffix = self._cmd_security_opts(node)
cluster_mirror_script = self.path.script("kafka-cluster-mirrors.sh", node)

cmd = fix_opts_for_new_jvm(node)
cmd += "%s --bootstrap-server %s --describe --json" % \
(cluster_mirror_script,
self.bootstrap_servers(self.security_protocol))
cmd += "%s%s --bootstrap-server %s --describe --json%s" % \
(env_prefix,
cluster_mirror_script,
self.bootstrap_servers(self.security_protocol),
cmd_suffix)
return self.run_cli_tool(node, cmd)


def _cluster_mirror_action(self, node, mirror_name, topics_regex, action, exclude=None):
assert topics_regex is not None and len(topics_regex) > 0
env_prefix, cmd_suffix = self._cmd_security_opts(node)
cluster_mirror_script = self.path.script("kafka-cluster-mirrors.sh", node)

cmd = fix_opts_for_new_jvm(node)
cmd += "%s --bootstrap-server %s --%s --mirror %s --topics %s" % \
(cluster_mirror_script,
cmd += "%s%s --bootstrap-server %s --%s --mirror %s --topics %s" % \
(env_prefix,
cluster_mirror_script,
self.bootstrap_servers(self.security_protocol),
action,
mirror_name,
topics_regex)
if exclude is not None:
cmd += " --exclude %s" % exclude
cmd += cmd_suffix
return self.run_cli_tool(node, cmd)

def start_cluster_mirror_topics(self, node, mirror_name, topics_regex, exclude=None):
Expand All @@ -2125,10 +2163,10 @@ def resume_cluster_mirror_topics(self, node, mirror_name, topics_regex):
return self._cluster_mirror_action(node, mirror_name, topics_regex, 'resume')

def alter_mirror_config(self, node, mirror_name, config):
config_script = self.path.script("kafka-configs.sh", node)
force_use_zk_connection = not self.all_nodes_configs_command_uses_bootstrap_server()
cmd = fix_opts_for_new_jvm(node)
cmd += "%s --bootstrap-server %s --entity-type mirrors --entity-name %s --alter --add-config %s" % \
(config_script, self.bootstrap_servers(self.security_protocol),
cmd += "%s --entity-type mirrors --entity-name %s --alter --add-config %s" % \
(self.kafka_configs_cmd_with_optional_security_settings(node, force_use_zk_connection),
mirror_name, config)
output = ""
self.logger.debug(cmd)
Expand All @@ -2139,10 +2177,10 @@ def alter_mirror_config(self, node, mirror_name, config):
return output

def describe_mirror_config(self, node, mirror_name):
config_script = self.path.script("kafka-configs.sh", node)
force_use_zk_connection = not self.all_nodes_configs_command_uses_bootstrap_server()
cmd = fix_opts_for_new_jvm(node)
cmd += "%s --bootstrap-server %s --entity-type mirrors --entity-name %s --describe" % \
(config_script, self.bootstrap_servers(self.security_protocol),
cmd += "%s --entity-type mirrors --entity-name %s --describe" % \
(self.kafka_configs_cmd_with_optional_security_settings(node, force_use_zk_connection),
mirror_name)
return self.run_cli_tool(node, cmd)

Expand Down
Loading