void verifyCheckSum(Consumer<String> println) {
            LOG.debug("Begin verification");
            maxLag = -1L;

            for (Map.Entry<TopicPartition, Map<Integer, FetchResponseData.PartitionData>> cacheEntry : recordsCache.entrySet()) {
                TopicPartition topicPartition = cacheEntry.getKey();
                Map<Integer, FetchResponseData.PartitionData> fetchResponsePerReplica = cacheEntry.getValue();

                LOG.debug("Verifying {}", topicPartition);
                assert fetchResponsePerReplica.size() == expectedReplicasPerTopicPartition.get(topicPartition) :
                    "fetched " + fetchResponsePerReplica.size() + " replicas for " + topicPartition +
                        ", but expected " + expectedReplicasPerTopicPartition.get(topicPartition) + " replicas";

                Map<Integer, Iterator<? extends RecordBatch>> recordBatchIteratorMap = new HashMap<>();
                for (Map.Entry<Integer, FetchResponseData.PartitionData> fetchResEntry : fetchResponsePerReplica.entrySet()) {
                    int replicaId = fetchResEntry.getKey();
                    FetchResponseData.PartitionData fetchResponse = fetchResEntry.getValue();
                    Iterator<? extends RecordBatch> recordIterator =
                        FetchResponse.recordsOrFail(fetchResponse).batches().iterator();
                    recordBatchIteratorMap.put(replicaId, recordIterator);
                }

                long maxHw = fetchResponsePerReplica.values().stream()
                    .mapToLong(FetchResponseData.PartitionData::highWatermark)
                    .max().orElse(-1L);

                boolean isMessageInAllReplicas = true;

                // iterate one message at a time from every replica, until high watermark is reached
                while (isMessageInAllReplicas) {
                    Optional<MessageInfo> messageInfoFromFirstReplicaOpt = Optional.empty();

                    for (Map.Entry<Integer, Iterator<? extends RecordBatch>> batchEntry : recordBatchIteratorMap.entrySet()) {
                        int replicaId = batchEntry.getKey();
                        Iterator<? extends RecordBatch> recordBatchIterator = batchEntry.getValue();

                        try {
                            if (recordBatchIterator.hasNext()) {
                                RecordBatch batch = recordBatchIterator.next();

                                // only verify up to the high watermark
                                if (batch.lastOffset() >= fetchResponsePerReplica.get(replicaId).highWatermark()) {
                                    isMessageInAllReplicas = false;
                                } else {
                                    if (!messageInfoFromFirstReplicaOpt.isPresent()) {
                                        messageInfoFromFirstReplicaOpt = Optional.of(
                                            new MessageInfo(replicaId, batch.lastOffset(), batch.nextOffset(), batch.checksum())
                                        );
                                    } else {
                                        MessageInfo messageInfoFromFirstReplica = messageInfoFromFirstReplicaOpt.get();

                                        if (messageInfoFromFirstReplica.offset != batch.lastOffset()) {
                                            synchronized (DATE_FORMAT) {
                                                println.accept(DATE_FORMAT.format(new Date(Time.SYSTEM.milliseconds())) +
                                                    ": partition " + topicPartition +
                                                    ": replica " + messageInfoFromFirstReplica.replicaId +
                                                    "'s offset " + messageInfoFromFirstReplica.offset +
                                                    " doesn't match replica " + replicaId +
                                                    "'s offset " + batch.lastOffset());
                                                Exit.exit(1);
                                            }
                                        }

                                        if (messageInfoFromFirstReplica.checksum != batch.checksum())
                                            synchronized (DATE_FORMAT) {
                                                println.accept(DATE_FORMAT.format(new Date(Time.SYSTEM.milliseconds())) +
                                                    ": partition " + topicPartition +
                                                    " has unmatched checksum at offset " + batch.lastOffset() +
                                                    "; replica " + messageInfoFromFirstReplica.replicaId +
                                                    "'s checksum " + messageInfoFromFirstReplica.checksum +
                                                    "; replica " + replicaId + "'s checksum " + batch.checksum());
                                            }
                                    }
                                }
                            } else {
                                isMessageInAllReplicas = false;
                            }
                        } catch (Throwable t) {
                            throw new RuntimeException("Error in processing replica " + replicaId +
                                " in partition " + topicPartition + " at offset " +
                                fetchOffsetMap.get(topicPartition), t);
                        }
                    }

                    if (isMessageInAllReplicas) {
                        long nextOffset = messageInfoFromFirstReplicaOpt.map(messageInfo -> messageInfo.nextOffset).orElse(-1L);
                        fetchOffsetMap.put(topicPartition, nextOffset);
                        LOG.debug("{} replicas match at offset {} for {}",
                            expectedReplicasPerTopicPartition.get(topicPartition), nextOffset, topicPartition);
                    }
                }

                if (maxHw - fetchOffsetMap.get(topicPartition) > maxLag) {
                    offsetWithMaxLag = fetchOffsetMap.get(topicPartition);
                    maxLag = maxHw - offsetWithMaxLag;
                    maxLagTopicAndPartition = topicPartition;
                }

                fetchResponsePerReplica.clear();
            }

            long currentTimeMs = Time.SYSTEM.milliseconds();
            if (currentTimeMs - lastReportTime > reportInterval) {
                synchronized (DATE_FORMAT) {
                    println.accept(DATE_FORMAT.format(new Date(currentTimeMs)) +
                        ": max lag is " + maxLag + " for partition " +
                        maxLagTopicAndPartition + " at offset " + offsetWithMaxLag +
                        " among " + recordsCache.size() + " partitions");
                }
                lastReportTime = currentTimeMs;
            }
        }
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