public Iterable<ConsumerRecord<byte[], byte[]>> getNewCommands(final Duration timeout) {
    final Iterable<ConsumerRecord<byte[], byte[]>> iterable = commandConsumer.poll(timeout);
    final List<ConsumerRecord<byte[], byte[]>> records = new ArrayList<>();

    if (iterable != null) {
      for (ConsumerRecord<byte[], byte[]> record : iterable) {
        try {
          backupRecord(Optional.of(commandTopicBackup), record);
        } catch (final CommandTopicCorruptionException e) {
          log.warn("Backup is out of sync with the current command topic. "
              + "Backups will not work until the previous command topic is "
              + "restored or all backup files are deleted.", e);
          return records;
        }
        records.add(record);
      }
    }

    return records;
  }
@Test
  public void shouldNotGetCommandsWhenCommandTopicCorruptionWhenBackingUp() {
    // Given:
    when(commandConsumer.poll(any(Duration.class))).thenReturn(consumerRecords);
    doNothing().doThrow(new CommandTopicCorruptionException("error")).when(commandTopicBackup).writeRecord(any());

    // When:
    final Iterable<ConsumerRecord<byte[], byte[]>> newCommands = commandTopic
        .getNewCommands(Duration.ofHours(1));
    final List<ConsumerRecord<byte[], byte[]>> newCommandsList = ImmutableList.copyOf(newCommands);

    // Then:
    assertThat(newCommandsList.size(), is(1));
    assertThat(newCommandsList, equalTo(ImmutableList.of(record1)));
    verify(commandTopicBackup, never()).writeRecord(record3);
  }