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 shouldCreateAndRestoreCommandTopic() throws ExecutionException, InterruptedException {
    // Given:
    when(topicClient.isTopicExists(COMMAND_TOPIC_NAME)).thenReturn(false);

    // When:
    restoreCommandTopic.restore(BACKUP_COMMANDS);

    // Then:
    verifyCreateCommandTopic();

    final InOrder inOrder = inOrder(kafkaProducer, future1, future2, future3);
    inOrder.verify(kafkaProducer).initTransactions();
    inOrder.verify(kafkaProducer).beginTransaction();
    inOrder.verify(kafkaProducer).send(RECORD_1);
    inOrder.verify(future1).get();
    inOrder.verify(kafkaProducer).commitTransaction();
    inOrder.verify(kafkaProducer).beginTransaction();
    inOrder.verify(kafkaProducer).send(RECORD_2);
    inOrder.verify(future2).get();
    inOrder.verify(kafkaProducer).commitTransaction();
    inOrder.verify(kafkaProducer).beginTransaction();
    inOrder.verify(kafkaProducer).send(RECORD_3);
    inOrder.verify(future3).get();
    inOrder.verify(kafkaProducer).commitTransaction();
    inOrder.verify(kafkaProducer).close();
    verifyNoMoreInteractions(kafkaProducer, future1, future2, future3);
  }