@Override
  public List<QueuedCommand> getNewCommands(final Duration timeout) {
    completeSatisfiedSequenceNumberFutures();

    final List<QueuedCommand> commands = Lists.newArrayList();

    final Iterable<ConsumerRecord<byte[], byte[]>> records = commandTopic.getNewCommands(timeout);
    for (ConsumerRecord<byte[], byte[]> record: records) {
      if (record.value() != null) {
        Optional<CommandStatusFuture> commandStatusFuture = Optional.empty();
        try {
          final CommandId commandId =
              commandIdDeserializer.deserialize(commandTopicName, record.key());
          commandStatusFuture = Optional.ofNullable(commandStatusMap.remove(commandId));
        } catch (Exception e) {
          LOG.warn(
              "Error while attempting to fetch from commandStatusMap for key {}",
              record.key(),
              e);
        }
        commands.add(new QueuedCommand(
            record.key(),
            record.value(),
            commandStatusFuture,
            record.offset()));
      }
    }

    return commands;
  }
@Test
  public void shouldCompleteFuturesWhenGettingNewCommands() {
    // Given:
    when(commandTopic.getCommandTopicConsumerPosition()).thenReturn(22L);

    // When:
    commandStore.getNewCommands(NEW_CMDS_TIMEOUT);

    // Then:
    final InOrder inOrder = inOrder(sequenceNumberFutureStore, commandTopic);
    inOrder.verify(sequenceNumberFutureStore)
        .completeFuturesUpToAndIncludingSequenceNumber(eq(21L));
    inOrder.verify(commandTopic).getNewCommands(any());
  }