public boolean execute(final File clusterDir)
    {
        if (!clusterDir.exists() || !clusterDir.isDirectory())
        {
            throw new IllegalArgumentException("invalid cluster directory: " + clusterDir.getAbsolutePath());
        }

        final RecordingLog.Entry entry = ClusterTool.findLatestValidSnapshot(clusterDir);
        if (null == entry)
        {
            throw new ClusterException("no valid snapshot found");
        }

        final long recordingId = entry.recordingId;
        final ClusterNodeControlProperties properties = ClusterTool.loadControlProperties(clusterDir);
        final RecordingSignalCapture recordingSignalCapture = new RecordingSignalCapture();
        try (Aeron aeron = Aeron.connect(new Aeron.Context().aeronDirectoryName(properties.aeronDirectoryName));
            AeronArchive archive = AeronArchive.connect(new AeronArchive.Context()
                .controlRequestChannel(archiveLocalRequestChannel)
                .controlRequestStreamId(archiveLocalRequestStreamId)
                .controlResponseChannel(IPC_CHANNEL)
                .recordingSignalConsumer(recordingSignalCapture)
                .aeron(aeron)))
        {
            final SnapshotReader snapshotReader = new SnapshotReader();
            replayLocalSnapshotRecording(aeron, archive, recordingId, snapshotReader);

            final long targetNextServiceSessionId = max(
                max(snapshotReader.nextServiceSessionId, snapshotReader.maxClusterSessionId + 1),
                snapshotReader.logServiceSessionId + 1 + snapshotReader.pendingServiceMessageCount);
            final long targetLogServiceSessionId =
                targetNextServiceSessionId - 1 - snapshotReader.pendingServiceMessageCount;

            if (targetNextServiceSessionId != snapshotReader.nextServiceSessionId ||
                targetLogServiceSessionId != snapshotReader.logServiceSessionId ||
                0 != snapshotReader.pendingServiceMessageCount &&
                (targetLogServiceSessionId + 1 != snapshotReader.minClusterSessionId ||
                targetNextServiceSessionId - 1 != snapshotReader.maxClusterSessionId))
            {
                final long tempRecordingId = createNewSnapshotRecording(
                    aeron, archive, recordingId, targetLogServiceSessionId, targetNextServiceSessionId);

                final long stopPosition = awaitRecordingStopPosition(archive, recordingId);
                final long newStopPosition = awaitRecordingStopPosition(archive, tempRecordingId);
                if (stopPosition != newStopPosition)
                {
                    throw new ClusterException("new snapshot recording incomplete: expectedStopPosition=" +
                        stopPosition + ", actualStopPosition=" + newStopPosition);
                }

                recordingSignalCapture.reset();
                archive.truncateRecording(recordingId, 0);
                recordingSignalCapture.awaitSignalForRecordingId(archive, recordingId, RecordingSignal.DELETE);

                final long replicationId = archive.replicate(
                    tempRecordingId, recordingId, archive.context().controlRequestStreamId(), IPC_CHANNEL, null);

                recordingSignalCapture.reset();
                recordingSignalCapture.awaitSignalForCorrelationId(archive, replicationId, RecordingSignal.SYNC);

                final long replicatedStopPosition = recordingSignalCapture.position();
                if (stopPosition != replicatedStopPosition)
                {
                    throw new ClusterException("incomplete replication of the new recording: expectedStopPosition=" +
                        stopPosition + ", replicatedStopPosition=" + replicatedStopPosition);
                }

                recordingSignalCapture.reset();
                archive.purgeRecording(tempRecordingId);
                recordingSignalCapture.awaitSignalForRecordingId(archive, tempRecordingId, RecordingSignal.DELETE);

                return true;
            }
        }

        return false;
    }
