public void createTopic(String topic, int partitions, int replication, Map<String, String> topicConfig, Map<String, Object> adminClientConfig) {
        if (replication > cluster.brokers().size()) {
            throw new InvalidReplicationFactorException("Insufficient brokers ("
                    + cluster.brokers().size() + ") for desired replication (" + replication + ")");
        }

        log.info("Creating topic { name: {}, partitions: {}, replication: {}, config: {} }",
                topic, partitions, replication, topicConfig);
        final NewTopic newTopic = new NewTopic(topic, partitions, (short) replication);
        newTopic.configs(topicConfig);

        try (final Admin adminClient = createAdminClient(adminClientConfig)) {
            adminClient.createTopics(Collections.singletonList(newTopic)).all().get();
        } catch (final InterruptedException | ExecutionException e) {
            throw new RuntimeException(e);
        }
    }
Create a Kafka topic with the given parameters.

@param topic             The name of the topic.
@param partitions        The number of partitions for this topic.
@param replication       The replication factor for (partitions of) this topic.
@param topicConfig       Additional topic-level configuration settings.
@param adminClientConfig Additional admin client configuration settings.
connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedKafkaCluster.java