@Override
        public void verify() {
            for (Map.Entry<Integer, PersistentState> nodeEntry : cluster.nodes.entrySet()) {
                int nodeId = nodeEntry.getKey();
                ReplicatedLog log = nodeEntry.getValue().log;
                log.earliestSnapshotId().ifPresent(earliestSnapshotId  -> {
                    long logStartOffset = log.startOffset();
                    ValidOffsetAndEpoch validateOffsetAndEpoch = log.validateOffsetAndEpoch(
                        earliestSnapshotId.offset(),
                        earliestSnapshotId.epoch()
                    );

                    assertTrue(
                        logStartOffset <= earliestSnapshotId.offset(),
                        () -> String.format(
                            "invalid log start offset (%s) and snapshotId offset (%s): nodeId = %s",
                            logStartOffset,
                            earliestSnapshotId.offset(),
                            nodeId
                        )
                    );
                    assertEquals(
                        ValidOffsetAndEpoch.valid(earliestSnapshotId),
                        validateOffsetAndEpoch,
                        () -> String.format("invalid leader epoch cache: nodeId = %s", nodeId)
                    );

                    if (logStartOffset > 0) {
                        assertEquals(
                            logStartOffset,
                            earliestSnapshotId.offset(),
                            () -> String.format("missing snapshot at log start offset: nodeId = %s", nodeId)
                        );
                    }
                });
            }
        }
Returns if the message should be sent to the destination.

Returns false when outbound request messages contains a destination {@code Node} that
matches the set of unreachable {@code InetSocketAddress}. Note that the {@code Node.id()}
and {@code Node.rack()} are not compared.

@param message the raft message
@return true if the message should be delivered, otherwise false
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java