@Test
    @SlowTest
    @InterruptAfter(60)
    void executeIsANoOpIfTheSnapshotIsValid()
    {
        final IntFunction<TestNode.TestService[]> servicesSupplier =
            (i) -> new TestNode.TestService[]
            {
                new TestNode.MessageTrackingService(1, i),
                new TestNode.MessageTrackingService(2, i)
            };
        final TestCluster cluster = aCluster()
            .withStaticNodes(3)
            .withTimerServiceSupplier(new PriorityHeapTimerServiceSupplier())
            .withServiceSupplier(servicesSupplier)
            .start();
        systemTestWatcher.cluster(cluster);
        final int serviceCount = cluster.node(0).services().length;

        final TestNode leader = cluster.awaitLeader();
        cluster.connectClient();

        TestNode.MessageTrackingService.delaySessionMessageProcessing(true);
        int messageCount = 0;
        final ExpandableArrayBuffer msgBuffer = cluster.msgBuffer();
        for (int i = 0; i < NUM_MESSAGES; i++)
        {
            msgBuffer.putInt(0, ++messageCount, LITTLE_ENDIAN);
            cluster.pollUntilMessageSent(SIZE_OF_INT);
        }

        cluster.takeSnapshot(leader);
        cluster.awaitSnapshotCount(1);
        TestNode.MessageTrackingService.delaySessionMessageProcessing(false);

        cluster.awaitResponseMessageCount(messageCount * serviceCount);
        awaitServiceMessages(cluster, serviceCount, messageCount);
        stopConsensusModulesAndServices(cluster);

        final File leaderClusterDir = leader.consensusModule().context().clusterDir();
        final RecordingLog.Entry leaderSnapshot = ClusterTool.findLatestValidSnapshot(leaderClusterDir);
        assertNotNull(leaderSnapshot);

        final MutableLong mutableNextSessionId = new MutableLong(NULL_SESSION_ID);
        final MutableLong mutableNextServiceSessionId = new MutableLong(NULL_SESSION_ID);
        final MutableLong mutableLogServiceSessionId = new MutableLong(NULL_SESSION_ID);
        final LongArrayList pendingMessageClusterSessionIds = new LongArrayList();
        final ConsensusModuleSnapshotListener stateReader = new NoOpConsensusModuleSnapshotListener()
            {
                public void onLoadConsensusModuleState(
                    final long nextSessionId,
                    final long nextServiceSessionId,
                    final long logServiceSessionId,
                    final int pendingMessageCapacity,
                    final DirectBuffer buffer,
                    final int offset,
                    final int length)
                {
                    mutableNextSessionId.set(nextSessionId);
                    mutableNextServiceSessionId.set(nextServiceSessionId);
                    mutableLogServiceSessionId.set(logServiceSessionId);
                }

                public void onLoadPendingMessage(
                    final long clusterSessionId, final DirectBuffer buffer, final int offset, final int length)
                {
                    pendingMessageClusterSessionIds.add(clusterSessionId);
                }
            };

        readSnapshotRecording(leader, leaderSnapshot.recordingId, stateReader);

        final long beforeNextSessionId = mutableNextSessionId.get();
        final long beforeNextServiceSessionId = mutableNextServiceSessionId.get();
        final long beforeLogServiceSessionId = mutableLogServiceSessionId.get();
        final long[] beforeClusterSessionIds = pendingMessageClusterSessionIds.toLongArray();
        assertNotEquals(NULL_SESSION_ID, beforeNextSessionId);
        assertNotEquals(NULL_SESSION_ID, beforeNextServiceSessionId);
        assertNotEquals(NULL_SESSION_ID, beforeLogServiceSessionId);
        assertNotEquals(beforeNextSessionId, beforeNextServiceSessionId);
        assertNotEquals(beforeNextSessionId, beforeLogServiceSessionId);
        assertNotEquals(beforeNextServiceSessionId, beforeLogServiceSessionId);
        assertNotEquals(0, beforeClusterSessionIds.length);

        final ConsensusModuleSnapshotPendingServiceMessagesPatch snapshotPatch =
            new ConsensusModuleSnapshotPendingServiceMessagesPatch(
            leader.archive().context().localControlChannel(),
            leader.archive().context().localControlStreamId());
        assertFalse(snapshotPatch.execute(leaderClusterDir));

        mutableNextSessionId.set(NULL_SESSION_ID);
        mutableNextServiceSessionId.set(NULL_SESSION_ID);
        mutableLogServiceSessionId.set(NULL_SESSION_ID);

        pendingMessageClusterSessionIds.clear();
        readSnapshotRecording(leader, leaderSnapshot.recordingId, stateReader);

        assertEquals(beforeNextSessionId, mutableNextSessionId.get());
        assertEquals(beforeNextServiceSessionId, mutableNextServiceSessionId.get());
        assertEquals(beforeLogServiceSessionId, mutableLogServiceSessionId.get());
        assertArrayEquals(beforeClusterSessionIds, pendingMessageClusterSessionIds.toLongArray());
    }