private void validateExactlyOnceConfigs(Map<String, String> props, List<ConfigValue> configValues) {
        if ("required".equals(props.get(EXACTLY_ONCE_SUPPORT_CONFIG))) {
            if (!consumerUsesReadCommitted(props)) {
                ConfigValue exactlyOnceSupport = configValues.stream()
                        .filter(cv -> EXACTLY_ONCE_SUPPORT_CONFIG.equals(cv.name()))
                        .findAny()
                        .orElseGet(() -> {
                            ConfigValue result = new ConfigValue(EXACTLY_ONCE_SUPPORT_CONFIG);
                            configValues.add(result);
                            return result;
                        });
                // The Connect framework will already generate an error for this property if we return ExactlyOnceSupport.UNSUPPORTED
                // from our exactlyOnceSupport method, but it will be fairly generic
                // We add a second error message here to give users more insight into why this specific connector can't support exactly-once
                // guarantees with the given configuration
                exactlyOnceSupport.addErrorMessage(
                        "MirrorSourceConnector can only provide exactly-once guarantees when its source consumer is configured with "
                                + ConsumerConfig.ISOLATION_LEVEL_CONFIG + " set to '" + READ_COMMITTED + "'; "
                                + "otherwise, records from aborted and uncommitted transactions will be replicated from the "
                                + "source cluster to the target cluster."
                );
            }
        }
    }
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