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

    // When
    commandTopic.getRestoreCommands(Duration.ofHours(1));

    // Then
    final InOrder inOrder = Mockito.inOrder(commandTopicBackup);
    inOrder.verify(commandTopicBackup, times(1)).writeRecord(record1);
    inOrder.verify(commandTopicBackup, times(1)).writeRecord(record2);
  }