### Question:
CommandStore implements Closeable { RestoreCommands getRestoreCommands() { final RestoreCommands restoreCommands = new RestoreCommands(); Collection<TopicPartition> commandTopicPartitions = getTopicPartitionsForTopic(commandTopic); commandConsumer.seekToBeginning(commandTopicPartitions); log.debug("Reading prior command records"); final Map<CommandId, ConsumerRecord<CommandId, Command>> commands = new LinkedHashMap<>(); ConsumerRecords<CommandId, Command> records = commandConsumer.poll(POLLING_TIMEOUT_FOR_COMMAND_TOPIC); while (!records.isEmpty()) { log.debug("Received {} records from poll", records.count()); for (ConsumerRecord<CommandId, Command> record : records) { restoreCommands.addCommand(record.key(), record.value()); } records = commandConsumer.poll(POLLING_TIMEOUT_FOR_COMMAND_TOPIC); } log.debug("Retrieved records:" + commands.size()); return restoreCommands; } CommandStore(
      String commandTopic,
      Consumer<CommandId, Command> commandConsumer,
      Producer<CommandId, Command> commandProducer,
      CommandIdAssigner commandIdAssigner
  ); @Override void close(); CommandId distributeStatement(
      String statementString,
      Statement statement,
      Map<String, Object> streamsProperties
  ); ConsumerRecords<CommandId, Command> getNewCommands();  }### Answer:
@Test public void shouldCollectTerminatedQueries() { final CommandId terminated = new CommandId(CommandId.Type.TERMINATE, "queryId", CommandId.Action.EXECUTE); final ConsumerRecords<CommandId, Command> records = new ConsumerRecords<>( Collections.singletonMap(new TopicPartition("topic", 0), Collections.singletonList( new ConsumerRecord<>("topic", 0, 0, terminated, new Command("terminate query 'queryId'", Collections.emptyMap())) ))); EasyMock.expect(commandConsumer.partitionsFor(COMMAND_TOPIC)).andReturn(Collections.emptyList()); EasyMock.expect(commandConsumer.poll(anyLong())).andReturn(records) .andReturn(new ConsumerRecords<>(Collections.emptyMap())); EasyMock.replay(commandConsumer); final CommandStore commandStore = new CommandStore(COMMAND_TOPIC, commandConsumer, commandProducer, new CommandIdAssigner(new MetaStoreImpl())); final RestoreCommands restoreCommands = commandStore.getRestoreCommands(); assertThat(restoreCommands.terminatedQueries(), equalTo(Collections.singletonMap(new QueryId("queryId"), terminated))); }