void verify()
    {
        final MutableInteger messageCount = new MutableInteger();
        final ExpandableRingBuffer.MessageConsumer messageConsumer =
            (buffer, offset, length, headOffset) ->
            {
                messageCount.increment();

                final int headerOffset = offset + MessageHeaderDecoder.ENCODED_LENGTH;
                final int clusterSessionIdOffset =
                    headerOffset + SessionMessageHeaderDecoder.clusterSessionIdEncodingOffset();

                final long clusterSessionId = buffer.getLong(
                    clusterSessionIdOffset, SessionMessageHeaderDecoder.BYTE_ORDER);

                if (clusterSessionId != (logServiceSessionId + messageCount.get()))
                {
                    throw new ClusterException("snapshot has incorrect pending message:" +
                        " serviceId=" + serviceId +
                        " nextServiceSessionId=" + nextServiceSessionId +
                        " logServiceSessionId=" + logServiceSessionId +
                        " clusterSessionId=" + clusterSessionId +
                        " pendingMessageIndex=" + messageCount.get(),
                        AeronException.Category.FATAL);
                }

                return true;
            };

        pendingMessages.forEach(messageConsumer, Integer.MAX_VALUE);

        if (nextServiceSessionId != (logServiceSessionId + messageCount.get() + 1))
        {
            throw new ClusterException("snapshot has incorrect pending message state:" +
                " serviceId=" + serviceId +
                " nextServiceSessionId=" + nextServiceSessionId +
                " logServiceSessionId=" + logServiceSessionId +
                " pendingMessageCount=" + messageCount.get(),
                AeronException.Category.FATAL);
        }
    }
@Test
    void snapshotEmpty()
    {
        final PendingServiceMessageTracker tracker = new PendingServiceMessageTracker(
            0, counter, logPublisher, clusterClock);

        final Snapshot snapshot = takeSnapshot(tracker);

        final PendingServiceMessageTracker trackerLoaded = new PendingServiceMessageTracker(
            0, counter, logPublisher, clusterClock);

        loadSnapshot(trackerLoaded, snapshot);
        trackerLoaded.verify();
    }