kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From gwens...@apache.org
Subject kafka git commit: KAFKA-2439; Add MirrorMaker service class for system tests
Date Sun, 23 Aug 2015 02:23:44 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk 6acd37720 -> 1d2ae89c5


KAFKA-2439; Add MirrorMaker service class for system tests

Added MirrorMaker service and a few corresponding sanity checks, as well as necessary config
template files. A few additional updates to accomodate the change in wait_until from ducktape0.2.0->0.3.0

Author: Geoff Anderson <geoff@confluent.io>

Reviewers: Ewen Cheslack-Postava, Gwen Shapira

Closes #148 from granders/KAFKA-2439 and squashes the following commits:

c7c3ebd [Geoff Anderson] MirrorMaker now can run as multi-node service. Added kill -9 to various
clean_node methods.
1e806f2 [Geoff Anderson] Various cleanups per review.
1b4b049 [Geoff Anderson] Added MirrorMaker service and a few corresponding sanity checks,
as well as necessary config template files. A few additional updates to accomodate the change
in wait_until from ducktape0.2.0->0.3.0


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

Branch: refs/heads/trunk
Commit: 1d2ae89c5a1dc5d18b8188bf737a8e1d195be325
Parents: 6acd377
Author: Geoff Anderson <geoff@confluent.io>
Authored: Sat Aug 22 19:23:36 2015 -0700
Committer: Gwen Shapira <cshapi@gmail.com>
Committed: Sat Aug 22 19:23:36 2015 -0700

----------------------------------------------------------------------
 .../sanity_checks/test_console_consumer.py      |  12 +-
 .../sanity_checks/test_mirror_maker.py          |  90 ++++++++++
 tests/kafkatest/services/console_consumer.py    |  17 +-
 tests/kafkatest/services/kafka.py               |   1 +
 tests/kafkatest/services/mirror_maker.py        | 165 +++++++++++++++++++
 .../templates/console_consumer.properties       |   4 +-
 .../templates/console_consumer_log4j.properties |  26 ---
 .../services/templates/consumer.properties      |  23 +++
 .../services/templates/kafka.properties         |  80 ---------
 .../services/templates/producer.properties      |  28 ++++
 .../services/templates/tools_log4j.properties   |  26 +++
 tests/kafkatest/services/verifiable_producer.py |   4 +
 tests/kafkatest/services/zookeeper.py           |  16 ++
 tests/kafkatest/tests/replication_test.py       |   8 +-
 tests/setup.py                                  |   2 +-
 15 files changed, 379 insertions(+), 123 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/1d2ae89c/tests/kafkatest/sanity_checks/test_console_consumer.py
----------------------------------------------------------------------
diff --git a/tests/kafkatest/sanity_checks/test_console_consumer.py b/tests/kafkatest/sanity_checks/test_console_consumer.py
index cd8c8f9..3e523e1 100644
--- a/tests/kafkatest/sanity_checks/test_console_consumer.py
+++ b/tests/kafkatest/sanity_checks/test_console_consumer.py
@@ -61,20 +61,20 @@ class ConsoleConsumerTest(Test):
         self.consumer.start()
         node = self.consumer.nodes[0]
 
-        if not wait_until(lambda: self.consumer.alive(node), timeout_sec=10, backoff_sec=.2):
-            raise Exception("Consumer was too slow to start")
+        wait_until(lambda: self.consumer.alive(node), 
+            timeout_sec=10, backoff_sec=.2, err_msg="Consumer was too slow to start")
         self.logger.info("consumer started in %s seconds " % str(time.time() - t0))
 
         # Verify that log output is happening
-        if not wait_until(lambda: file_exists(node, ConsoleConsumer.LOG_FILE), timeout_sec=10):
-            raise Exception("Timed out waiting for log file to exist")
+        wait_until(lambda: file_exists(node, ConsoleConsumer.LOG_FILE), timeout_sec=10,
+                   err_msg="Timed out waiting for logging to start.")
         assert line_count(node, ConsoleConsumer.LOG_FILE) > 0
 
         # Verify no consumed messages
         assert line_count(node, ConsoleConsumer.STDOUT_CAPTURE) == 0
 
         self.consumer.stop_node(node)
