void incrementalAlterConfigs(Map<String, Config> topicConfigs) throws ExecutionException, InterruptedException {
        Map<ConfigResource, Collection<AlterConfigOp>> configOps = new HashMap<>();
        for (Map.Entry<String, Config> topicConfig : topicConfigs.entrySet()) {
            Collection<AlterConfigOp> ops = new ArrayList<>();
            ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, topicConfig.getKey());
            for (ConfigEntry config : topicConfig.getValue().entries()) {
                if (config.isDefault() && !shouldReplicateSourceDefault(config.name())) {
                    ops.add(new AlterConfigOp(config, AlterConfigOp.OpType.DELETE));
                } else {
                    ops.add(new AlterConfigOp(config, AlterConfigOp.OpType.SET));
                }
            }
            configOps.put(configResource, ops);
        }
        log.trace("Syncing configs for {} topics.", configOps.size());
        AtomicReference<Boolean> encounteredError = new AtomicReference<>(false);
        adminCall(
                () -> {
                    targetAdminClient.incrementalAlterConfigs(configOps).values().forEach((k, v) -> v.whenComplete((x, e) -> {
                        if (e != null) {
                            if (config.useIncrementalAlterConfigs().equals(MirrorSourceConfig.REQUEST_INCREMENTAL_ALTER_CONFIGS)
                                    && e instanceof UnsupportedVersionException && !encounteredError.get()) {
                                //Fallback logic
                                log.warn("The target cluster {} is not compatible with IncrementalAlterConfigs API. "
                                                + "Therefore using deprecated AlterConfigs API for syncing configs for topic {}",
                                        sourceAndTarget.target(), k.name(), e);
                                encounteredError.set(true);
                                useIncrementalAlterConfigs = false;
                            } else if (config.useIncrementalAlterConfigs().equals(MirrorSourceConfig.REQUIRE_INCREMENTAL_ALTER_CONFIGS)
                                    && e instanceof UnsupportedVersionException && !encounteredError.get()) {
                                log.error("Failed to sync configs for topic {} on cluster {} with IncrementalAlterConfigs API", k.name(), sourceAndTarget.target(), e);
                                encounteredError.set(true);
                                context.raiseError(new ConnectException("use.incremental.alter.configs was set to \"required\", but the target cluster '"
                                        + sourceAndTarget.target() + "' is not compatible with IncrementalAlterConfigs API", e));
                            } else {
                                log.warn("Could not alter configuration of topic {}.", k.name(), e);
                            }
                        }
                    }));
                    return null;
                },
                () -> String.format("incremental alter topic configs %s on %s cluster", topicConfigs, 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