@Override
                public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                    remainingMessages.set(partitions.stream()
                        .mapToLong(partition -> messagesRemaining(consumer, partition)).sum());
                    numMessagesProcessedSinceLastRebalance.set(0);
                    // We use message cap for remaining here as the remainingMessages are not set yet.
                    System.out.println(statusAsJson(
                        "RebalanceComplete",
                        totalMessageProcessed.get(),
                        numMessagesProcessedSinceLastRebalance.get(),
                        remainingMessages.get(),
                        transactionalId
                    ));
                }
Get the command-line argument parser.
tools/src/main/java/org/apache/kafka/tools/TransactionalMessageCopier.java