-        if not wait_until(lambda: not self.consumer.alive(node), timeout_sec=10, backoff_sec=.2):
-            raise Exception("Took too long for consumer to die.")
+
+
 
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/1d2ae89c/tests/kafkatest/sanity_checks/test_mirror_maker.py
----------------------------------------------------------------------
diff --git a/tests/kafkatest/sanity_checks/test_mirror_maker.py b/tests/kafkatest/sanity_checks/test_mirror_maker.py
new file mode 100644
index 0000000..3481d7a
--- /dev/null
+++ b/tests/kafkatest/sanity_checks/test_mirror_maker.py
@@ -0,0 +1,90 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+from ducktape.tests.test import Test
+
+from kafkatest.services.zookeeper import ZookeeperService
+from kafkatest.services.kafka import KafkaService
+from kafkatest.services.console_consumer import ConsoleConsumer
+from kafkatest.services.verifiable_producer import VerifiableProducer
+from kafkatest.services.mirror_maker import MirrorMaker
+
+
+class TestMirrorMakerService(Test):
+    """Sanity checks on mirror maker service class."""
+    def __init__(self, test_context):
+        super(TestMirrorMakerService, self).__init__(test_context)
+
+        self.topic = "topic"
+        self.source_zk = ZookeeperService(test_context, num_nodes=1)
+        self.target_zk = ZookeeperService(test_context, num_nodes=1)
+
+        self.source_kafka = KafkaService(test_context, num_nodes=1, zk=self.source_zk,
+                                  topics={self.topic: {"partitions": 1, "replication-factor":
1}})
+        self.target_kafka = KafkaService(test_context, num_nodes=1, zk=self.target_zk,
+                                  topics={self.topic: {"partitions": 1, "replication-factor":
1}})
+
+        self.num_messages = 1000
+        # This will produce to source kafka cluster
+        self.producer = VerifiableProducer(test_context, num_nodes=1, kafka=self.source_kafka,
topic=self.topic,
+                                           max_messages=self.num_messages, throughput=1000)
+
+        # Use a regex whitelist to check that the start command is well-formed in this case
+        self.mirror_maker = MirrorMaker(test_context, num_nodes=1, source=self.source_kafka,
target=self.target_kafka,
+                                        whitelist=".*", consumer_timeout_ms=2000)
+
+        # This will consume from target kafka cluster
+        self.consumer = ConsoleConsumer(test_context, num_nodes=1, kafka=self.target_kafka,
topic=self.topic,
+                                        consumer_timeout_ms=1000)
+
+    def setUp(self):
+        # Source cluster
+        self.source_zk.start()
+        self.source_kafka.start()
+
+        # Target cluster
+        self.target_zk.start()
+        self.target_kafka.start()
+
+    def test_end_to_end(self):
+        """
+        Test end-to-end behavior under non-failure conditions.
+
+        Setup: two single node Kafka clusters, each connected to its own single node zookeeper
cluster.
+        One is source, and the other is target. Single-node mirror maker mirrors from source
to target.
+
+        - Start mirror maker.
+        - Produce a small number of messages to the source cluster.
+        - Consume messages from target.
+        - Verify that number of consumed messages matches the number produced.
+        """
+        self.mirror_maker.start()
+        # Check that consumer_timeout_ms setting made it to config file
+        self.mirror_maker.nodes[0].account.ssh(
+            "grep \"consumer\.timeout\.ms\" %s" % MirrorMaker.CONSUMER_CONFIG, allow_fail=False)
+
+        self.producer.start()
+        self.producer.wait()
+        self.consumer.start()
+        self.consumer.wait()
+
+        num_consumed = len(self.consumer.messages_consumed[1])
+        num_produced = self.producer.num_acked
+        assert num_produced == self.num_messages, "num_produced: %d, num_messages: %d" %
(num_produced, self.num_messages)
+        assert num_produced == num_consumed, "num_produced: %d, num_consumed: %d" % (num_produced,
num_consumed)
+
+        self.mirror_maker.stop()
+

http://git-wip-us.apache.org/repos/asf/kafka/blob/1d2ae89c/tests/kafkatest/services/console_consumer.py
----------------------------------------------------------------------
diff --git a/tests/kafkatest/services/console_consumer.py b/tests/kafkatest/services/console_consumer.py
index 18c9f63..ffde6a2 100644
--- a/tests/kafkatest/services/console_consumer.py
+++ b/tests/kafkatest/services/console_consumer.py
@@ -14,8 +14,10 @@
 # limitations under the License.
 
 from ducktape.services.background_thread import BackgroundThreadService
+from ducktape.utils.util import wait_until
 
 import os
+import subprocess
 
 
 def is_int(msg):
