kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From j...@apache.org
Subject [kafka] branch 2.0 updated: KAFKA-7773; Add end to end system test relying on verifiable consumer (#6070)
Date Sat, 18 May 2019 23:08:46 GMT
This is an automated email from the ASF dual-hosted git repository.

jgus pushed a commit to branch 2.0
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/2.0 by this push:
     new 3de3fb2  KAFKA-7773; Add end to end system test relying on verifiable consumer (#6070)
3de3fb2 is described below

commit 3de3fb27865b319d0f21ff84b33ece31c4d2e72e
Author: Jason Gustafson <jason@confluent.io>
AuthorDate: Tue Jan 8 06:14:51 2019 -0800

    KAFKA-7773; Add end to end system test relying on verifiable consumer (#6070)
    
    This commit creates an EndToEndTest base class which relies on the verifiable consumer.
This will ultimately replace ProduceConsumeValidateTest which depends on the console consumer.
The advantage is that the verifiable consumer exposes more information to use for validation.
It also allows for a nicer shutdown pattern. Rather than relying on the console consumer idle
timeout, which requires a minimum wait time, we can halt consumption after we have reached
the last acked offsets. Thi [...]
---
 tests/kafkatest/services/kafka/kafka.py            |   4 +
 .../services/kafka/templates/kafka.properties      |   1 +
 tests/kafkatest/services/verifiable_consumer.py    |   8 +-
 tests/kafkatest/services/verifiable_producer.py    |   9 ++
 tests/kafkatest/tests/core/replication_test.py     |  64 ++++-----
 tests/kafkatest/tests/core/security_test.py        |  61 +++------
 tests/kafkatest/tests/end_to_end.py                | 151 +++++++++++++++++++++
 tests/kafkatest/tests/produce_consume_validate.py  |  68 ++--------
 tests/kafkatest/utils/__init__.py                  |   2 +-
 tests/kafkatest/utils/util.py                      |  56 ++++++++
 10 files changed, 291 insertions(+), 133 deletions(-)

diff --git a/tests/kafkatest/services/kafka/kafka.py b/tests/kafkatest/services/kafka/kafka.py
index 3d198b1..283bbe5 100644
--- a/tests/kafkatest/services/kafka/kafka.py
+++ b/tests/kafkatest/services/kafka/kafka.py
@@ -521,6 +521,10 @@ class KafkaService(KafkaPathResolverMixin, JmxMixin, Service):
 
         return missing
 
+    def restart_cluster(self, clean_shutdown=True):
+        for node in self.nodes:
+            self.restart_node(node, clean_shutdown=clean_shutdown)
+
     def restart_node(self, node, clean_shutdown=True):
         """Restart the given node."""
         self.stop_node(node, clean_shutdown)
diff --git a/tests/kafkatest/services/kafka/templates/kafka.properties b/tests/kafkatest/services/kafka/templates/kafka.properties
index 8cca14f..19deee3 100644
--- a/tests/kafkatest/services/kafka/templates/kafka.properties
+++ b/tests/kafkatest/services/kafka/templates/kafka.properties
@@ -62,6 +62,7 @@ replica.lag.time.max.ms={{replica_lag}}
 {% if auto_create_topics_enable is defined and auto_create_topics_enable is not none %}
 auto.create.topics.enable={{ auto_create_topics_enable }}
 {% endif %}
+offsets.topic.num.partitions={{ num_nodes }}
 offsets.topic.replication.factor={{ 3 if num_nodes > 3 else num_nodes }}
 # Set to a low, but non-zero value to exercise this path without making tests much slower
 group.initial.rebalance.delay.ms=100
diff --git a/tests/kafkatest/services/verifiable_consumer.py b/tests/kafkatest/services/verifiable_consumer.py
index 95970d9..c0e186f 100644
--- a/tests/kafkatest/services/verifiable_consumer.py
+++ b/tests/kafkatest/services/verifiable_consumer.py
@@ -161,7 +161,8 @@ class VerifiableConsumer(KafkaPathResolverMixin, VerifiableClientMixin,
Backgrou
     def __init__(self, context, num_nodes, kafka, topic, group_id,
                  max_messages=-1, session_timeout_sec=30, enable_autocommit=False,
                  assignment_strategy="org.apache.kafka.clients.consumer.RangeAssignor",
-                 version=DEV_BRANCH, stop_timeout_sec=30, log_level="INFO"):
+                 version=DEV_BRANCH, stop_timeout_sec=30, log_level="INFO",
+                 on_record_consumed=None):
         super(VerifiableConsumer, self).__init__(context, num_nodes)
         self.log_level = log_level
         
@@ -174,6 +175,7 @@ class VerifiableConsumer(KafkaPathResolverMixin, VerifiableClientMixin,
Backgrou
         self.assignment_strategy = assignment_strategy
         self.prop_file = ""
         self.stop_timeout_sec = stop_timeout_sec
+        self.on_record_consumed = on_record_consumed
 
         self.event_handlers = {}
         self.global_position = {}
@@ -223,6 +225,8 @@ class VerifiableConsumer(KafkaPathResolverMixin, VerifiableClientMixin,
Backgrou
                     elif name == "records_consumed":
                         handler.handle_records_consumed(event)
                         self._update_global_position(event, node)
+                    elif name == "record_data" and self.on_record_consumed:
+                        self.on_record_consumed(event, node)
                     elif name == "partitions_revoked":
                         handler.handle_partitions_revoked(event)
                     elif name == "partitions_assigned":
@@ -263,6 +267,8 @@ class VerifiableConsumer(KafkaPathResolverMixin, VerifiableClientMixin,
Backgrou
         cmd += " export KAFKA_OPTS=%s;" % self.security_config.kafka_opts
         cmd += " export KAFKA_LOG4J_OPTS=\"-Dlog4j.configuration=file:%s\"; " % VerifiableConsumer.LOG4J_CONFIG
         cmd += self.impl.exec_cmd(node)
+        if self.on_record_consumed:
+            cmd += " --verbose"
         cmd += " --group-id %s --topic %s --broker-list %s --session-timeout %s --assignment-strategy
%s %s" % \
                (self.group_id, self.topic, self.kafka.bootstrap_servers(self.security_config.security_protocol),
                self.session_timeout_sec*1000, self.assignment_strategy, "--enable-autocommit"
if self.enable_autocommit else "")
diff --git a/tests/kafkatest/services/verifiable_producer.py b/tests/kafkatest/services/verifiable_producer.py
index cbce27e..5c7152d 100644
--- a/tests/kafkatest/services/verifiable_producer.py
+++ b/tests/kafkatest/services/verifiable_producer.py
@@ -20,6 +20,7 @@ import time
 from ducktape.cluster.remoteaccount import RemoteCommandError
 from ducktape.services.background_thread import BackgroundThreadService
 from kafkatest.directory_layout.kafka_path import KafkaPathResolverMixin
+from kafkatest.services.kafka import TopicPartition
 from kafkatest.services.verifiable_client import VerifiableClientMixin
 from kafkatest.utils import is_int, is_int_with_prefix
 from kafkatest.version import DEV_BRANCH
@@ -84,6 +85,7 @@ class VerifiableProducer(KafkaPathResolverMixin, VerifiableClientMixin,
Backgrou
         for node in self.nodes:
             node.version = version
         self.acked_values = []
+        self._last_acked_offsets = {}
         self.not_acked_values = []
         self.produced_count = {}
         self.clean_shutdown_nodes = set()
@@ -156,7 +158,9 @@ class VerifiableProducer(KafkaPathResolverMixin, VerifiableClientMixin,
Backgrou
                         self.produced_count[idx] += 1
 
                     elif data["name"] == "producer_send_success":
+                        partition = TopicPartition(data["topic"], data["partition"])
                         self.acked_values.append(self.message_validator(data["value"]))
+                        self._last_acked_offsets[partition] = data["offset"]
                         self.produced_count[idx] += 1
 
                         # Log information if there is a large gap between successively acknowledged
messages
@@ -218,6 +222,11 @@ class VerifiableProducer(KafkaPathResolverMixin, VerifiableClientMixin,
Backgrou
         return len(self.pids(node)) > 0
 
     @property
+    def last_acked_offsets(self):
+        with self.lock:
+            return self._last_acked_offsets
+
+    @property
     def acked(self):
         with self.lock:
             return self.acked_values
diff --git a/tests/kafkatest/tests/core/replication_test.py b/tests/kafkatest/tests/core/replication_test.py
index c16d679..f5c6422 100644
--- a/tests/kafkatest/tests/core/replication_test.py
+++ b/tests/kafkatest/tests/core/replication_test.py
@@ -19,12 +19,7 @@ from ducktape.mark import matrix
 from ducktape.mark import parametrize
 from ducktape.mark.resource import cluster
 
-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.tests.produce_consume_validate import ProduceConsumeValidateTest
-from kafkatest.utils import is_int
+from kafkatest.tests.end_to_end import EndToEndTest
 
 import signal
 
@@ -83,7 +78,7 @@ failures = {
 }
 
 
-class ReplicationTest(ProduceConsumeValidateTest):
+class ReplicationTest(EndToEndTest):
     """
     Note that consuming is a bit tricky, at least with console consumer. The goal is to consume
all messages
     (foreach partition) in the topic. In this case, waiting for the last message may cause
the consumer to stop
@@ -98,25 +93,16 @@ class ReplicationTest(ProduceConsumeValidateTest):
     indicator that nothing is left to consume.
     """
 
+    TOPIC_CONFIG = {
+        "partitions": 3,
+        "replication-factor": 3,
+        "configs": {"min.insync.replicas": 2}
+    }
+ 
     def __init__(self, test_context):
         """:type test_context: ducktape.tests.test.TestContext"""
-        super(ReplicationTest, self).__init__(test_context=test_context)
-
-        self.topic = "test_topic"
-        self.zk = ZookeeperService(test_context, num_nodes=1)
-        self.kafka = KafkaService(test_context, num_nodes=3, zk=self.zk,
-                                  topics={self.topic: {
-                                      "partitions": 3,
-                                      "replication-factor": 3,
-                                      'configs': {"min.insync.replicas": 2}}
-                                  })
-        self.producer_throughput = 1000
-        self.num_producers = 1
-        self.num_consumers = 1
-
-    def setUp(self):
-        self.zk.start()
-
+        super(ReplicationTest, self).__init__(test_context=test_context, topic_config=self.TOPIC_CONFIG)
+ 
     def min_cluster_size(self):
         """Override this since we're adding services outside of the constructor"""
         return super(ReplicationTest, self).min_cluster_size() + self.num_producers + self.num_consumers
@@ -156,15 +142,23 @@ class ReplicationTest(ProduceConsumeValidateTest):
             - Validate that every acked message was consumed
         """
 
-        self.kafka.security_protocol = security_protocol
-        self.kafka.interbroker_security_protocol = security_protocol
-        self.kafka.client_sasl_mechanism = client_sasl_mechanism
-        self.kafka.interbroker_sasl_mechanism = interbroker_sasl_mechanism
-        self.enable_idempotence = enable_idempotence
-        compression_types = None if not compression_type else [compression_type] * self.num_producers
-        self.producer = VerifiableProducer(self.test_context, self.num_producers, self.kafka,
self.topic,
-                                           throughput=self.producer_throughput, compression_types=compression_types,
-                                           enable_idempotence=enable_idempotence)
-        self.consumer = ConsoleConsumer(self.test_context, self.num_consumers, self.kafka,
self.topic, consumer_timeout_ms=60000, message_validator=is_int)
+        self.create_zookeeper()
+        self.zk.start()
+
+        self.create_kafka(num_nodes=3,
+                          security_protocol=security_protocol,
+                          interbroker_security_protocol=security_protocol,
+                          client_sasl_mechanism=client_sasl_mechanism,
+                          interbroker_sasl_mechanism=interbroker_sasl_mechanism)
         self.kafka.start()
-        self.run_produce_consume_validate(core_test_action=lambda: failures[failure_mode](self,
broker_type))
+
+        compression_types = None if not compression_type else [compression_type]
+        self.create_producer(compression_types=compression_types, enable_idempotence=enable_idempotence)
+        self.producer.start()
+
+        self.create_consumer(log_level="DEBUG")
+        self.consumer.start()
+
+        self.await_startup()
+        failures[failure_mode](self, broker_type)
+        self.run_validation(enable_idempotence=enable_idempotence)
diff --git a/tests/kafkatest/tests/core/security_test.py b/tests/kafkatest/tests/core/security_test.py
index 4edbcff..d62735a 100644
--- a/tests/kafkatest/tests/core/security_test.py
+++ b/tests/kafkatest/tests/core/security_test.py
@@ -19,14 +19,9 @@ from ducktape.mark.resource import cluster
 from ducktape.utils.util import wait_until
 from ducktape.errors import TimeoutError
 
-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.security.security_config import SecurityConfig
 from kafkatest.services.security.security_config import SslStores
-from kafkatest.tests.produce_consume_validate import ProduceConsumeValidateTest
-from kafkatest.utils import is_int
+from kafkatest.tests.end_to_end import EndToEndTest
 
 class TestSslStores(SslStores):
     def __init__(self, local_scratch_dir, valid_hostname=True):
@@ -41,7 +36,7 @@ class TestSslStores(SslStores):
         else:
             return "invalidhostname"
 
-class SecurityTest(ProduceConsumeValidateTest):
+class SecurityTest(EndToEndTest):
     """
     These tests validate security features.
     """
@@ -50,21 +45,6 @@ class SecurityTest(ProduceConsumeValidateTest):
         """:type test_context: ducktape.tests.test.TestContext"""
         super(SecurityTest, self).__init__(test_context=test_context)
 
-        self.topic = "test_topic"
-        self.zk = ZookeeperService(test_context, num_nodes=1)
-        self.kafka = KafkaService(test_context, num_nodes=1, zk=self.zk, topics={self.topic:
{
-                                                                    "partitions": 2,
-                                                                    "replication-factor":
1}
-                                                                })
-        self.num_partitions = 2
-        self.timeout_sec = 10000
-        self.producer_throughput = 1000
-        self.num_producers = 1
-        self.num_consumers = 1
-
-    def setUp(self):
-        self.zk.start()
-
     def producer_consumer_have_expected_error(self, error):
         try:
             for node in self.producer.nodes:
@@ -87,16 +67,19 @@ class SecurityTest(ProduceConsumeValidateTest):
         with hostname verification failure. Hence clients are expected to fail with LEADER_NOT_AVAILABLE.
         """
 
-        self.kafka.security_protocol = security_protocol
-        self.kafka.interbroker_security_protocol = interbroker_security_protocol
-        SecurityConfig.ssl_stores = TestSslStores(self.test_context.local_scratch_dir, valid_hostname=False)
+        SecurityConfig.ssl_stores = TestSslStores(self.test_context.local_scratch_dir,
+                                                  valid_hostname=False)
+
+        self.create_zookeeper()
+        self.zk.start()
 
+        self.create_kafka(security_protocol=security_protocol,
+                          interbroker_security_protocol=interbroker_security_protocol)
         self.kafka.start()
-        self.create_producer_and_consumer()
-        self.producer.log_level = "TRACE"
 
-        self.producer.start()
-        self.consumer.start()
+        # We need more verbose logging to catch the expected errors
+        self.create_and_start_clients(log_level="DEBUG")
+
         try:
             wait_until(lambda: self.producer.num_acked > 0, timeout_sec=5)
 
@@ -108,20 +91,18 @@ class SecurityTest(ProduceConsumeValidateTest):
             pass
 
         error = 'SSLHandshakeException' if security_protocol == 'SSL' else 'LEADER_NOT_AVAILABLE'
-        wait_until(lambda: self.producer_consumer_have_expected_error(error), timeout_sec=5)
-
+        wait_until(lambda: self.producer_consumer_have_expected_error(error), timeout_sec=30)
         self.producer.stop()
         self.consumer.stop()
-        self.producer.log_level = "INFO"
 
         SecurityConfig.ssl_stores.valid_hostname = True
-        for node in self.kafka.nodes:
-            self.kafka.restart_node(node, clean_shutdown=True)
-
-        self.create_producer_and_consumer()
-        self.run_produce_consume_validate()
+        self.kafka.restart_cluster()
+        self.create_and_start_clients(log_level="INFO")
+        self.run_validation()
 
-    def create_producer_and_consumer(self):
-        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=10000, message_validator=is_int)
+    def create_and_start_clients(self, log_level):
+        self.create_producer(log_level=log_level)
+        self.producer.start()
 
+        self.create_consumer(log_level=log_level)
+        self.consumer.start()
diff --git a/tests/kafkatest/tests/end_to_end.py b/tests/kafkatest/tests/end_to_end.py
new file mode 100644
index 0000000..9cc6b41
--- /dev/null
+++ b/tests/kafkatest/tests/end_to_end.py
@@ -0,0 +1,151 @@
+# 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 ducktape.utils.util import wait_until
+
+from kafkatest.services.kafka import KafkaService
+from kafkatest.services.kafka import TopicPartition
+from kafkatest.services.verifiable_producer import VerifiableProducer
+from kafkatest.services.verifiable_consumer import VerifiableConsumer
+from kafkatest.services.zookeeper import ZookeeperService
+from kafkatest.utils import validate_delivery
+
+import time
+
+class EndToEndTest(Test):
+    """This class provides a shared template for tests which follow the common pattern of:
+
+        - produce to a topic in the background
+        - consume from that topic in the background
+        - run some logic, e.g. fail topic leader etc.
+        - perform validation
+    """
+
+    DEFAULT_TOPIC_CONFIG = {"partitions": 2, "replication-factor": 1}
+
+    def __init__(self, test_context, topic="test_topic", topic_config=DEFAULT_TOPIC_CONFIG):
+        super(EndToEndTest, self).__init__(test_context=test_context)
+        self.topic = topic
+        self.topic_config = topic_config
+        self.records_consumed = []
+        self.last_consumed_offsets = {}
+        
+    def create_zookeeper(self, num_nodes=1, **kwargs):
+        self.zk = ZookeeperService(self.test_context, num_nodes=num_nodes, **kwargs)
+
+    def create_kafka(self, num_nodes=1, **kwargs):
+        group_metadata_config = {
+            "partitions": num_nodes,
+            "replication-factor": min(num_nodes, 3),
+            "configs": {"cleanup.policy": "compact"}
+        }
+
+        topics = {
+            self.topic: self.topic_config,
+            "__consumer_offsets": group_metadata_config
+        }
+        self.kafka = KafkaService(self.test_context, num_nodes=num_nodes,
+                                  zk=self.zk, topics=topics, **kwargs)
+
+    def create_consumer(self, num_nodes=1, group_id="test_group", **kwargs):
+        self.consumer = VerifiableConsumer(self.test_context,
+                                           num_nodes=num_nodes,
+                                           kafka=self.kafka,
+                                           topic=self.topic,
+                                           group_id=group_id,
+                                           on_record_consumed=self.on_record_consumed,
+                                           **kwargs)
+                                    
+
+    def create_producer(self, num_nodes=1, throughput=1000, **kwargs):
+        self.producer = VerifiableProducer(self.test_context,
+                                           num_nodes=num_nodes,
+                                           kafka=self.kafka,
+                                           topic=self.topic,
+                                           throughput=throughput,
+                                           **kwargs)
+
+    def on_record_consumed(self, record, node):
+        partition = TopicPartition(record["topic"], record["partition"])
+        record_id = int(record["value"])
+        offset = record["offset"]
+        self.last_consumed_offsets[partition] = offset
+        self.records_consumed.append(record_id)
+
+    def await_consumed_offsets(self, last_acked_offsets, timeout_sec):
+        def has_finished_consuming():
+            for partition, offset in last_acked_offsets.iteritems():
+                if not partition in self.last_consumed_offsets:
+                    return False
+                if self.last_consumed_offsets[partition] < offset:
+                    return False
+            return True
+
+        wait_until(has_finished_consuming,
+                   timeout_sec=timeout_sec,
+                   err_msg="Consumer failed to consume up to offsets %s after waiting %ds."
%\
+                   (str(last_acked_offsets), timeout_sec))
+
+
+    def _collect_all_logs(self):
+        for s in self.test_context.services:
+            self.mark_for_collect(s)
+
+    def await_startup(self, min_records=5, timeout_sec=30):
+        try:
+            wait_until(lambda: self.consumer.total_consumed() >= min_records,
+                       timeout_sec=timeout_sec,
+                       err_msg="Timed out after %ds while awaiting initial record delivery
of %d records" %\
+                       (timeout_sec, min_records))
+        except BaseException:
+            self._collect_all_logs()
+            raise
+
+    def run_validation(self, min_records=5000, producer_timeout_sec=30,
+                       consumer_timeout_sec=30, enable_idempotence=False):
+        try:
+            wait_until(lambda: self.producer.num_acked > min_records,
+                       timeout_sec=producer_timeout_sec,
+                       err_msg="Producer failed to produce messages for %ds." %\
+                       producer_timeout_sec)
+
+            self.logger.info("Stopping producer after writing up to offsets %s" %\
+                         str(self.producer.last_acked_offsets))
+            self.producer.stop()
+
+            self.await_consumed_offsets(self.producer.last_acked_offsets, consumer_timeout_sec)
+            self.consumer.stop()
+            
+            self.validate(enable_idempotence)
+        except BaseException:
+            self._collect_all_logs()
+            raise
+
+    def validate(self, enable_idempotence):
+        self.logger.info("Number of acked records: %d" % len(self.producer.acked))
+        self.logger.info("Number of consumed records: %d" % len(self.records_consumed))
+
+        def check_lost_data(missing_records):
+            return self.kafka.search_data_files(self.topic, missing_records)
+
+        succeeded, error_msg = validate_delivery(self.producer.acked, self.records_consumed,
+                                                 enable_idempotence, check_lost_data)
+
+        # Collect all logs if validation fails
+        if not succeeded:
+            self._collect_all_logs()
+
+        assert succeeded, error_msg
diff --git a/tests/kafkatest/tests/produce_consume_validate.py b/tests/kafkatest/tests/produce_consume_validate.py
index 7a78da3..e49d02e 100644
--- a/tests/kafkatest/tests/produce_consume_validate.py
+++ b/tests/kafkatest/tests/produce_consume_validate.py
@@ -15,6 +15,9 @@
 
 from ducktape.tests.test import Test
 from ducktape.utils.util import wait_until
+
+from kafkatest.utils import validate_delivery
+
 import time
 
 class ProduceConsumeValidateTest(Test):
@@ -115,68 +118,21 @@ class ProduceConsumeValidateTest(Test):
                 self.mark_for_collect(s)
             raise
 
-    @staticmethod
-    def annotate_missing_msgs(missing, acked, consumed, msg):
-        missing_list = list(missing)
-        msg += "%s acked message did not make it to the Consumer. They are: " %\
-            len(missing_list)
-        if len(missing_list) < 20:
-            msg += str(missing_list) + ". "
-        else:
-            msg += ", ".join(str(m) for m in missing_list[:20])
-            msg += "...plus %s more. Total Acked: %s, Total Consumed: %s. " \
-                   % (len(missing_list) - 20, len(set(acked)), len(set(consumed)))
-        return msg
-
-    @staticmethod
-    def annotate_data_lost(data_lost, msg, number_validated):
-        print_limit = 10
-        if len(data_lost) > 0:
-            msg += "The first %s missing messages were validated to ensure they are in Kafka's
data files. " \
-                   "%s were missing. This suggests data loss. Here are some of the messages
not found in the data files: %s\n" \
-                   % (number_validated, len(data_lost), str(data_lost[0:print_limit]) if
len(data_lost) > print_limit else str(data_lost))
-        else:
-            msg += "We validated that the first %s of these missing messages correctly made
it into Kafka's data files. " \
-                   "This suggests they were lost on their way to the consumer." % number_validated
-        return msg
-
     def validate(self):
-        """Check that each acked message was consumed."""
-        success = True
-        msg = ""
-        acked = self.producer.acked
-        consumed = self.consumer.messages_consumed[1]
-        # Correctness of the set difference operation depends on using equivalent message_validators
in procuder and consumer
-        missing = set(acked) - set(consumed)
-
-        self.logger.info("num consumed:  %d" % len(consumed))
-
-        # Were all acked messages consumed?
-        if len(missing) > 0:
-            msg = self.annotate_missing_msgs(missing, acked, consumed, msg)
-            success = False
-
-            #Did we miss anything due to data loss?
-            to_validate = list(missing)[0:1000 if len(missing) > 1000 else len(missing)]
-            data_lost = self.kafka.search_data_files(self.topic, to_validate)
-            msg = self.annotate_data_lost(data_lost, msg, len(to_validate))
+        messages_consumed = self.consumer.messages_consumed[1]
 
+        self.logger.info("Number of acked records: %d" % len(self.producer.acked))
+        self.logger.info("Number of consumed records: %d" % len(messages_consumed))
 
-        if self.enable_idempotence:
-            self.logger.info("Ran a test with idempotence enabled. We expect no duplicates")
-        else:
-            self.logger.info("Ran a test with idempotence disabled.")
+        def check_lost_data(missing_records):
+            return self.kafka.search_data_files(self.topic, missing_records)
 
-        # Are there duplicates?
-        if len(set(consumed)) != len(consumed):
-            num_duplicates = abs(len(set(consumed)) - len(consumed))
-            msg += "(There are also %s duplicate messages in the log - but that is an acceptable
outcome)\n" % num_duplicates
-            if self.enable_idempotence:
-                assert False, "Detected %s duplicates even though idempotence was enabled."
% num_duplicates
+        succeeded, error_msg = validate_delivery(self.producer.acked, messages_consumed,
+                                                 self.enable_idempotence, check_lost_data)
 
         # Collect all logs if validation fails
-        if not success:
+        if not succeeded:
             for s in self.test_context.services:
                 self.mark_for_collect(s)
 
-        assert success, msg
+        assert succeeded, error_msg
diff --git a/tests/kafkatest/utils/__init__.py b/tests/kafkatest/utils/__init__.py
index 8c473bf..1c1d5e0 100644
--- a/tests/kafkatest/utils/__init__.py
+++ b/tests/kafkatest/utils/__init__.py
@@ -13,4 +13,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from util import kafkatest_version, is_version, is_int, is_int_with_prefix, node_is_reachable
+from util import kafkatest_version, is_version, is_int, is_int_with_prefix, node_is_reachable,
validate_delivery
diff --git a/tests/kafkatest/utils/util.py b/tests/kafkatest/utils/util.py
index dd20273..b9ccaf8 100644
--- a/tests/kafkatest/utils/util.py
+++ b/tests/kafkatest/utils/util.py
@@ -112,3 +112,59 @@ def node_is_reachable(src_node, dst_node):
     :return:                True only if dst is reachable from src.
     """
     return 0 == src_node.account.ssh("nc -w 3 -z %s 22" % dst_node.account.hostname, allow_fail=True)
+
+
+def annotate_missing_msgs(missing, acked, consumed, msg):
+    missing_list = list(missing)
+    msg += "%s acked message did not make it to the Consumer. They are: " %\
+        len(missing_list)
+    if len(missing_list) < 20:
+        msg += str(missing_list) + ". "
+    else:
+        msg += ", ".join(str(m) for m in missing_list[:20])
+        msg += "...plus %s more. Total Acked: %s, Total Consumed: %s. " \
+            % (len(missing_list) - 20, len(set(acked)), len(set(consumed)))
+    return msg
+
+def annotate_data_lost(data_lost, msg, number_validated):
+    print_limit = 10
+    if len(data_lost) > 0:
+        msg += "The first %s missing messages were validated to ensure they are in Kafka's
data files. " \
+            "%s were missing. This suggests data loss. Here are some of the messages not
found in the data files: %s\n" \
+            % (number_validated, len(data_lost), str(data_lost[0:print_limit]) if len(data_lost)
> print_limit else str(data_lost))
+    else:
+        msg += "We validated that the first %s of these missing messages correctly made it
into Kafka's data files. " \
+            "This suggests they were lost on their way to the consumer." % number_validated
+    return msg
+
+def validate_delivery(acked, consumed, idempotence_enabled=False, check_lost_data=None):
+    """Check that each acked message was consumed."""
+    success = True
+    msg = ""
+
+    # Correctness of the set difference operation depends on using equivalent
+    # message_validators in producer and consumer
+    missing = set(acked) - set(consumed)
+    
+    # Were all acked messages consumed?
+    if len(missing) > 0:
+        msg = annotate_missing_msgs(missing, acked, consumed, msg)
+        success = False
+        
+        # Did we miss anything due to data loss?
+        if check_lost_data:
+            to_validate = list(missing)[0:1000 if len(missing) > 1000 else len(missing)]
+            data_lost = check_lost_data(to_validate)
+            msg = annotate_data_lost(data_lost, msg, len(to_validate))
+
+    # Are there duplicates?
+    if len(set(consumed)) != len(consumed):
+        num_duplicates = abs(len(set(consumed)) - len(consumed))
+
+        if idempotence_enabled:
+            success = False
+            msg += "Detected %d duplicates even though idempotence was enabled.\n" % num_duplicates
+        else:
+            msg += "(There are also %d duplicate messages in the log - but that is an acceptable
outcome)\n" % num_duplicates
+
+    return success, msg


Mime
View raw message