From 6b33dc8a8ba441e6d9a427849e04aeec6f489d51 Mon Sep 17 00:00:00 2001 From: Konstantine Karantasis Date: Fri, 12 Jan 2018 14:15:39 -0800 Subject: [PATCH 1/7] MINOR: Add async and different sync startup modes in ConnectService test class --- tests/kafkatest/services/connect.py | 64 ++++++++++++++++++++++++----- 1 file changed, 53 insertions(+), 11 deletions(-) diff --git a/tests/kafkatest/services/connect.py b/tests/kafkatest/services/connect.py index 399e53c2126ce..6009f763f83f3 100644 --- a/tests/kafkatest/services/connect.py +++ b/tests/kafkatest/services/connect.py @@ -20,6 +20,7 @@ import time import requests +from ducktape.cluster.remoteaccount import RemoteCommandError from ducktape.errors import DucktapeError from ducktape.services.service import Service from ducktape.utils.util import wait_until @@ -57,6 +58,7 @@ def __init__(self, context, num_nodes, kafka, files): self.kafka = kafka self.security_config = kafka.security_config.client_config() self.files = files + self.startup_mode = 'LISTEN' self.environment = {} def pids(self, node): @@ -76,6 +78,41 @@ def set_configs(self, config_template_func, connector_config_templates=None): self.config_template_func = config_template_func self.connector_config_templates = connector_config_templates + def listening(self, node, port=8083): + try: + cmd = "nc -z %s %s" % (node.account.hostname, port) + node.account.ssh_output(cmd, allow_fail=False) + self.logger.debug("Connect worker started accepting connections at: '%s:%s')", node.account.hostname, port) + return True + except (RemoteCommandError, ValueError) as e: + return False + + # Currently the Connect worker supports waiting on three modes: + # INSTANT: return immediately + # LOAD: return after discovering and loading plugins + # LISTEN: return after opening the REST port. + def start_async(self, mode='LISTEN'): + self.startup_mode = mode + self.start() + + def start_and_return_immediately(self, node, worker_type, remote_connector_configs): + cmd = self.start_cmd(node, remote_connector_configs) + self.logger.debug("Connect %s command: %s", worker_type, cmd) + node.account.ssh(cmd) + + def start_and_wait_to_load_plugins(self, node, worker_type, remote_connector_configs): + with node.account.monitor_log(self.LOG_FILE) as monitor: + self.start_and_return_immediately(node, worker_type, remote_connector_configs) + monitor.wait_until('Kafka Connect started', timeout_sec=60, + err_msg="Never saw message indicating Kafka Connect finished startup on node: " + + "%s in condition mode: %s" % (str(node.account), self.startup_mode)) + + def start_and_wait_to_start_listening(self, node, worker_type, remote_connector_configs): + self.start_and_return_immediately(node, worker_type, remote_connector_configs) + wait_until(lambda: self.listening(node), timeout_sec=60, + err_msg="Kafka Connect failed to start on node: %s in condition mode: %s" % + (str(node.account), self.startup_mode)) + def stop_node(self, node, clean_shutdown=True): self.logger.info((clean_shutdown and "Cleanly" or "Forcibly") + " stopping Kafka Connect on " + str(node.account)) pids = self.pids(node) @@ -229,11 +266,13 @@ def start_node(self, node): remote_connector_configs.append(target_file) self.logger.info("Starting Kafka Connect standalone process on " + str(node.account)) - with node.account.monitor_log(self.LOG_FILE) as monitor: - cmd = self.start_cmd(node, remote_connector_configs) - self.logger.debug("Connect standalone command: %s", cmd) - node.account.ssh(cmd) - monitor.wait_until('Kafka Connect started', timeout_sec=60, err_msg="Never saw message indicating Kafka Connect finished startup on " + str(node.account)) + if self.startup_mode == 'LOAD': + self.start_and_wait_to_load_plugins(node, 'standalone', remote_connector_configs) + elif self.startup_mode == 'LISTEN': + self.start_and_wait_to_start_listening(node, 'standalone', remote_connector_configs) + else: + # mode 'INSTANT' is implied + self.start_and_return_immediately(node, 'standalone', remote_connector_configs) if len(self.pids(node)) == 0: raise RuntimeError("No process ids recorded") @@ -249,7 +288,8 @@ def __init__(self, context, num_nodes, kafka, files, offsets_topic="connect-offs self.configs_topic = configs_topic self.status_topic = status_topic - def start_cmd(self, node): + # connector_configs argument is intentionally ignored in distributed service. + def start_cmd(self, node, connector_configs): cmd = "( export KAFKA_LOG4J_OPTS=\"-Dlog4j.configuration=file:%s\"; " % self.LOG4J_CONFIG_FILE cmd += "export KAFKA_OPTS=%s; " % self.security_config.kafka_opts for envvar in self.environment: @@ -268,11 +308,13 @@ def start_node(self, node): raise DucktapeError("Config files are not valid in distributed mode, submit connectors via the REST API") self.logger.info("Starting Kafka Connect distributed process on " + str(node.account)) - with node.account.monitor_log(self.LOG_FILE) as monitor: - cmd = self.start_cmd(node) - self.logger.debug("Connect distributed command: %s", cmd) - node.account.ssh(cmd) - monitor.wait_until('Kafka Connect started', timeout_sec=60, err_msg="Never saw message indicating Kafka Connect finished startup on " + str(node.account)) + if self.startup_mode == 'LOAD': + self.start_and_wait_to_load_plugins(node, 'distributed', '') + elif self.startup_mode == 'LISTEN': + self.start_and_wait_to_start_listening(node, 'distributed', '') + else: + # mode 'INSTANT' is implied + self.start_and_return_immediately(node, 'distributed', '') if len(self.pids(node)) == 0: raise RuntimeError("No process ids recorded") From c9f38654b4bd20a22b2907663782b1a18f5605f9 Mon Sep 17 00:00:00 2001 From: Konstantine Karantasis Date: Mon, 15 Jan 2018 10:12:05 -0800 Subject: [PATCH 2/7] Rename function start_cmd to worker_cmd --- tests/kafkatest/services/connect.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/tests/kafkatest/services/connect.py b/tests/kafkatest/services/connect.py index 6009f763f83f3..0fd96034b47b1 100644 --- a/tests/kafkatest/services/connect.py +++ b/tests/kafkatest/services/connect.py @@ -96,7 +96,7 @@ def start_async(self, mode='LISTEN'): self.start() def start_and_return_immediately(self, node, worker_type, remote_connector_configs): - cmd = self.start_cmd(node, remote_connector_configs) + cmd = self.worker_cmd(node, remote_connector_configs) self.logger.debug("Connect %s command: %s", worker_type, cmd) node.account.ssh(cmd) @@ -243,7 +243,7 @@ def __init__(self, context, kafka, files): def node(self): return self.nodes[0] - def start_cmd(self, node, connector_configs): + def worker_cmd(self, node, connector_configs): cmd = "( export KAFKA_LOG4J_OPTS=\"-Dlog4j.configuration=file:%s\"; " % self.LOG4J_CONFIG_FILE cmd += "export KAFKA_OPTS=%s; " % self.security_config.kafka_opts for envvar in self.environment: @@ -289,7 +289,7 @@ def __init__(self, context, num_nodes, kafka, files, offsets_topic="connect-offs self.status_topic = status_topic # connector_configs argument is intentionally ignored in distributed service. - def start_cmd(self, node, connector_configs): + def worker_cmd(self, node, connector_configs): cmd = "( export KAFKA_LOG4J_OPTS=\"-Dlog4j.configuration=file:%s\"; " % self.LOG4J_CONFIG_FILE cmd += "export KAFKA_OPTS=%s; " % self.security_config.kafka_opts for envvar in self.environment: From 848d5d65dff98f76cadf7a6ee4283ae97e510d74 Mon Sep 17 00:00:00 2001 From: Konstantine Karantasis Date: Mon, 15 Jan 2018 13:38:27 -0800 Subject: [PATCH 3/7] Fix condition for startup after plugin loading. --- tests/kafkatest/services/connect.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/kafkatest/services/connect.py b/tests/kafkatest/services/connect.py index 0fd96034b47b1..2797e850aafb7 100644 --- a/tests/kafkatest/services/connect.py +++ b/tests/kafkatest/services/connect.py @@ -103,7 +103,7 @@ def start_and_return_immediately(self, node, worker_type, remote_connector_confi def start_and_wait_to_load_plugins(self, node, worker_type, remote_connector_configs): with node.account.monitor_log(self.LOG_FILE) as monitor: self.start_and_return_immediately(node, worker_type, remote_connector_configs) - monitor.wait_until('Kafka Connect started', timeout_sec=60, + monitor.wait_until('Kafka version', timeout_sec=60, err_msg="Never saw message indicating Kafka Connect finished startup on node: " + "%s in condition mode: %s" % (str(node.account), self.startup_mode)) From d5a10b447675597a4c7a29d8d9159f18c0600df6 Mon Sep 17 00:00:00 2001 From: Konstantine Karantasis Date: Tue, 16 Jan 2018 15:46:18 -0800 Subject: [PATCH 4/7] Lift connect's rest port to a global variable. --- tests/kafkatest/services/connect.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/tests/kafkatest/services/connect.py b/tests/kafkatest/services/connect.py index 2797e850aafb7..9a5d76625ad09 100644 --- a/tests/kafkatest/services/connect.py +++ b/tests/kafkatest/services/connect.py @@ -40,6 +40,7 @@ class ConnectServiceBase(KafkaPathResolverMixin, Service): STDERR_FILE = os.path.join(PERSISTENT_ROOT, "connect.stderr") LOG4J_CONFIG_FILE = os.path.join(PERSISTENT_ROOT, "connect-log4j.properties") PID_FILE = os.path.join(PERSISTENT_ROOT, "connect.pid") + CONNECT_REST_PORT = 8083 logs = { "connect_log": { @@ -78,11 +79,12 @@ def set_configs(self, config_template_func, connector_config_templates=None): self.config_template_func = config_template_func self.connector_config_templates = connector_config_templates - def listening(self, node, port=8083): + def listening(self, node): try: - cmd = "nc -z %s %s" % (node.account.hostname, port) + cmd = "nc -z %s %s" % (node.account.hostname, self.CONNECT_REST_PORT) node.account.ssh_output(cmd, allow_fail=False) - self.logger.debug("Connect worker started accepting connections at: '%s:%s')", node.account.hostname, port) + self.logger.debug("Connect worker started accepting connections at: '%s:%s')", node.account.hostname, + self.CONNECT_REST_PORT) return True except (RemoteCommandError, ValueError) as e: return False @@ -229,7 +231,7 @@ def _rest_with_retry(self, path, body=None, node=None, method="GET", retries=40, raise exception_to_throw def _base_url(self, node): - return 'http://' + node.account.externally_routable_ip + ':' + '8083' + return 'http://' + node.account.externally_routable_ip + ':' + str(self.CONNECT_REST_PORT) class ConnectStandaloneService(ConnectServiceBase): From 441276f3cc14382a62469b302ce27cd80198e0c9 Mon Sep 17 00:00:00 2001 From: Konstantine Karantasis Date: Tue, 16 Jan 2018 17:54:02 -0800 Subject: [PATCH 5/7] Fix startup modes as static vars. --- tests/kafkatest/services/connect.py | 39 ++++++++++++++++------------- 1 file changed, 22 insertions(+), 17 deletions(-) diff --git a/tests/kafkatest/services/connect.py b/tests/kafkatest/services/connect.py index 9a5d76625ad09..3d4c65f56eb9b 100644 --- a/tests/kafkatest/services/connect.py +++ b/tests/kafkatest/services/connect.py @@ -41,6 +41,15 @@ class ConnectServiceBase(KafkaPathResolverMixin, Service): LOG4J_CONFIG_FILE = os.path.join(PERSISTENT_ROOT, "connect-log4j.properties") PID_FILE = os.path.join(PERSISTENT_ROOT, "connect.pid") CONNECT_REST_PORT = 8083 + """ + Currently the Connect worker supports waiting on three modes: + INSTANT: return immediately + LOAD: return after discovering and loading plugins + LISTEN: return after opening the REST port. + """ + STARTUP_MODE_INSTANT = 'INSTANT' + STARTUP_MODE_LOAD = 'LOAD' + STARTUP_MODE_LISTEN = 'LISTEN' logs = { "connect_log": { @@ -59,7 +68,7 @@ def __init__(self, context, num_nodes, kafka, files): self.kafka = kafka self.security_config = kafka.security_config.client_config() self.files = files - self.startup_mode = 'LISTEN' + self.startup_mode = self.STARTUP_MODE_LISTEN self.environment = {} def pids(self, node): @@ -89,13 +98,9 @@ def listening(self, node): except (RemoteCommandError, ValueError) as e: return False - # Currently the Connect worker supports waiting on three modes: - # INSTANT: return immediately - # LOAD: return after discovering and loading plugins - # LISTEN: return after opening the REST port. - def start_async(self, mode='LISTEN'): + def start(self, mode=STARTUP_MODE_LISTEN): self.startup_mode = mode - self.start() + super(ConnectServiceBase, self).start() def start_and_return_immediately(self, node, worker_type, remote_connector_configs): cmd = self.worker_cmd(node, remote_connector_configs) @@ -268,13 +273,13 @@ def start_node(self, node): remote_connector_configs.append(target_file) self.logger.info("Starting Kafka Connect standalone process on " + str(node.account)) - if self.startup_mode == 'LOAD': + if self.startup_mode == self.STARTUP_MODE_LOAD: self.start_and_wait_to_load_plugins(node, 'standalone', remote_connector_configs) - elif self.startup_mode == 'LISTEN': - self.start_and_wait_to_start_listening(node, 'standalone', remote_connector_configs) - else: - # mode 'INSTANT' is implied + elif self.startup_mode == self.STARTUP_MODE_INSTANT: self.start_and_return_immediately(node, 'standalone', remote_connector_configs) + else: + # The default mode is to wait until the complete startup of the worker + self.start_and_wait_to_start_listening(node, 'standalone', remote_connector_configs) if len(self.pids(node)) == 0: raise RuntimeError("No process ids recorded") @@ -310,13 +315,13 @@ def start_node(self, node): raise DucktapeError("Config files are not valid in distributed mode, submit connectors via the REST API") self.logger.info("Starting Kafka Connect distributed process on " + str(node.account)) - if self.startup_mode == 'LOAD': + if self.startup_mode == self.STARTUP_MODE_LOAD: self.start_and_wait_to_load_plugins(node, 'distributed', '') - elif self.startup_mode == 'LISTEN': - self.start_and_wait_to_start_listening(node, 'distributed', '') - else: - # mode 'INSTANT' is implied + elif self.startup_mode == self.STARTUP_MODE_INSTANT: self.start_and_return_immediately(node, 'distributed', '') + else: + # The default mode is to wait until the complete startup of the worker + self.start_and_wait_to_start_listening(node, 'distributed', '') if len(self.pids(node)) == 0: raise RuntimeError("No process ids recorded") From f76e5aba8f3de99d62bbf6f6189789d0943399a1 Mon Sep 17 00:00:00 2001 From: Konstantine Karantasis Date: Wed, 17 Jan 2018 07:34:17 -0800 Subject: [PATCH 6/7] Fix docstring for constants. --- tests/kafkatest/services/connect.py | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/tests/kafkatest/services/connect.py b/tests/kafkatest/services/connect.py index 3d4c65f56eb9b..605b9bdfa41b9 100644 --- a/tests/kafkatest/services/connect.py +++ b/tests/kafkatest/services/connect.py @@ -41,15 +41,14 @@ class ConnectServiceBase(KafkaPathResolverMixin, Service): LOG4J_CONFIG_FILE = os.path.join(PERSISTENT_ROOT, "connect-log4j.properties") PID_FILE = os.path.join(PERSISTENT_ROOT, "connect.pid") CONNECT_REST_PORT = 8083 - """ - Currently the Connect worker supports waiting on three modes: - INSTANT: return immediately - LOAD: return after discovering and loading plugins - LISTEN: return after opening the REST port. - """ + + # Currently the Connect worker supports waiting on three modes: STARTUP_MODE_INSTANT = 'INSTANT' + """STARTUP_MODE_INSTANT: Start Connect worker and return immediately""" STARTUP_MODE_LOAD = 'LOAD' + """STARTUP_MODE_LOAD: Start Connect worker and return after discovering and loading plugins""" STARTUP_MODE_LISTEN = 'LISTEN' + """STARTUP_MODE_LISTEN: Start Connect worker and return after opening the REST port.""" logs = { "connect_log": { From da728aa523967de0c2f124545bcd15fa754a759b Mon Sep 17 00:00:00 2001 From: Konstantine Karantasis Date: Wed, 17 Jan 2018 07:44:20 -0800 Subject: [PATCH 7/7] Restore start_cmd in connect service classes --- tests/kafkatest/services/connect.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/tests/kafkatest/services/connect.py b/tests/kafkatest/services/connect.py index 605b9bdfa41b9..d7ef204b8a21e 100644 --- a/tests/kafkatest/services/connect.py +++ b/tests/kafkatest/services/connect.py @@ -102,7 +102,7 @@ def start(self, mode=STARTUP_MODE_LISTEN): super(ConnectServiceBase, self).start() def start_and_return_immediately(self, node, worker_type, remote_connector_configs): - cmd = self.worker_cmd(node, remote_connector_configs) + cmd = self.start_cmd(node, remote_connector_configs) self.logger.debug("Connect %s command: %s", worker_type, cmd) node.account.ssh(cmd) @@ -249,7 +249,7 @@ def __init__(self, context, kafka, files): def node(self): return self.nodes[0] - def worker_cmd(self, node, connector_configs): + def start_cmd(self, node, connector_configs): cmd = "( export KAFKA_LOG4J_OPTS=\"-Dlog4j.configuration=file:%s\"; " % self.LOG4J_CONFIG_FILE cmd += "export KAFKA_OPTS=%s; " % self.security_config.kafka_opts for envvar in self.environment: @@ -295,7 +295,7 @@ def __init__(self, context, num_nodes, kafka, files, offsets_topic="connect-offs self.status_topic = status_topic # connector_configs argument is intentionally ignored in distributed service. - def worker_cmd(self, node, connector_configs): + def start_cmd(self, node, connector_configs): cmd = "( export KAFKA_LOG4J_OPTS=\"-Dlog4j.configuration=file:%s\"; " % self.LOG4J_CONFIG_FILE cmd += "export KAFKA_OPTS=%s; " % self.security_config.kafka_opts for envvar in self.environment: