void fetchAndRunCommands() {
    lastPollTime.set(clock.instant());
    final List<QueuedCommand> commands = commandStore.getNewCommands(NEW_CMDS_TIMEOUT);
    if (commands.isEmpty()) {
      if (!commandTopicExists.get()) {
        commandTopicDeleted = true;
      }
      return;
    }

    final List<QueuedCommand> compatibleCommands = checkForIncompatibleCommands(commands);
    final Optional<QueuedCommand> terminateCmd =
        findTerminateCommand(compatibleCommands, commandDeserializer);
    if (terminateCmd.isPresent()) {
      terminateCluster(terminateCmd.get().getAndDeserializeCommand(commandDeserializer));
      return;
    }

    LOG.debug("Found {} new writes to command topic", compatibleCommands.size());
    for (final QueuedCommand command : compatibleCommands) {
      if (closed) {
        return;
      }

      executeStatement(command);
    }
  }
@Test
  public void shouldPullAndRunStatements() {
    // Given:
    givenQueuedCommands(queuedCommand1, queuedCommand2, queuedCommand3);

    // When:
    commandRunner.fetchAndRunCommands();

    // Then:
    final InOrder inOrder = inOrder(statementExecutor);
    inOrder.verify(statementExecutor).handleStatement(queuedCommand1);
    inOrder.verify(statementExecutor).handleStatement(queuedCommand2);
    inOrder.verify(statementExecutor).handleStatement(queuedCommand3);
  }