public void restore(final List<Pair<byte[], byte[]>> backupCommands) {
    // Delete the command topic
    deleteCommandTopicIfExists();

    // Create the command topic
    KsqlInternalTopicUtils.ensureTopic(commandTopicName, serverConfig, topicClient);

    // Restore the commands
    restoreCommandTopic(backupCommands);
  }
@SuppressFBWarnings("RV_RETURN_VALUE_IGNORED_NO_SIDE_EFFECT")
  @Test
  public void shouldThrowIfCannotCreateTopic() {
    // Given:
    when(topicClient.isTopicExists(COMMAND_TOPIC_NAME)).thenReturn(false);
    doThrow(new RuntimeException("denied")).when(topicClient)
        .createTopic(COMMAND_TOPIC_NAME, INTERNAL_TOPIC_PARTITION_COUNT,
            INTERNAL_TOPIC_REPLICAS_COUNT, INTERNAL_TOPIC_CONFIG);

    // When:
    final Exception e = assertThrows(
        RuntimeException.class,
        () -> restoreCommandTopic.restore(Collections.singletonList(BACKUP_COMMANDS.get(0))));

    // Then:
    assertThat(e.getMessage(), containsString("denied"));
    verify(topicClient, times(2)).isTopicExists(COMMAND_TOPIC_NAME);
    verifyCreateCommandTopic();
    verifyNoMoreInteractions(topicClient);
    verifyNoMoreInteractions(kafkaProducer);
  }