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 shouldGetRestoreCommandsCorrectly() {
    // Given:
    when(commandConsumer.poll(any(Duration.class)))
        .thenReturn(someConsumerRecords(
            record1,
            record2))
        .thenReturn(someConsumerRecords(
            record3));
    when(commandConsumer.endOffsets(any())).thenReturn(ImmutableMap.of(TOPIC_PARTITION, 3L));
    when(commandConsumer.position(TOPIC_PARTITION)).thenReturn(0L, 2L, 3L);

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

    // Then:
    verify(commandConsumer).seekToBeginning(topicPartitionsCaptor.capture());
    assertThat(topicPartitionsCaptor.getValue(),
        equalTo(Collections.singletonList(new TopicPartition(COMMAND_TOPIC_NAME, 0))));
    assertThat(queuedCommandList, equalTo(ImmutableList.of(
        new QueuedCommand(commandId1, command1, Optional.empty(), 0L),
        new QueuedCommand(commandId2, command2, Optional.empty(),1L),
        new QueuedCommand(commandId3, command3, Optional.empty(), 2L))));
  }