@@ -91,7 +93,7 @@ class ConsoleConsumer(BackgroundThreadService):
             "collect_default": True}
         }
 
-    def __init__(self, context, num_nodes, kafka, topic, message_validator=is_int, from_beginning=True,
consumer_timeout_ms=None):
+    def __init__(self, context, num_nodes, kafka, topic, message_validator=None, from_beginning=True,
consumer_timeout_ms=None):
         """
         Args:
             context:                    standard context
@@ -141,7 +143,7 @@ class ConsoleConsumer(BackgroundThreadService):
             cmd = "ps ax | grep -i console_consumer | grep java | grep -v grep | awk '{print
$1}'"
             pid_arr = [pid for pid in node.account.ssh_capture(cmd, allow_fail=True, callback=int)]
             return pid_arr
-        except:
+        except (subprocess.CalledProcessError, ValueError) as e:
             return []
 
     def alive(self, node):
@@ -161,7 +163,7 @@ class ConsoleConsumer(BackgroundThreadService):
         node.account.create_file(ConsoleConsumer.CONFIG_FILE, prop_file)
 
         # Create and upload log properties
-        log_config = self.render('console_consumer_log4j.properties', log_file=ConsoleConsumer.LOG_FILE)
+        log_config = self.render('tools_log4j.properties', log_file=ConsoleConsumer.LOG_FILE)
         node.account.create_file(ConsoleConsumer.LOG4J_CONFIG, log_config)
 
         # Run and capture output
@@ -169,7 +171,8 @@ class ConsoleConsumer(BackgroundThreadService):
         self.logger.debug("Console consumer %d command: %s", idx, cmd)
         for line in node.account.ssh_capture(cmd, allow_fail=False):
             msg = line.strip()
-            msg = self.message_validator(msg)
+            if self.message_validator is not None:
+                msg = self.message_validator(msg)
             if msg is not None:
                 self.logger.debug("consumed a message: " + str(msg))
                 self.messages_consumed[idx].append(msg)
@@ -179,7 +182,13 @@ class ConsoleConsumer(BackgroundThreadService):
 
     def stop_node(self, node):
         node.account.kill_process("java", allow_fail=True)
+        wait_until(lambda: not self.alive(node), timeout_sec=10, backoff_sec=.2,
+                   err_msg="Timed out waiting for consumer to stop.")
 
     def clean_node(self, node):
+        if self.alive(node):
+            self.logger.warn("%s %s was still alive at cleanup time. Killing forcefully..."
%
+                             (self.__class__.__name__, node.account))
+        node.account.kill_process("java", clean_shutdown=False, allow_fail=True)
         node.account.ssh("rm -rf %s" % ConsoleConsumer.PERSISTENT_ROOT, allow_fail=False)
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/1d2ae89c/tests/kafkatest/services/kafka.py
----------------------------------------------------------------------
diff --git a/tests/kafkatest/services/kafka.py b/tests/kafkatest/services/kafka.py
index 34ec5ef..76f9cf6 100644
--- a/tests/kafkatest/services/kafka.py
+++ b/tests/kafkatest/services/kafka.py
@@ -93,6 +93,7 @@ class KafkaService(Service):
         node.account.ssh("rm -f /mnt/kafka.pid", allow_fail=False)
 
     def clean_node(self, node):
+        node.account.kill_process("kafka", clean_shutdown=False, allow_fail=True)
         node.account.ssh("rm -rf /mnt/kafka-logs /mnt/kafka.properties /mnt/kafka.log /mnt/kafka.pid",
allow_fail=False)
 
     def create_topic(self, topic_cfg):

http://git-wip-us.apache.org/repos/asf/kafka/blob/1d2ae89c/tests/kafkatest/services/mirror_maker.py
----------------------------------------------------------------------
diff --git a/tests/kafkatest/services/mirror_maker.py b/tests/kafkatest/services/mirror_maker.py
new file mode 100644
index 0000000..afbed13
--- /dev/null
+++ b/tests/kafkatest/services/mirror_maker.py
@@ -0,0 +1,165 @@
+
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from ducktape.services.service import Service
+from ducktape.utils.util import wait_until
+
+import os
+import subprocess
+
+"""
+0.8.2.1 MirrorMaker options
+
+Option                                  Description
+------                                  -----------
+--abort.on.send.failure <Stop the       Configure the mirror maker to exit on
+  entire mirror maker when a send         a failed send. (default: true)
+  failure occurs>
+--blacklist <Java regex (String)>       Blacklist of topics to mirror.
+--consumer.config <config file>         Embedded consumer config for consuming
+                                          from the source cluster.
+--consumer.rebalance.listener <A        The consumer rebalance listener to use
+  custom rebalance listener of type       for mirror maker consumer.
+  ConsumerRebalanceListener>
+--help                                  Print this message.
+--message.handler <A custom message     Message handler which will process
+  handler of type                         every record in-between consumer and
+  MirrorMakerMessageHandler>              producer.
+--message.handler.args <Arguments       Arguments used by custom rebalance
+  passed to message handler               listener for mirror maker consumer
+  constructor.>
+--num.streams <Integer: Number of       Number of consumption streams.
+  threads>                                (default: 1)
+--offset.commit.interval.ms <Integer:   Offset commit interval in ms (default:
+  offset commit interval in               60000)
+  millisecond>
+--producer.config <config file>         Embedded producer config.
+--rebalance.listener.args <Arguments    Arguments used by custom rebalance
+  passed to custom rebalance listener     listener for mirror maker consumer
+  constructor as a string.>
+--whitelist <Java regex (String)>       Whitelist of topics to mirror.
+"""
+
+
+class MirrorMaker(Service):
+
+    # Root directory for persistent output
+    PERSISTENT_ROOT = "/mnt/mirror_maker"
+    LOG_DIR = os.path.join(PERSISTENT_ROOT, "logs")
+    LOG_FILE = os.path.join(LOG_DIR, "mirror_maker.log")
+    LOG4J_CONFIG = os.path.join(PERSISTENT_ROOT, "tools-log4j.properties")
+    PRODUCER_CONFIG = os.path.join(PERSISTENT_ROOT, "producer.properties")
+    CONSUMER_CONFIG = os.path.join(PERSISTENT_ROOT, "consumer.properties")
+    KAFKA_HOME = "/opt/kafka/"
+
+    logs = {
+        "mirror_maker_log": {
+            "path": LOG_FILE,
+            "collect_default": True}
+        }
+
+    def __init__(self, context, num_nodes, source, target, whitelist=None, blacklist=None,
num_streams=1, consumer_timeout_ms=None):
+        """
+        MirrorMaker mirrors messages from one or more source clusters to a single destination
cluster.
+
+        Args:
+            context:                    standard context
+            source:                     source Kafka cluster
+            target:                     target Kafka cluster to which data will be mirrored
+            whitelist:                  whitelist regex for topics to mirror
+            blacklist:                  blacklist regex for topics not to mirror
+            num_streams:                number of consumer threads to create; can be a single
int, or a list with
+                                            one value per node, allowing num_streams to be
the same for each node,
+                                            or configured independently per-node
+            consumer_timeout_ms:        consumer stops if t > consumer_timeout_ms elapses
between consecutive messages
+        """
+        super(MirrorMaker, self).__init__(context, num_nodes=num_nodes)
+
+        self.consumer_timeout_ms = consumer_timeout_ms
+        self.num_streams = num_streams
+        if not isinstance(num_streams, int):
+            # if not an integer, num_streams should be configured per-node
+            assert len(num_streams) == num_nodes
+        self.whitelist = whitelist
+        self.blacklist = blacklist
+        self.source = source
+        self.target = target
+
+    def start_cmd(self, node):
+        cmd = "export LOG_DIR=%s;" % MirrorMaker.LOG_DIR
+        cmd += " export KAFKA_LOG4J_OPTS=\"-Dlog4j.configuration=file:%s\";" % MirrorMaker.LOG4J_CONFIG
+        cmd += " %s/bin/kafka-run-class.sh kafka.tools.MirrorMaker" % MirrorMaker.KAFKA_HOME
+        cmd += " --consumer.config %s" % MirrorMaker.CONSUMER_CONFIG
+        cmd += " --producer.config %s" % MirrorMaker.PRODUCER_CONFIG
+        if isinstance(self.num_streams, int):
+            cmd += " --num.streams %d" % self.num_streams
+        else:
+            # config num_streams separately on each node
+            cmd += " --num.streams %d" % self.num_streams[self.idx(node) - 1]
+        if self.whitelist is not None:
+            cmd += " --whitelist=\"%s\"" % self.whitelist
+        if self.blacklist is not None:
+            cmd += " --blacklist=\"%s\"" % self.blacklist
+        cmd += " 1>> %s 2>> %s &" % (MirrorMaker.LOG_FILE, MirrorMaker.LOG_FILE)
+        return cmd
+
+    def pids(self, node):
+        try:
+            cmd = "ps ax | grep -i MirrorMaker | grep java | grep -v grep | awk '{print $1}'"
+            pid_arr = [pid for pid in node.account.ssh_capture(cmd, allow_fail=True, callback=int)]
+            return pid_arr
+        except (subprocess.CalledProcessError, ValueError) as e:
+            return []
+
+    def alive(self, node):
+        return len(self.pids(node)) > 0
+
+    def start_node(self, node):
+        node.account.ssh("mkdir -p %s" % MirrorMaker.PERSISTENT_ROOT, allow_fail=False)
+        node.account.ssh("mkdir -p %s" % MirrorMaker.LOG_DIR, allow_fail=False)
+
+        # Create, upload one consumer config file for source cluster
+        consumer_props = self.render('consumer.properties', zookeeper_connect=self.source.zk.connect_setting())
+        node.account.create_file(MirrorMaker.CONSUMER_CONFIG, consumer_props)
+
+        # Create, upload producer properties file for target cluster
+        producer_props = self.render('producer.properties',  broker_list=self.target.bootstrap_servers(),
+                                     producer_type="async")
+        node.account.create_file(MirrorMaker.PRODUCER_CONFIG, producer_props)
+
+        # Create and upload log properties
+        log_config = self.render('tools_log4j.properties', log_file=MirrorMaker.LOG_FILE)
+        node.account.create_file(MirrorMaker.LOG4J_CONFIG, log_config)
+
+        # Run mirror maker
+        cmd = self.start_cmd(node)
+        self.logger.debug("Mirror maker command: %s", cmd)
+        node.account.ssh(cmd, allow_fail=False)
+        wait_until(lambda: self.alive(node), timeout_sec=10, backoff_sec=.5,
+                   err_msg="Mirror maker took to long to start.")
+        self.logger.debug("Mirror maker is alive")
+
+    def stop_node(self, node):
+        node.account.kill_process("java", allow_fail=True)
+        wait_until(lambda: not self.alive(node), timeout_sec=10, backoff_sec=.5,
+                   err_msg="Mirror maker took to long to stop.")
+
+    def clean_node(self, node):
+        if self.alive(node):
+            self.logger.warn("%s %s was still alive at cleanup time. Killing forcefully..."
%
+                             (self.__class__.__name__, node.account))
+        node.account.kill_process("java", clean_shutdown=False, allow_fail=True)
+        node.account.ssh("rm -rf %s" % MirrorMaker.PERSISTENT_ROOT, allow_fail=False)

http://git-wip-us.apache.org/repos/asf/kafka/blob/1d2ae89c/tests/kafkatest/services/templates/console_consumer.properties
----------------------------------------------------------------------
diff --git a/tests/kafkatest/services/templates/console_consumer.properties b/tests/kafkatest/services/templates/console_consumer.properties
index 944c2c9..7143179 100644
--- a/tests/kafkatest/services/templates/console_consumer.properties
+++ b/tests/kafkatest/services/templates/console_consumer.properties
@@ -14,6 +14,6 @@
 # limitations under the License.
 # see kafka.server.KafkaConfig for additional details and defaults
 
-{% if consumer_timeout_ms is not none %}
+{% if consumer_timeout_ms is defined and consumer_timeout_ms is not none %}
 consumer.timeout.ms={{ consumer_timeout_ms }}
-{% endif %}
\ No newline at end of file
+{% endif %}

http://git-wip-us.apache.org/repos/asf/kafka/blob/1d2ae89c/tests/kafkatest/services/templates/console_consumer_log4j.properties
----------------------------------------------------------------------
diff --git a/tests/kafkatest/services/templates/console_consumer_log4j.properties b/tests/kafkatest/services/templates/console_consumer_log4j.properties
deleted file mode 100644
index e63e6d6..0000000
--- a/tests/kafkatest/services/templates/console_consumer_log4j.properties
+++ /dev/null
@@ -1,26 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements.  See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License.  You may obtain a copy of the License at
-#
-#    http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-# Define the root logger with appender file
-log4j.rootLogger = INFO, FILE
-
-log4j.appender.FILE=org.apache.log4j.FileAppender
-log4j.appender.FILE.File={{ log_file }}
-log4j.appender.FILE.ImmediateFlush=true
-log4j.appender.FILE.Threshold=debug
-# Set the append to false, overwrite
-log4j.appender.FILE.Append=false
-log4j.appender.FILE.layout=org.apache.log4j.PatternLayout
-log4j.appender.FILE.layout.conversionPattern=[%d] %p %m (%c)%n
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/kafka/blob/1d2ae89c/tests/kafkatest/services/templates/consumer.properties
----------------------------------------------------------------------
diff --git a/tests/kafkatest/services/templates/consumer.properties b/tests/kafkatest/services/templates/consumer.properties
new file mode 100644
index 0000000..b8723d1
--- /dev/null
+++ b/tests/kafkatest/services/templates/consumer.properties
@@ -0,0 +1,23 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+# 
+#    http://www.apache.org/licenses/LICENSE-2.0
+# 
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# see kafka.consumer.ConsumerConfig for more details
+
+zookeeper.connect={{ zookeeper_connect }}
+zookeeper.connection.timeout.ms={{ zookeeper_connection_timeout_ms|default(6000) }}
+group.id={{ group_id|default('test-consumer-group') }}
+
+{% if consumer_timeout_ms is defined and consumer_timeout_ms is not none %}
+consumer.timeout.ms={{ consumer_timeout_ms }}
+{% endif %}

http://git-wip-us.apache.org/repos/asf/kafka/blob/1d2ae89c/tests/kafkatest/services/templates/kafka.properties
----------------------------------------------------------------------
diff --git a/tests/kafkatest/services/templates/kafka.properties b/tests/kafkatest/services/templates/kafka.properties
index db1077a..6650d23 100644
--- a/tests/kafkatest/services/templates/kafka.properties
+++ b/tests/kafkatest/services/templates/kafka.properties
@@ -14,108 +14,28 @@
 # limitations under the License.
 # see kafka.server.KafkaConfig for additional details and defaults
 
-############################# Server Basics #############################
 
-# The id of the broker. This must be set to a unique integer for each broker.
 broker.id={{ broker_id }}
-
-############################# Socket Server Settings #############################
-
-# The port the socket server listens on
 port=9092
-
-# Hostname the broker will bind to. If not set, the server will bind to all interfaces
 #host.name=localhost
-
-# Hostname the broker will advertise to producers and consumers. If not set, it uses the
-# value for "host.name" if configured.  Otherwise, it will use the value returned from
-# java.net.InetAddress.getCanonicalHostName().
 advertised.host.name={{ node.account.hostname }}
-
-# The port to publish to ZooKeeper for clients to use. If this is not set,
-# it will publish the same port that the broker binds to.
 #advertised.port=<port accessible by clients>
-
-# The number of threads handling network requests
 num.network.threads=3
- 
-# The number of threads doing disk I/O
 num.io.threads=8
-
-# The send buffer (SO_SNDBUF) used by the socket server
 socket.send.buffer.bytes=102400
-
-# The receive buffer (SO_RCVBUF) used by the socket server
 socket.receive.buffer.bytes=65536
-
-# The maximum size of a request that the socket server will accept (protection against OOM)
 socket.request.max.bytes=104857600
 
-
-############################# Log Basics #############################
-
-# A comma seperated list of directories under which to store log files
 log.dirs=/mnt/kafka-logs
-
-# The default number of log partitions per topic. More partitions allow greater
-# parallelism for consumption, but this will also result in more files across
-# the brokers.
 num.partitions=1
-
-# The number of threads per data directory to be used for log recovery at startup and flushing
at shutdown.
-# This value is recommended to be increased for installations with data dirs located in RAID
array.
 num.recovery.threads.per.data.dir=1
-
-############################# Log Flush Policy #############################
-
-# Messages are immediately written to the filesystem but by default we only fsync() to sync
-# the OS cache lazily. The following configurations control the flush of data to disk. 
-# There are a few important trade-offs here:
-#    1. Durability: Unflushed data may be lost if you are not using replication.
-#    2. Latency: Very large flush intervals may lead to latency spikes when the flush does
occur as there will be a lot of data to flush.
-#    3. Throughput: The flush is generally the most expensive operation, and a small flush
interval may lead to exceessive seeks. 
-# The settings below allow one to configure the flush policy to flush data after a period
of time or
-# every N messages (or both). This can be done globally and overridden on a per-topic basis.
-
-# The number of messages to accept before forcing a flush of data to disk
 #log.flush.interval.messages=10000
-
-# The maximum amount of time a message can sit in a log before we force a flush
 #log.flush.interval.ms=1000
-
-############################# Log Retention Policy #############################
-
-# The following configurations control the disposal of log segments. The policy can
-# be set to delete segments after a period of time, or after a given size has accumulated.
-# A segment will be deleted whenever *either* of these criteria are met. Deletion always
happens
-# from the end of the log.
-
-# The minimum age of a log file to be eligible for deletion
 log.retention.hours=168
-
-# A size-based retention policy for logs. Segments are pruned from the log as long as the
remaining
-# segments don't drop below log.retention.bytes.
 #log.retention.bytes=1073741824
-
-# The maximum size of a log segment file. When this size is reached a new log segment will
be created.
 log.segment.bytes=1073741824
-
-# The interval at which log segments are checked to see if they can be deleted according

-# to the retention policies
 log.retention.check.interval.ms=300000
-
-# By default the log cleaner is disabled and the log retention policy will default to just
delete segments after their retention expires.
-# If log.cleaner.enable=true is set the cleaner will be enabled and individual logs can then
be marked for log compaction.
 log.cleaner.enable=false
 
-############################# Zookeeper #############################
-
-# Zookeeper connection string (see zookeeper docs for details).
-# This is a comma separated host:port pairs, each corresponding to a zk
-# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
-# You can also append an optional chroot string to the urls to specify the
-# root directory for all kafka znodes.
 zookeeper.connect={{ zk.connect_setting() }}
-
-# Timeout in ms for connecting to zookeeper
 zookeeper.connection.timeout.ms=2000

http://git-wip-us.apache.org/repos/asf/kafka/blob/1d2ae89c/tests/kafkatest/services/templates/producer.properties
----------------------------------------------------------------------
diff --git a/tests/kafkatest/services/templates/producer.properties b/tests/kafkatest/services/templates/producer.properties
new file mode 100644
index 0000000..ede60c8
--- /dev/null
+++ b/tests/kafkatest/services/templates/producer.properties
@@ -0,0 +1,28 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# see kafka.producer.ProducerConfig for more details
+
+metadata.broker.list={{ broker_list }}
+bootstrap.servers = {{ broker_list }}
+producer.type={{ producer_type }}  # sync or async
+compression.codec=none
+serializer.class=kafka.serializer.DefaultEncoder
+
+#partitioner.class=
+#compressed.topics=
+#queue.buffering.max.ms=
+#queue.buffering.max.messages=
+#queue.enqueue.timeout.ms=
+#batch.num.messages=

http://git-wip-us.apache.org/repos/asf/kafka/blob/1d2ae89c/tests/kafkatest/services/templates/tools_log4j.properties
----------------------------------------------------------------------
diff --git a/tests/kafkatest/services/templates/tools_log4j.properties b/tests/kafkatest/services/templates/tools_log4j.properties
new file mode 100644
index 0000000..e63e6d6
--- /dev/null
+++ b/tests/kafkatest/services/templates/tools_log4j.properties
@@ -0,0 +1,26 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# Define the root logger with appender file
+log4j.rootLogger = INFO, FILE
+
+log4j.appender.FILE=org.apache.log4j.FileAppender
+log4j.appender.FILE.File={{ log_file }}
+log4j.appender.FILE.ImmediateFlush=true
+log4j.appender.FILE.Threshold=debug
+# Set the append to false, overwrite
+log4j.appender.FILE.Append=false
+log4j.appender.FILE.layout=org.apache.log4j.PatternLayout
+log4j.appender.FILE.layout.conversionPattern=[%d] %p %m (%c)%n
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/kafka/blob/1d2ae89c/tests/kafkatest/services/verifiable_producer.py
----------------------------------------------------------------------
diff --git a/tests/kafkatest/services/verifiable_producer.py b/tests/kafkatest/services/verifiable_producer.py
index cca8227..158db7a 100644
--- a/tests/kafkatest/services/verifiable_producer.py
+++ b/tests/kafkatest/services/verifiable_producer.py
@@ -89,12 +89,16 @@ class VerifiableProducer(BackgroundThreadService):
 
     def stop_node(self, node):
         node.account.kill_process("VerifiableProducer", allow_fail=False)
+        if self.worker_threads is None:
+            return
+
         # block until the corresponding thread exits
         if len(self.worker_threads) >= self.idx(node):
             # Need to guard this because stop is preemptively called before the worker threads
are added and started
             self.worker_threads[self.idx(node) - 1].join()
 
     def clean_node(self, node):
+        node.account.kill_process("VerifiableProducer", clean_shutdown=False, allow_fail=False)
         node.account.ssh("rm -rf /mnt/producer.log", allow_fail=False)
 
     def try_parse_json(self, string):

http://git-wip-us.apache.org/repos/asf/kafka/blob/1d2ae89c/tests/kafkatest/services/zookeeper.py
----------------------------------------------------------------------
diff --git a/tests/kafkatest/services/zookeeper.py b/tests/kafkatest/services/zookeeper.py
index 56f4606..09bec35 100644
--- a/tests/kafkatest/services/zookeeper.py
+++ b/tests/kafkatest/services/zookeeper.py
@@ -16,6 +16,7 @@
 
 from ducktape.services.service import Service
 
+import subprocess
 import time
 
 
@@ -51,6 +52,17 @@ class ZookeeperService(Service):
 
         time.sleep(5)  # give it some time to start
 
+    def pids(self, node):
+        try:
+            cmd = "ps ax | grep -i zookeeper | grep java | grep -v grep | awk '{print $1}'"
+            pid_arr = [pid for pid in node.account.ssh_capture(cmd, allow_fail=True, callback=int)]
+            return pid_arr
+        except (subprocess.CalledProcessError, ValueError) as e:
+            return []
+
+    def alive(self, node):
+        return len(self.pids(node)) > 0
+
     def stop_node(self, node):
         idx = self.idx(node)
         self.logger.info("Stopping %s node %d on %s" % (type(self).__name__, idx, node.account.hostname))
@@ -58,6 +70,10 @@ class ZookeeperService(Service):
 
     def clean_node(self, node):
         self.logger.info("Cleaning ZK node %d on %s", self.idx(node), node.account.hostname)
+        if self.alive(node):
+            self.logger.warn("%s %s was still alive at cleanup time. Killing forcefully..."
%
+                             (self.__class__.__name__, node.account))
+        node.account.kill_process("zookeeper", clean_shutdown=False, allow_fail=True)
         node.account.ssh("rm -rf /mnt/zookeeper /mnt/zookeeper.properties /mnt/zk.log", allow_fail=False)
 
     def connect_setting(self):

http://git-wip-us.apache.org/repos/asf/kafka/blob/1d2ae89c/tests/kafkatest/tests/replication_test.py
----------------------------------------------------------------------
diff --git a/tests/kafkatest/tests/replication_test.py b/tests/kafkatest/tests/replication_test.py
index fed1ea1..755fb42 100644
--- a/tests/kafkatest/tests/replication_test.py
+++ b/tests/kafkatest/tests/replication_test.py
@@ -19,7 +19,7 @@ from ducktape.utils.util import wait_until
 from kafkatest.services.zookeeper import ZookeeperService
 from kafkatest.services.kafka import KafkaService
 from kafkatest.services.verifiable_producer import VerifiableProducer
-from kafkatest.services.console_consumer import ConsoleConsumer
+from kafkatest.services.console_consumer import ConsoleConsumer, is_int
 
 import signal
 import time
@@ -76,12 +76,12 @@ class ReplicationTest(Test):
 
         """
         self.producer = VerifiableProducer(self.test_context, self.num_producers, self.kafka,
self.topic, throughput=self.producer_throughput)
-        self.consumer = ConsoleConsumer(self.test_context, self.num_consumers, self.kafka,
self.topic, consumer_timeout_ms=3000)
+        self.consumer = ConsoleConsumer(self.test_context, self.num_consumers, self.kafka,
self.topic, consumer_timeout_ms=3000, message_validator=is_int)
 
         # Produce in a background thread while driving broker failures
         self.producer.start()
-        if not wait_until(lambda: self.producer.num_acked > 5, timeout_sec=5):
-            raise RuntimeError("Producer failed to start in a reasonable amount of time.")
+        wait_until(lambda: self.producer.num_acked > 5, timeout_sec=5,
+             err_msg="Producer failed to start in a reasonable amount of time.")
         failure()
         self.producer.stop()
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/1d2ae89c/tests/setup.py
----------------------------------------------------------------------
diff --git a/tests/setup.py b/tests/setup.py
index 5ce4bb7..a2fa71a 100644
--- a/tests/setup.py
+++ b/tests/setup.py
@@ -23,5 +23,5 @@ setup(name="kafkatest",
       platforms=["any"], 
       license="apache2.0",
       packages=find_packages(),
-      requires=["ducktape(>=0.2.0)"]
+      requires=["ducktape(==0.3.0)"]
       )


Mime
View raw message