public static boolean createTopic(AdminClient adminClient, NewTopic topicToBeCreated) {
    try {
      boolean retryResponse = CruiseControlMetricsUtils.retry(() -> {
        try {
          CreateTopicsResult createTopicsResult = adminClient.createTopics(Collections.singletonList(topicToBeCreated));
          createTopicsResult.values().get(topicToBeCreated.name()).get(CLIENT_REQUEST_TIMEOUT_MS, TimeUnit.MILLISECONDS);
          LOG.info("Topic {} has been created.", topicToBeCreated.name());
          return false;
        } catch (InterruptedException | ExecutionException | TimeoutException e) {
          if (e.getCause() instanceof TopicExistsException) {
            throw (TopicExistsException) e.getCause();
          }
          LOG.warn("Unable to create topic {}.", topicToBeCreated.name(), e);
          return true;
        }
      }, TOPIC_AUTO_CREATE_RETRIES);
      if (!retryResponse) {
        throw new IllegalStateException(String.format("Failed to create topic %s.", topicToBeCreated.name()));
      }
    } catch (TopicExistsException e) {
      return false;
    }
    return true;
  }
@Test
  public void testCreateTopic() throws ExecutionException, InterruptedException, TimeoutException {
    AdminClient adminClient = EasyMock.createMock(AdminClient.class);
    CreateTopicsResult createTopicsResult = EasyMock.createMock(CreateTopicsResult.class);
    KafkaFuture<Void> createTopicsResultFuture = EasyMock.createMock(KafkaFuture.class);
    Map<String, KafkaFuture<Void>> createTopicsResultValues = Collections.singletonMap(TEST_TOPIC.name(), createTopicsResultFuture);

    // 1. Existing topic
    EasyMock.expect(adminClient.createTopics(Collections.singletonList(TEST_TOPIC))).andReturn(createTopicsResult).once();
    EasyMock.expect(createTopicsResult.values()).andReturn(createTopicsResultValues).once();
    EasyMock.expect(createTopicsResultFuture.get(CLIENT_REQUEST_TIMEOUT_MS, TimeUnit.MILLISECONDS))
            .andThrow(new ExecutionException(Errors.TOPIC_ALREADY_EXISTS.exception()));

    EasyMock.replay(adminClient, createTopicsResult, createTopicsResultFuture);

    assertFalse(createTopic(adminClient, TEST_TOPIC));
    EasyMock.verify(adminClient, createTopicsResult, createTopicsResultFuture);
    EasyMock.reset(adminClient, createTopicsResult, createTopicsResultFuture);

    // 2. New topic
    EasyMock.expect(adminClient.createTopics(Collections.singletonList(TEST_TOPIC))).andReturn(createTopicsResult).once();
    EasyMock.expect(createTopicsResult.values()).andReturn(createTopicsResultValues).once();
    EasyMock.expect(createTopicsResultFuture.get(CLIENT_REQUEST_TIMEOUT_MS, TimeUnit.MILLISECONDS)).andReturn(null).once();

    EasyMock.replay(adminClient, createTopicsResult, createTopicsResultFuture);

    assertTrue(createTopic(adminClient, TEST_TOPIC));
    EasyMock.verify(adminClient, createTopicsResult, createTopicsResultFuture);
  }