[GitHub] [kafka] ableegoldman commented on a change in pull request #8913: KAFKA-10191 fix flaky StreamsOptimizedTest

[GitBox <gi...@apache.org>]

ableegoldman commented on a change in pull request #8913:
URL: https://github.com/apache/kafka/pull/8913#discussion_r448698768



##########
File path: tests/kafkatest/services/streams.py
##########
@@ -445,6 +449,40 @@ def __init__(self, test_context, kafka, configs):
                                                         "org.apache.kafka.streams.tests.StreamsStandByReplicaTest",
                                                         configs)
 
+class StreamsResetter(StreamsTestBaseService):
+    def __init__(self, test_context, kafka):
+        super(StreamsResetter, self).__init__(test_context,
+                                              kafka,
+                                              "kafka.tools.StreamsResetter",
+                                              "")
+
+    @property
+    def expectedMessage(self):
+        return 'succeed to reset stream application'
+
+    def start_cmd(self, node):
+        args = self.args.copy()
+        args['bootstrap.servers'] = self.kafka.bootstrap_servers()
+        args['stdout'] = self.STDOUT_FILE
+        args['stderr'] = self.STDERR_FILE
+        args['pidfile'] = self.PID_FILE
+        args['log4j'] = self.LOG4J_CONFIG_FILE
+        args['kafka_run_class'] = self.path.script("kafka-run-class.sh", node)
+
+        cmd = "(export KAFKA_LOG4J_OPTS=\"-Dlog4j.configuration=file:%(log4j)s\"; " \
+              "%(kafka_run_class)s %(streams_class_name)s " \
+              "--bootstrap-servers %(bootstrap.servers)s " \
+              "--force " \
+              "--application-id StreamsOptimizedTest " \

Review comment:
       If you look at the StreamsResetter logs, it prints `No input or intermediate topics specified. Skipping seek.` I think we need to also supply the input topic here -- it's named `"input.topic"`

##########
File path: core/src/main/scala/kafka/tools/StreamsResetter.java
##########
@@ -168,7 +168,7 @@ public int run(final String[] args,
             consumerConfig.putAll(properties);
             exitCode = maybeResetInputAndSeekToEndIntermediateTopicOffsets(consumerConfig, dryRun);
             maybeDeleteInternalTopics(adminClient, dryRun);
-
+            System.out.println("succeed to reset stream application: " + groupId);

Review comment:
       super nit: we already print `Done` upon successful reset of the application (at the end of `#maybeResetInputAndSeekToEndIntermediateTopicOffsets`), can we just reuse that in the system test so we don't have to add any new print statements?




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org