run kafka bin/ scripts for test fixtures

Author: dpkp

test/fixtures.py

@@ -116,6 +116,12 @@ def test_resource(cls, filename):
             return path
         return os.path.join(cls.project_root, "servers", "resources", "default", filename)
 
+    @classmethod
+    def run_script(cls, script, *args):
+        result = [os.path.join(cls.kafka_root, 'bin', script)]
+        result.extend([str(arg) for arg in args])
+        return result
+
     @classmethod
     def kafka_run_class_args(cls, *args):
         result = [os.path.join(cls.kafka_root, 'bin', 'kafka-run-class.sh')]
@@ -202,6 +208,7 @@ def open(self):
         # Configure Zookeeper child process
         template = self.test_resource("zookeeper.properties")
         properties = self.tmp_dir.join("zookeeper.properties")
+        # Consider replacing w/ run_script('zookeper-server-start.sh', ...)
         args = self.kafka_run_class_args("org.apache.zookeeper.server.quorum.QuorumPeerMain",
                                          properties.strpath)
         env = self.kafka_run_class_env()
@@ -348,8 +355,7 @@ def _jaas_config(self):
 
     def _add_scram_user(self):
         self.out("Adding SCRAM credentials for user {} to zookeeper.".format(self.broker_user))
-        args = self.kafka_run_class_args(
-            "kafka.admin.ConfigCommand",
+        args = self.run_script('kafka-configs.sh',
             "--zookeeper",
             "%s:%d/%s" % (self.zookeeper.host,
                        self.zookeeper.port,
@@ -390,8 +396,7 @@ def out(self, message):
 
     def _create_zk_chroot(self):
         self.out("Creating Zookeeper chroot node...")
-        args = self.kafka_run_class_args("org.apache.zookeeper.ZooKeeperMain",
-                                         "-server",
+        args = self.run_script('zookeeper-shell.sh',
                                          "%s:%d" % (self.zookeeper.host,
                                                     self.zookeeper.port),
                                          "create",
@@ -416,6 +421,7 @@ def start(self):
         properties_template = self.test_resource("kafka.properties")
         jaas_conf_template = self.test_resource("kafka_server_jaas.conf")
 
+        # Consider replacing w/ run_script('kafka-server-start.sh', ...)
         args = self.kafka_run_class_args("kafka.Kafka", properties.strpath)
         env = self.kafka_run_class_env()
         if self.sasl_enabled:
@@ -590,7 +596,7 @@ def _create_topic_via_admin_api(self, topic_name, num_partitions, replication_fa
                 raise errors.for_code(error_code)
 
     def _create_topic_via_cli(self, topic_name, num_partitions, replication_factor):
-        args = self.kafka_run_class_args('kafka.admin.TopicCommand',
+        args = self.run_script('kafka-topics.sh',
                                          '--zookeeper', '%s:%s/%s' % (self.zookeeper.host,
                                                                       self.zookeeper.port,
                                                                       self.zk_chroot),
@@ -614,12 +620,12 @@ def _create_topic_via_cli(self, topic_name, num_partitions, replication_factor):
                 raise RuntimeError("Failed to create topic %s" % (topic_name,))
 
     def get_topic_names(self):
-        args = self.kafka_run_class_args('kafka.admin.TopicCommand',
+        args = self.run_script('kafka-topics.sh',
                                          '--zookeeper', '%s:%s/%s' % (self.zookeeper.host,
                                                                       self.zookeeper.port,
                                                                       self.zk_chroot),
                                          '--list'
-                                         )
+                              )
         env = self.kafka_run_class_env()
         env.pop('KAFKA_LOG4J_OPTS')
         proc = subprocess.Popen(args, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE)