  @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);
  }