public synchronized String createTopic(String topicName, int partitions)
      throws KafkaResourceManagerException {
    checkArgument(partitions > 0, "partitions must be positive.");

    String uniqueName = KafkaResourceManagerUtils.generateTopicName(topicName);
    try {
      Set<String> currentTopics = kafkaClient.listTopics().names().get();
      if (!currentTopics.contains(uniqueName)) {
        kafkaClient
            .createTopics(
                Collections.singletonList(new NewTopic(uniqueName, partitions, (short) 1)))
            .all()
            .get();
        topicNames.add(uniqueName);
      }
    } catch (Exception e) {
      throw new KafkaResourceManagerException("Error creating topics.", e);
    }

    LOG.info("Successfully created topic {}.", uniqueName);

    return uniqueName;
  }
@Test
  public void testCreateTopicShouldThrowErrorWhenKafkaFailsToCreateTopic()
      throws ExecutionException, InterruptedException {
    when(kafkaClient.createTopics(any(Collection.class)).all().get())
        .thenThrow(new ExecutionException(new RuntimeException("create topic future fails")));

    assertThrows(KafkaResourceManagerException.class, () -> testManager.createTopic(TOPIC_NAME, 1));
  }