KAFKA-10138: Prefer --bootstrap-server for reassign_partitions command in ducktape tests (#8898)

Reviewers: Colin P. McCabe <cmccabe@apache.org>

tests/kafkatest/services/kafka/kafka.py

@@ -593,6 +593,13 @@ class KafkaService(KafkaPathResolverMixin, JmxMixin, Service):
                  "replicas": map(int, fields[3].split(','))})
         return {"partitions": partitions}
 
+
+    def _connect_setting_reassign_partitions(self, node):
+        if node.version.reassign_partitions_command_supports_bootstrap_server():
+            return "--bootstrap-server %s " % self.bootstrap_servers(self.security_protocol)
+        else:
+            return "--zookeeper %s " % self.zk_connect_setting()
+
     def verify_reassign_partitions(self, reassignment, node=None):
         """Run the reassign partitions admin tool in "verify" mode
         """
@@ -609,7 +616,7 @@ class KafkaService(KafkaPathResolverMixin, JmxMixin, Service):
         cmd = fix_opts_for_new_jvm(node)
         cmd += "echo %s > %s && " % (json_str, json_file)
         cmd += "%s " % self.path.script("kafka-reassign-partitions.sh", node)
-        cmd += "--zookeeper %s " % self.zk_connect_setting()
+        cmd += self._connect_setting_reassign_partitions(node)
         cmd += "--reassignment-json-file %s " % json_file
         cmd += "--verify "
         cmd += "&& sleep 1 && rm -f %s" % json_file
@@ -649,7 +656,7 @@ class KafkaService(KafkaPathResolverMixin, JmxMixin, Service):
         cmd = fix_opts_for_new_jvm(node)
         cmd += "echo %s > %s && " % (json_str, json_file)
         cmd += "%s " % self.path.script( "kafka-reassign-partitions.sh", node)
-        cmd += "--zookeeper %s " % self.zk_connect_setting()
+        cmd += self._connect_setting_reassign_partitions(node)
         cmd += "--reassignment-json-file %s " % json_file
         cmd += "--execute"
         if throttle is not None:

tests/kafkatest/version.py

@@ -59,6 +59,9 @@ class KafkaVersion(LooseVersion):
         # indicate if KIP-515 is available
         return self > LATEST_2_4
 
+    def reassign_partitions_command_supports_bootstrap_server(self):
+        return self >= V_2_5_0
+
 def get_version(node=None):
     """Return the version attached to the given node.
     Default to DEV_BRANCH if node or node.version is undefined (aka None)
@@ -147,3 +150,7 @@ LATEST_2_4 = V_2_4_1
 # 2.5.x versions
 V_2_5_0 = KafkaVersion("2.5.0")
 LATEST_2_5 = V_2_5_0
+
+# 2.6.x versions
+V_2_6_0 = KafkaVersion("2.6.0")
+LATEST_2_6 = V_2_6_0