class 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;
  }

  public  CommandStore(
      String commandTopic,
      Consumer<CommandId, Command> commandConsumer,
      Producer<CommandId, Command> commandProducer,
      CommandIdAssigner commandIdAssigner
  );

}