@Test
    public void testSyncTopicConfigs() throws InterruptedException {
        mm2Config = new MirrorMakerConfig(mm2Props);

        waitUntilMirrorMakerIsRunning(backup, CONNECTOR_LIST, mm2Config, PRIMARY_CLUSTER_ALIAS, BACKUP_CLUSTER_ALIAS);

        // create topic with configuration to test:
        final Map<String, String> topicConfig = new HashMap<>();
        topicConfig.put("delete.retention.ms", "1000"); // should be excluded (default value is 86400000)
        topicConfig.put("retention.bytes", "1000"); // should be included, default value is -1

        final String topic = "test-topic-with-config";
        final String backupTopic = remoteTopicName(topic, PRIMARY_CLUSTER_ALIAS);

        primary.kafka().createTopic(topic, NUM_PARTITIONS, 1, topicConfig);
        waitForTopicCreated(backup, backupTopic);

        // alter configs on the target topic
        ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, backupTopic);
        Collection<AlterConfigOp> ops = new ArrayList<>();
        ops.add(new AlterConfigOp(new ConfigEntry("delete.retention.ms", "2000"), AlterConfigOp.OpType.SET));
        ops.add(new AlterConfigOp(new ConfigEntry("retention.bytes", "2000"), AlterConfigOp.OpType.SET));
        Map<ConfigResource, Collection<AlterConfigOp>> configOps = Collections.singletonMap(configResource, ops);
        // alter configs on target cluster
        backup.kafka().incrementalAlterConfigs(configOps);

        waitForCondition(() -> {
            String primaryConfig, backupConfig;
            primaryConfig = getTopicConfig(primary.kafka(), topic, "delete.retention.ms");
            backupConfig = getTopicConfig(backup.kafka(), backupTopic, "delete.retention.ms");
            assertNotEquals(primaryConfig, backupConfig,
                    "`delete.retention.ms` should be different, because it's in exclude filter! ");
            assertEquals("2000", backupConfig, "`delete.retention.ms` should be 2000, because it's explicitly defined on the target topic! ");

            // regression test for the config that are still supposed to be replicated
            primaryConfig = getTopicConfig(primary.kafka(), topic, "retention.bytes");
            backupConfig = getTopicConfig(backup.kafka(), backupTopic, "retention.bytes");
            assertEquals(primaryConfig, backupConfig,
                    "`retention.bytes` should be the same, because it isn't in exclude filter! ");
            return true;
        }, 30000, "Topic configurations were not synced");
    }
Tests MM2 replication and failover/failback logic.

MM2 is configured with active/active replication between two Kafka clusters. Tests validate that
records sent to either cluster arrive at the other cluster. Then, a consumer group is migrated from
one cluster to the other and back. Tests validate that consumer offsets are translated and replicated
between clusters during this failover and failback.
connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationBaseTest.java