@Override
        public void doWork() {
            CountDownLatch fetcherBarrier = replicaBuffer.getFetcherBarrier();
            CountDownLatch verificationBarrier = replicaBuffer.getVerificationBarrier();

            Map<TopicPartition, FetchRequest.PartitionData> requestMap = new LinkedHashMap<>();
            for (TopicPartition topicPartition : topicPartitions) {
                requestMap.put(topicPartition, new FetchRequest.PartitionData(
                    topicIds.getOrDefault(topicPartition.topic(), Uuid.ZERO_UUID),
                    replicaBuffer.getOffset(topicPartition),
                    0L,
                    fetchSize,
                    Optional.empty()
                ));
            }

            FetchRequest.Builder fetchRequestBuilder = FetchRequest.Builder.forReplica(
                ApiKeys.FETCH.latestVersion(),
                FetchRequest.DEBUGGING_CONSUMER_ID,
                -1,
                maxWait,
                minBytes,
                requestMap
            );

            LOG.debug("Issuing fetch request");

            FetchResponse fetchResponse = null;
            try {
                ClientResponse clientResponse = fetchEndpoint.sendRequest(fetchRequestBuilder);
                fetchResponse = (FetchResponse) clientResponse.responseBody();
            } catch (Throwable t) {
                if (!isRunning())
                    throw new RuntimeException(t);
            }

            if (fetchResponse != null) {
                fetchResponse.responseData(topicNames, ApiKeys.FETCH.latestVersion()).forEach((tp, partitionData) ->
                    replicaBuffer.addFetchedData(tp, sourceBroker.id(), partitionData));
            } else {
                for (TopicPartition topicAndPartition : topicPartitions) {
                    replicaBuffer.addFetchedData(
                        topicAndPartition,
                        sourceBroker.id(),
                        FetchResponse.partitionResponse(topicAndPartition.partition(), Errors.NONE)
                    );
                }
            }

            fetcherBarrier.countDown();
            LOG.debug("Done fetching");

            // wait for all fetchers to finish
            try {
                fetcherBarrier.await();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return;
            }
            LOG.debug("Ready for verification");

            // one of the fetchers will do the verification
            if (doVerification) {
                LOG.debug("Do verification");
                replicaBuffer.verifyCheckSum(System.out::println);
                replicaBuffer.createNewFetcherBarrier();
                replicaBuffer.createNewVerificationBarrier();
                LOG.debug("Created new barrier");
                verificationBarrier.countDown();
            }

            try {
                verificationBarrier.await();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return;
            }
            LOG.debug("Done verification");
        }
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