public static void runEventLoop(Namespace parsedArgs) {
        final String transactionalId = parsedArgs.getString("transactionalId");
        final String outputTopic = parsedArgs.getString("outputTopic");

        String consumerGroup = parsedArgs.getString("consumerGroup");

        final KafkaProducer<String, String> producer = createProducer(parsedArgs);
        final KafkaConsumer<String, String> consumer = createConsumer(parsedArgs);

        final AtomicLong remainingMessages = new AtomicLong(
            parsedArgs.getInt("maxMessages") == -1 ? Long.MAX_VALUE : parsedArgs.getInt("maxMessages"));

        boolean groupMode = parsedArgs.getBoolean("groupMode");
        String topicName = parsedArgs.getString("inputTopic");
        final AtomicLong numMessagesProcessedSinceLastRebalance = new AtomicLong(0);
        final AtomicLong totalMessageProcessed = new AtomicLong(0);
        if (groupMode) {
            consumer.subscribe(Collections.singleton(topicName), new ConsumerRebalanceListener() {
                @Override
                public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                }

                @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
                    ));
                }
            });
        } else {
            TopicPartition inputPartition = new TopicPartition(topicName, parsedArgs.getInt("inputPartition"));
            consumer.assign(singleton(inputPartition));
            remainingMessages.set(Math.min(messagesRemaining(consumer, inputPartition), remainingMessages.get()));
        }

        final boolean enableRandomAborts = parsedArgs.getBoolean("enableRandomAborts");

        producer.initTransactions();

        final AtomicBoolean isShuttingDown = new AtomicBoolean(false);

        Exit.addShutdownHook("transactional-message-copier-shutdown-hook", () -> {
            isShuttingDown.set(true);
            consumer.wakeup();
            System.out.println(statusAsJson(
                "ShutdownComplete",
                totalMessageProcessed.get(),
                numMessagesProcessedSinceLastRebalance.get(),
                remainingMessages.get(),
                transactionalId
            ));
        });

        final boolean useGroupMetadata = parsedArgs.getBoolean("useGroupMetadata");
        try {
            Random random = new Random();
            while (!isShuttingDown.get() && remainingMessages.get() > 0) {
                System.out.println(statusAsJson(
                    "ProcessLoop",
                    totalMessageProcessed.get(),
                    numMessagesProcessedSinceLastRebalance.get(),
                    remainingMessages.get(),
                    transactionalId
                ));

                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(200));
                if (records.count() > 0) {
                    try {
                        producer.beginTransaction();

                        for (ConsumerRecord<String, String> record : records) {
                            producer.send(producerRecordFromConsumerRecord(outputTopic, record));
                        }

                        long messagesSentWithinCurrentTxn = records.count();

                        ConsumerGroupMetadata groupMetadata = useGroupMetadata ? consumer.groupMetadata() : new ConsumerGroupMetadata(consumerGroup);
                        producer.sendOffsetsToTransaction(consumerPositions(consumer), groupMetadata);

                        if (enableRandomAborts && random.nextInt() % 3 == 0) {
                            abortTransactionAndResetPosition(producer, consumer);
                        } else {
                            producer.commitTransaction();
                            remainingMessages.getAndAdd(-messagesSentWithinCurrentTxn);
                            numMessagesProcessedSinceLastRebalance.getAndAdd(messagesSentWithinCurrentTxn);
                            totalMessageProcessed.getAndAdd(messagesSentWithinCurrentTxn);
                        }
                    } catch (ProducerFencedException e) {
                        throw new KafkaException(String.format("The transactional.id %s has been claimed by another process", transactionalId), e);
                    } catch (KafkaException e) {
                        log.debug("Aborting transaction after catching exception", e);
                        abortTransactionAndResetPosition(producer, consumer);
                    }
                }
            }
        } catch (WakeupException e) {
            if (!isShuttingDown.get()) {
                // Let the exception propagate if the exception was not raised
                // as part of shutdown.
                throw e;
            }
        } finally {
            Utils.closeQuietly(producer, "producer");
            Utils.closeQuietly(consumer, "consumer");
        }
    }
Get the command-line argument parser.
tools/src/main/java/org/apache/kafka/tools/TransactionalMessageCopier.java