public static void main(String[] args) {
        try {
            LOG.warn("This tool is deprecated and may be removed in a future major release.");

            ReplicaVerificationToolOptions options = new ReplicaVerificationToolOptions(args);
            // getting topic metadata
            LOG.info("Getting topic metadata...");
            String brokerList = options.brokerHostsAndPorts();

            try (Admin adminClient = createAdminClient(brokerList)) {
                Collection<TopicDescription> topicsMetadata = listTopicsMetadata(adminClient);
                Map<Integer, Node> brokerInfo = brokerDetails(adminClient);

                Map<String, Uuid> topicIds = topicsMetadata.stream().collect(Collectors.toMap(TopicDescription::name, TopicDescription::topicId));

                List<TopicDescription> filteredTopicMetadata = topicsMetadata.stream().filter(
                    topicMetadata -> options.topicsIncludeFilter().isTopicAllowed(topicMetadata.name(), false)
                ).collect(Collectors.toList());

                if (filteredTopicMetadata.isEmpty()) {
                    LOG.error("No topics found. {} if specified, is either filtering out all topics or there is no topic.", options.topicsIncludeOpt);
                    Exit.exit(1);
                }

                List<TopicPartitionReplica> topicPartitionReplicas = filteredTopicMetadata.stream().flatMap(
                    topicMetadata -> topicMetadata.partitions().stream().flatMap(
                        partitionMetadata -> partitionMetadata.replicas().stream().map(
                            node -> new TopicPartitionReplica(topicMetadata.name(), partitionMetadata.partition(), node.id())
                        )
                    )
                ).collect(Collectors.toList());
                LOG.debug("Selected topic partitions: {}", topicPartitionReplicas);

                Map<Integer, List<TopicPartition>> brokerToTopicPartitions = topicPartitionReplicas.stream()
                    .collect(Collectors.groupingBy(
                        TopicPartitionReplica::brokerId,
                        Collectors.mapping(
                            replica -> new TopicPartition(replica.topic(), replica.partition()),
                            Collectors.toList()
                        )
                    ));
                LOG.debug("Topic partitions per broker: {}", brokerToTopicPartitions);

                Map<TopicPartition, Integer> expectedReplicasPerTopicPartition = topicPartitionReplicas.stream()
                    .collect(Collectors.groupingBy(
                        replica -> new TopicPartition(replica.topic(), replica.partition()),
                        Collectors.collectingAndThen(
                            Collectors.toList(),
                            List::size
                        )
                    ));
                LOG.debug("Expected replicas per topic partition: {}", expectedReplicasPerTopicPartition);

                List<TopicPartition> topicPartitions = filteredTopicMetadata.stream()
                    .flatMap(topicMetadata -> topicMetadata.partitions().stream()
                        .map(partitionMetadata -> new TopicPartition(topicMetadata.name(), partitionMetadata.partition()))
                    )
                    .collect(Collectors.toList());

                Properties consumerProps = consumerConfig(brokerList);

                ReplicaBuffer replicaBuffer = new ReplicaBuffer(expectedReplicasPerTopicPartition,
                    initialOffsets(topicPartitions, consumerProps, options.initialOffsetTime()),
                    brokerToTopicPartitions.size(), options.reportInterval());

                // create all replica fetcher threads
                int verificationBrokerId = brokerToTopicPartitions.entrySet().iterator().next().getKey();
                AtomicInteger counter = new AtomicInteger(0);
                List<ReplicaFetcher> fetcherThreads = brokerToTopicPartitions.entrySet().stream()
                    .map(entry -> {
                        int brokerId = entry.getKey();
                        Iterable<TopicPartition> partitions = entry.getValue();
                        return new ReplicaFetcher(
                            "ReplicaFetcher-" + brokerId,
                            brokerInfo.get(brokerId),
                            partitions,
                            topicIds,
                            replicaBuffer,
                            options.fetchSize(),
                            options.maxWaitMs(),
                            1,
                            brokerId == verificationBrokerId,
                            consumerProps,
                            counter.incrementAndGet()
                        );
                    })
                    .collect(Collectors.toList());

                Runtime.getRuntime().addShutdownHook(new Thread(() -> {
                    LOG.info("Stopping all fetchers");
                    fetcherThreads.forEach(replicaFetcher -> {
                        try {
                            replicaFetcher.shutdown();
                        } catch (InterruptedException ignored) {
                        }
                    });
                }, "ReplicaVerificationToolShutdownHook"));

                fetcherThreads.forEach(Thread::start);
                synchronized (DATE_FORMAT) {
                    System.out.printf("%s: verification process is started%n",
                        DATE_FORMAT.format(new Date(Time.SYSTEM.milliseconds())));
                }
            }
        } catch (Throwable e) {
            System.err.println(e.getMessage());
            System.err.println(Utils.stackTrace(e));
            Exit.exit(1);
        }
    }
For verifying the consistency among replicas.
<p>
 1. start a fetcher on every broker
 2. each fetcher does the following
   2.1 issues fetch request
   2.2 puts the fetched result in a shared buffer
   2.3 waits for all other fetchers to finish step 2.2
   2.4 one of the fetchers verifies the consistency of fetched results among replicas
<p>
The consistency verification is up to the high watermark. The tool reports the
max lag between the verified offset and the high watermark among all partitions.
<p>
If a broker goes down, the verification of the partitions on that broker is delayed
until the broker is up again.
<p>
Caveats:
1. The tool needs all brokers to be up at startup time.
2. The tool doesn't handle out of range offsets.
tools/src/main/java/org/apache/kafka/tools/ReplicaVerificationTool.java