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:
us...@infra.apache.org
