Repository: apache/kafka
File: connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorSourceConnector.java (Java)

Code before:
    }

    private Collection<TopicDescription> describeTopics(Admin adminClient, Collection<String> topics)
            throws InterruptedException, ExecutionException {
        return adminCall(
                () -> adminClient.describeTopics(topics).allTopicNames().get().values(),
                () -> String.format("describe topics %s on %s cluster", topics, actualClusterAlias(adminClient))
        );
    }

    static Map<String, String> configToMap(Config config) {
        return config.entries().stream()
                .collect(Collectors.toMap(ConfigEntry::name, ConfigEntry::value));
    }

    // visible for testing
    // use deprecated alterConfigs API for broker compatibility back to 0.11.0
    @SuppressWarnings("deprecation")
    void deprecatedAlterConfigs(Map<String, Config> topicConfigs) throws ExecutionException, InterruptedException {
        Map<ConfigResource, Config> configs = topicConfigs.entrySet().stream()
            .collect(Collectors.toMap(x ->
                new ConfigResource(ConfigResource.Type.TOPIC, x.getKey()), Entry::getValue));
        log.trace("Syncing configs for {} topics.", configs.size());
        adminCall(
                () -> {
                    targetAdminClient.alterConfigs(configs).values().forEach((k, v) -> v.whenComplete((x, e) -> {
                        if (e != null) {
                            log.warn("Could not alter configuration of topic {}.", k.name(), e);
                        }
                    }));
                    return null;
                },
                () -> String.format("alter topic configs %s on %s cluster", topicConfigs, config.targetClusterAlias())
        );
    }

    // visible for testing
    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);
        }

Review comment (suggestion):
Unexpected WARN log will be printed even if the alterConfig completes without error (i.e. `e == null`). Opened [KAFKA-20017](https://issues.apache.org/jira/browse/KAFKA-20017) to fix it.

Code after:
            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());
        adminCall(() -> {
            targetAdminClient.incrementalAlterConfigs(configOps).values()
                .forEach((k, v) -> v.whenComplete((x, e) -> {
                    if (e instanceof UnsupportedVersionException) {
                        log.error("Failed to sync configs for topic {} on cluster {} with " +
                                "IncrementalAlterConfigs API", k.name(), sourceAndTarget.target(), e);
                        context.raiseError(new ConnectException("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()));
    }

    private void updateTopicAcls(List<AclBinding> bindings) throws ExecutionException, InterruptedException {
        log.trace("Syncing {} topic ACL bindings.", bindings.size());
        adminCall(
                () -> {
                    targetAdminClient.createAcls(bindings).values().forEach((k, v) -> v.whenComplete((x, e) -> {
                        if (e != null) {
                            log.warn("Could not sync ACL of topic {}.", k.pattern().name(), e);
                        }
                    }));
                    return null;
                },
                () -> String.format("create ACLs %s on %s cluster", bindings, config.targetClusterAlias())
        );
    }

    private static Stream<TopicPartition> expandTopicDescription(TopicDescription description) {
        String topic = description.name();
        return description.partitions().stream()
            .map(x -> new TopicPartition(topic, x.partition()));