public List<QueuedCommand> getRestoreCommands(final Duration duration) {
    if (commandTopicBackup.commandTopicCorruption()) {
      log.warn("Corruption detected. "
          + "Use backup to restore command topic.");
      return Collections.emptyList();
    }
    return getAllCommandsInCommandTopic(
        commandConsumer,
        commandTopicPartition,
        Optional.of(commandTopicBackup),
        duration
    );
  }
@Test
  public void shouldGetRestoreCommandsCorrectlyOnPollTimeout() {
    // Given:
    when(commandConsumer.poll(any(Duration.class)))
        .thenReturn(ConsumerRecords.empty())
        .thenReturn(someConsumerRecords(
            new ConsumerRecord<>("topic", 0, 0, commandId1, command1),
            new ConsumerRecord<>("topic", 0, 1, commandId2, command2)));
    when(commandConsumer.endOffsets(any())).thenReturn(ImmutableMap.of(TOPIC_PARTITION, 2L));
    when(commandConsumer.position(TOPIC_PARTITION)).thenReturn(0L, 0L, 2L);

    // When:
    final List<QueuedCommand> queuedCommandList = commandTopic
        .getRestoreCommands(Duration.ofMillis(1));

    // Then:
    assertThat(queuedCommandList, equalTo(ImmutableList.of(
        new QueuedCommand(commandId1, command1, Optional.empty(), 0L),
        new QueuedCommand(commandId2, command2, Optional.empty(), 1L))));
  }