### Question:
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); } } } CommandRunner(
      StatementExecutor statementExecutor,
      CommandStore commandStore
  ); @Override void run(); @Override void close(); void processPriorCommands();  }### Answer:
@Test public void shouldFetchAndRunNewCommandsFromCommandTopic() throws Exception { StatementExecutor statementExecutor = mock(StatementExecutor.class); statementExecutor.handleStatement(anyObject(), anyObject()); expectLastCall().times(4); replay(statementExecutor); CommandStore commandStore = mock(CommandStore.class); expect(commandStore.getNewCommands()).andReturn(new ConsumerRecords<>(getRecordMap())); replay(commandStore); CommandRunner commandRunner = new CommandRunner(statementExecutor, commandStore); commandRunner.fetchAndRunCommands(); verify(statementExecutor); }