private boolean createTopic(Admin adminClient, NewTopic newTopic) {
        boolean doesTopicExist = false;
        String topic = newTopic.name();
        try {
            doesTopicExist = doesTopicExist(adminClient, topic);
            if (!doesTopicExist) {
                CreateTopicsResult result = adminClient.createTopics(Collections.singleton(newTopic));
                result.all().get();
                List<String> overriddenConfigs = result.config(topic).get()
                        .entries()
                        .stream()
                        .filter(entry -> !entry.isDefault())
                        .map(entry -> entry.name() + "=" + entry.value())
                        .collect(Collectors.toList());
                log.info("Topic {} created. TopicId: {}, numPartitions: {}, replicationFactor: {}, config: {}",
                        topic, result.topicId(topic).get(), result.numPartitions(topic).get(),
                        result.replicationFactor(topic).get(), overriddenConfigs);
                doesTopicExist = true;
            }
        } catch (Exception e) {
            // This exception can still occur as multiple brokers may call create topics and one of them may become
            // successful and other would throw TopicExistsException
            if (e.getCause() instanceof TopicExistsException) {
                log.info("Topic [{}] already exists", topic);
                doesTopicExist = true;
            } else {
                log.error("Encountered error while creating {} topic.", topic, e);
            }
        }
        return doesTopicExist;
    }
@param newTopic topic to be created.
@return Returns true if the topic already exists, or it is created successfully.
storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManager.java