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);
  }
@Test
  public void shouldRestoreCommandTopicWithEmptyCommands() {
    // Given:
    when(topicClient.isTopicExists(COMMAND_TOPIC_NAME)).thenReturn(false);

    // When:
    restoreCommandTopic.restore(Collections.emptyList());

    // Then:
    verifyCreateCommandTopic();
    final InOrder inOrder = inOrder(kafkaProducer);
    inOrder.verify(kafkaProducer).initTransactions();
    inOrder.verify(kafkaProducer).close();
    verifyNoMoreInteractions(kafkaProducer, future1, future2, future3);
  }