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
    void executeThrowsIllegalArgumentExceptionIfClusterDirIsNotADirectory(
        final @TempDir File tempDir) throws IOException
    {
        final File clusterDir = new File(tempDir, "file.txt");
        assertTrue(clusterDir.createNewFile());

        final IllegalArgumentException exception = assertThrowsExactly(
            IllegalArgumentException.class,
            () -> new ConsensusModuleSnapshotPendingServiceMessagesPatch().execute(clusterDir));
        assertEquals("invalid cluster directory: " + clusterDir.getAbsolutePath(), exception.getMessage());
    }