public void processPriorCommands(final PersistentQueryCleanupImpl queryCleanup) {
    try {
      final List<QueuedCommand> restoreCommands = commandStore.getRestoreCommands();
      final List<QueuedCommand> compatibleCommands = checkForIncompatibleCommands(restoreCommands);

      LOG.info("Restoring previous state from {} commands.", compatibleCommands.size());

      final Optional<QueuedCommand> terminateCmd =
          findTerminateCommand(compatibleCommands, commandDeserializer);
      if (terminateCmd.isPresent()) {
        LOG.info("Cluster previously terminated: terminating.");
        terminateCluster(terminateCmd.get().getAndDeserializeCommand(commandDeserializer));
        return;
      }

      final List<QueuedCommand> compacted = compactor.apply(compatibleCommands);

      compacted.forEach(
          command -> {
            currentCommandRef.set(new Pair<>(command, clock.instant()));
            RetryUtil.retryWithBackoff(
                maxRetries,
                STATEMENT_RETRY_MS,
                MAX_STATEMENT_RETRY_MS,
                () -> statementExecutor.handleRestore(command),
                WakeupException.class
            );
            currentCommandRef.set(null);
          }
      );

      final List<PersistentQueryMetadata> queries = statementExecutor
          .getKsqlEngine()
          .getPersistentQueries();

      if (commandStore.corruptionDetected()) {
        LOG.info("Corruption detected, queries will not be started.");
        queries.forEach(QueryMetadata::setCorruptionQueryError);
      } else {
        LOG.info("Restarting {} queries.", queries.size());
        queries.forEach(PersistentQueryMetadata::start);
        queryCleanup.cleanupLeakedQueries(queries);
        //We only want to clean up if the queries are read properly
        //We do not want to clean up potentially important stuff
        //when the cluster is in a bad state
      }

      LOG.info("Restore complete");

    } catch (final Exception e) {
      LOG.error("Error during restore", e);
      throw e;
    }
  }
@Test
  public void shouldEarlyOutIfRestoreContainsTerminate() {
    // Given:
    givenQueuedCommands(queuedCommand1, queuedCommand2, queuedCommand3);
    when(queuedCommand2.getAndDeserializeCommand(commandDeserializer)).thenReturn(clusterTerminate);

    // When:
    commandRunner.processPriorCommands(persistentQueryCleanupImpl);

    // Then:
    verify(statementExecutor, never()).handleRestore(any());
  }