chia7712 commented on code in PR #16019:
URL: https://github.com/apache/kafka/pull/16019#discussion_r1685752276
##########
tools/src/test/java/org/apache/kafka/tools/consumer/ConsoleConsumerTest.java:
##########
@@ -239,4 +284,73 @@ public void shouldWorkWithoutTopicOption() throws
IOException {
verify(mockConsumer).subscribe(any(Pattern.class));
consumer.cleanup();
}
+
+ @ClusterTest(brokers = 3)
+ public void testTransactionLogMessageFormatter(ClusterInstance cluster)
throws Exception {
+ try (Admin admin = cluster.createAdminClient()) {
+
+ NewTopic newTopic = new NewTopic(topic, 1, (short) 1);
+ admin.createTopics(singleton(newTopic));
+ produceMessages(cluster);
+
+ String[] transactionLogMessageFormatter = new String[]{
+ "--bootstrap-server", cluster.bootstrapServers(),
+ "--topic", Topic.TRANSACTION_STATE_TOPIC_NAME,
+ "--formatter",
"org.apache.kafka.tools.consumer.TransactionLogMessageFormatter",
+ "--from-beginning"
+ };
+
+ ConsoleConsumerOptions options = new
ConsoleConsumerOptions(transactionLogMessageFormatter);
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ PrintStream output = new PrintStream(out);
+ ConsoleConsumer.process(1, options.formatter(),
Review Comment:
please call `cleanup` to close consumer ... otherwise, @FrankYang0529 will
hate your PR due to thread leak
--
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.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
