kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ewe...@apache.org
Subject kafka git commit: MINOR: bug fixes to ducktape services
Date Fri, 01 Jul 2016 04:16:18 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk 44ad7b574 -> 7edaa3dd8


MINOR: bug fixes to ducktape services

Here's a (mostly successful) run with these changes:

http://testing.confluent.io/confluent-kafka-branch-builder-system-test-results/?prefix=2016-06-27--001.1467080884--alexlod--ducktape-fixes--ad85493/

At least one of the failed tests is failing in trunk, too:

http://testing.confluent.io/confluent-kafka-branch-builder-system-test-results/?prefix=2016-06-28--001.1467090978--alexlod--ducktape-fixes--ad85493/

The contribution is my original work and I license the work to the project under the project's
open source license.

Author: Alex Loddengaard <alexloddengaard@gmail.com>

Reviewers: Geoff Anderson <geoff@confluent.io>, Ismael Juma <ismael@juma.me.uk>,
Ewen Cheslack-Postava <ewen@confluent.io>

Closes #1566 from alexlod/ducktape-fixes


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/7edaa3dd
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/7edaa3dd
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/7edaa3dd

Branch: refs/heads/trunk
Commit: 7edaa3dd89d1a7300a12e101e23f66459a9cc33d
Parents: 44ad7b5
Author: Alex Loddengaard <alexloddengaard@gmail.com>
Authored: Thu Jun 30 21:16:13 2016 -0700
Committer: Ewen Cheslack-Postava <me@ewencp.org>
Committed: Thu Jun 30 21:16:13 2016 -0700

----------------------------------------------------------------------
 tests/kafkatest/services/kafka/kafka.py         | 13 ++++---
 .../services/kafka/templates/kafka.properties   |  4 +-
 tests/kafkatest/services/security/minikdc.py    |  2 +-
 .../services/security/security_config.py        | 40 +++++++++++---------
 tests/kafkatest/services/zookeeper.py           |  3 +-
 .../core/zookeeper_security_upgrade_test.py     |  2 +-
 6 files changed, 36 insertions(+), 28 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/7edaa3dd/tests/kafkatest/services/kafka/kafka.py
----------------------------------------------------------------------
diff --git a/tests/kafkatest/services/kafka/kafka.py b/tests/kafkatest/services/kafka/kafka.py
index a843a12..734eb5c 100644
--- a/tests/kafkatest/services/kafka/kafka.py
+++ b/tests/kafkatest/services/kafka/kafka.py
@@ -38,8 +38,7 @@ Port = collections.namedtuple('Port', ['name', 'number', 'open'])
 class KafkaService(KafkaPathResolverMixin, JmxMixin, Service):
 
     PERSISTENT_ROOT = "/mnt"
-    STDOUT_CAPTURE = os.path.join(PERSISTENT_ROOT, "kafka.log")
-    STDERR_CAPTURE = os.path.join(PERSISTENT_ROOT, "kafka.log")
+    STDOUT_STDERR_CAPTURE = os.path.join(PERSISTENT_ROOT, "server-start-stdout-stderr.log")
     LOG4J_CONFIG = os.path.join(PERSISTENT_ROOT, "kafka-log4j.properties")
     # Logs such as controller.log, server.log, etc all go here
     OPERATIONAL_LOG_DIR = os.path.join(PERSISTENT_ROOT, "kafka-operational-logs")
@@ -52,6 +51,9 @@ class KafkaService(KafkaPathResolverMixin, JmxMixin, Service):
     SIMPLE_AUTHORIZER = "kafka.security.auth.SimpleAclAuthorizer"
 
     logs = {
+        "kafka_server_start_stdout_stderr": {
+            "path": STDOUT_STDERR_CAPTURE,
+            "collect_default": True},
         "kafka_operational_logs_info": {
             "path": OPERATIONAL_LOG_INFO_DIR,
             "collect_default": True},
@@ -85,6 +87,7 @@ class KafkaService(KafkaPathResolverMixin, JmxMixin, Service):
         self.topics = topics
         self.minikdc = None
         self.authorizer_class_name = authorizer_class_name
+        self.zk_set_acl = False
 
         #
         # In a heavily loaded and not very fast machine, it is
@@ -184,8 +187,8 @@ class KafkaService(KafkaPathResolverMixin, JmxMixin, Service):
         cmd += "%s %s 1>> %s 2>> %s &" % \
                (self.path.script("kafka-server-start.sh", node),
                 KafkaService.CONFIG_FILE,
-                KafkaService.STDOUT_CAPTURE,
-                KafkaService.STDERR_CAPTURE)
+                KafkaService.STDOUT_STDERR_CAPTURE,
+                KafkaService.STDOUT_STDERR_CAPTURE)
         return cmd
 
     def start_node(self, node):
