void createNewTopics(Map<String, NewTopic> newTopics) throws ExecutionException, InterruptedException {
        adminCall(
                () -> {
                    targetAdminClient.createTopics(newTopics.values(), new CreateTopicsOptions()).values()
                            .forEach((k, v) -> v.whenComplete((x, e) -> {
                                if (e != null) {
                                    log.warn("Could not create topic {}.", k, e);
                                } else {
                                    log.info("Created remote topic {} with {} partitions.", k, newTopics.get(k).numPartitions());
                                }
                            }));
                    return null;
                },
                () -> String.format("create topics %s on %s cluster", newTopics, config.targetClusterAlias())
        );
    }
Replicate data, configuration, and ACLs between clusters.

 @see MirrorSourceConfig for supported config properties.
connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorSourceConnector.java