class CommandRunner implements Runnable, Closeable {

  void fetchAndRunCommands() {
    ConsumerRecords<CommandId, Command> records = commandStore.getNewCommands();
    log.trace("Found {} new writes to command topic", records.count());
    for (ConsumerRecord<CommandId, Command> record : records) {
      CommandId commandId = record.key();
      Command command = record.value();
      if (command != null) {
        executeStatement(command, commandId);
      }
    }
  }

  public  CommandRunner(
      StatementExecutor statementExecutor,
      CommandStore commandStore
  );

}