@@ -199,7 +202,7 @@ class KafkaService(KafkaPathResolverMixin, JmxMixin, Service):
 
         cmd = self.start_cmd(node)
         self.logger.debug("Attempting to start KafkaService on %s with command: %s" % (str(node.account),
cmd))
-        with node.account.monitor_log(KafkaService.STDOUT_CAPTURE) as monitor:
+        with node.account.monitor_log(KafkaService.STDOUT_STDERR_CAPTURE) as monitor:
             node.account.ssh(cmd)
             monitor.wait_until("Kafka Server.*started", timeout_sec=30, err_msg="Kafka server
didn't finish startup")
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/7edaa3dd/tests/kafkatest/services/kafka/templates/kafka.properties
----------------------------------------------------------------------
diff --git a/tests/kafkatest/services/kafka/templates/kafka.properties b/tests/kafkatest/services/kafka/templates/kafka.properties
index 1f23713..4408144 100644
--- a/tests/kafkatest/services/kafka/templates/kafka.properties
+++ b/tests/kafkatest/services/kafka/templates/kafka.properties
@@ -67,9 +67,7 @@ ssl.client.auth=required
 authorizer.class.name={{ authorizer_class_name }}
 {% endif %}
 
-{% if zk_set_acl is defined %}
-zookeeper.set.acl={{zk_set_acl}}
-{% endif %}
+zookeeper.set.acl={{"true" if zk_set_acl else "false"}}
 
 zookeeper.connection.timeout.ms={{ zk_connect_timeout }}
 zookeeper.session.timeout.ms={{ zk_session_timeout }}

http://git-wip-us.apache.org/repos/asf/kafka/blob/7edaa3dd/tests/kafkatest/services/security/minikdc.py
----------------------------------------------------------------------
diff --git a/tests/kafkatest/services/security/minikdc.py b/tests/kafkatest/services/security/minikdc.py
index 0e7bb1b..3189ddc 100644
--- a/tests/kafkatest/services/security/minikdc.py
+++ b/tests/kafkatest/services/security/minikdc.py
@@ -65,7 +65,7 @@ class MiniKdc(KafkaPathResolverMixin, Service):
         self.logger.info(props_file)
 
         kafka_principals = ' '.join(['kafka/' + kafka_node.account.hostname for kafka_node
in self.kafka_nodes])
-        principals = 'client ' + kafka_principals + self.extra_principals
+        principals = 'client ' + kafka_principals + ' ' + self.extra_principals
         self.logger.info("Starting MiniKdc with principals " + principals)
 
         core_libs_jar = self.path.jar(CORE_LIBS_JAR_NAME, TRUNK)

http://git-wip-us.apache.org/repos/asf/kafka/blob/7edaa3dd/tests/kafkatest/services/security/security_config.py
----------------------------------------------------------------------
diff --git a/tests/kafkatest/services/security/security_config.py b/tests/kafkatest/services/security/security_config.py
index d7cc3c0..59a0ed4 100644
--- a/tests/kafkatest/services/security/security_config.py
+++ b/tests/kafkatest/services/security/security_config.py
@@ -115,27 +115,33 @@ class SecurityConfig(TemplateRenderer):
     def client_config(self, template_props=""):
         return SecurityConfig(self.security_protocol, client_sasl_mechanism=self.client_sasl_mechanism,
template_props=template_props)
 
+    def setup_ssl(self, node):
+        node.account.ssh("mkdir -p %s" % SecurityConfig.CONFIG_DIR, allow_fail=False)
+        node.account.scp_to(SecurityConfig.ssl_stores['ssl.keystore.location'], SecurityConfig.KEYSTORE_PATH)
+        node.account.scp_to(SecurityConfig.ssl_stores['ssl.truststore.location'], SecurityConfig.TRUSTSTORE_PATH)
+
+    def setup_sasl(self, node):
+        node.account.ssh("mkdir -p %s" % SecurityConfig.CONFIG_DIR, allow_fail=False)
+        jaas_conf_file = "jaas.conf"
+        java_version = node.account.ssh_capture("java -version")
+        if any('IBM' in line for line in java_version):
+            is_ibm_jdk = True
+        else:
+            is_ibm_jdk = False
+        jaas_conf = self.render(jaas_conf_file,  node=node, is_ibm_jdk=is_ibm_jdk,
+                                client_sasl_mechanism=self.client_sasl_mechanism,
+                                enabled_sasl_mechanisms=self.enabled_sasl_mechanisms)
+        node.account.create_file(SecurityConfig.JAAS_CONF_PATH, jaas_conf)
+        if self.has_sasl_kerberos:
+            node.account.scp_to(MiniKdc.LOCAL_KEYTAB_FILE, SecurityConfig.KEYTAB_PATH)
+            node.account.scp_to(MiniKdc.LOCAL_KRB5CONF_FILE, SecurityConfig.KRB5CONF_PATH)
+
     def setup_node(self, node):
         if self.has_ssl:
-            node.account.ssh("mkdir -p %s" % SecurityConfig.CONFIG_DIR, allow_fail=False)
-            node.account.scp_to(SecurityConfig.ssl_stores['ssl.keystore.location'], SecurityConfig.KEYSTORE_PATH)
-            node.account.scp_to(SecurityConfig.ssl_stores['ssl.truststore.location'], SecurityConfig.TRUSTSTORE_PATH)
+            self.setup_ssl(node)
 
         if self.has_sasl:
-            node.account.ssh("mkdir -p %s" % SecurityConfig.CONFIG_DIR, allow_fail=False)
-            jaas_conf_file = "jaas.conf"
-            java_version = node.account.ssh_capture("java -version")
-            if any('IBM' in line for line in java_version):
-                is_ibm_jdk = True
-            else:
-                is_ibm_jdk = False
-            jaas_conf = self.render(jaas_conf_file,  node=node, is_ibm_jdk=is_ibm_jdk,
-                                    client_sasl_mechanism=self.client_sasl_mechanism,
-                                    enabled_sasl_mechanisms=self.enabled_sasl_mechanisms)
-            node.account.create_file(SecurityConfig.JAAS_CONF_PATH, jaas_conf)
-            if self.has_sasl_kerberos:
-                node.account.scp_to(MiniKdc.LOCAL_KEYTAB_FILE, SecurityConfig.KEYTAB_PATH)
-                node.account.scp_to(MiniKdc.LOCAL_KRB5CONF_FILE, SecurityConfig.KRB5CONF_PATH)
+            self.setup_sasl(node)
 
     def clean_node(self, node):
         if self.security_protocol != SecurityConfig.PLAINTEXT:

http://git-wip-us.apache.org/repos/asf/kafka/blob/7edaa3dd/tests/kafkatest/services/zookeeper.py
----------------------------------------------------------------------
diff --git a/tests/kafkatest/services/zookeeper.py b/tests/kafkatest/services/zookeeper.py
index 07e2c0c..2019889 100644
--- a/tests/kafkatest/services/zookeeper.py
+++ b/tests/kafkatest/services/zookeeper.py
@@ -72,7 +72,8 @@ class ZookeeperService(KafkaPathResolverMixin, Service):
         self.logger.info(config_file)
         node.account.create_file("/mnt/zookeeper.properties", config_file)
 
-        start_cmd = "export KAFKA_OPTS=\"%s\";" % self.kafka_opts 
+        start_cmd = "export KAFKA_OPTS=\"%s\";" % (self.kafka_opts + ' ' + self.security_system_properties)
\
+            if self.security_config.zk_sasl else self.kafka_opts
         start_cmd += "%s " % self.path.script("zookeeper-server-start.sh", node)
         start_cmd += "/mnt/zookeeper.properties 1>> %(path)s 2>> %(path)s &"
% self.logs["zk_log"]
         node.account.ssh(start_cmd)

http://git-wip-us.apache.org/repos/asf/kafka/blob/7edaa3dd/tests/kafkatest/tests/core/zookeeper_security_upgrade_test.py
----------------------------------------------------------------------
diff --git a/tests/kafkatest/tests/core/zookeeper_security_upgrade_test.py b/tests/kafkatest/tests/core/zookeeper_security_upgrade_test.py
index 7e722f7..582eb68 100644
--- a/tests/kafkatest/tests/core/zookeeper_security_upgrade_test.py
+++ b/tests/kafkatest/tests/core/zookeeper_security_upgrade_test.py
@@ -87,7 +87,7 @@ class ZooKeeperSecurityUpgradeTest(ProduceConsumeValidateTest):
             self.zk.zookeeper_migration(node, "secure")
 
         # restart broker with zookeeper.set.acl=true and acls
-        self.kafka.zk_set_acl = "true"
+        self.kafka.zk_set_acl = True
         for node in self.kafka.nodes:
             self.kafka.stop_node(node)
             self.kafka.start_node(node)


Mime
View raw message