KAFKA-9662: Wait for consumer offset reset in throttle test to avoid …

…losing early messages (apache#8227)

tests/kafkatest/services/console_consumer.py

@@ -16,6 +16,7 @@
 import itertools
 import os
 
+from ducktape.cluster.remoteaccount import RemoteCommandError
 from ducktape.services.background_thread import BackgroundThreadService
 from ducktape.utils.util import wait_until
 
@@ -320,3 +321,18 @@ def clean_node(self, node):
 
     def java_class_name(self):
         return "ConsoleConsumer"
+
+    def has_log_message(self, node, message):
+        try:
+            node.account.ssh("grep '%s' %s" % (message, ConsoleConsumer.LOG_FILE))
+        except RemoteCommandError:
+            return False
+        return True
+
+    def wait_for_offset_reset(self, node, topic, num_partitions):
+        for partition in range(num_partitions):
+            message = "Resetting offset for partition %s-%d" % (topic, partition)
+            wait_until(lambda: self.has_log_message(node, message),
+                       timeout_sec=60,
+                       err_msg="Offset not reset for partition %s-%d" % (topic, partition))
+

tests/kafkatest/tests/produce_consume_validate.py

@@ -59,6 +59,11 @@ def start_producer_and_consumer(self):
                        err_msg="Consumer process took more than %d s to fork" %\
                        self.consumer_init_timeout_sec)
 
+        # If consuming only latest messages, wait for offset reset to ensure all messages are consumed
+        if not self.consumer.from_beginning:
+            self.consumer.wait_for_offset_reset(self.consumer.nodes[0], self.topic, self.num_partitions)
+
+
         self.producer.start()
         wait_until(lambda: self.producer.num_acked > 5,
                    timeout_sec=self.producer_start_timeout_sec,