private void testOneWayReplicationWithOffsetSyncs(int offsetLagMax) throws InterruptedException {
        produceMessages(primaryProducer, "test-topic-1");
        String backupTopic1 = remoteTopicName("test-topic-1", PRIMARY_CLUSTER_ALIAS);
        String consumerGroupName = "consumer-group-testOneWayReplicationWithAutoOffsetSync";
        Map<String, Object> consumerProps  = new HashMap<String, Object>() {{
                put("group.id", consumerGroupName);
                put("auto.offset.reset", "earliest");
            }};
        // create consumers before starting the connectors, so we don't need to wait for discovery
        try (Consumer<byte[], byte[]> primaryConsumer = primary.kafka().createConsumerAndSubscribeTo(consumerProps, 
                "test-topic-1")) {
            // we need to wait for consuming all the records for MM2 replicating the expected offsets
            waitForConsumingAllRecords(primaryConsumer, NUM_RECORDS_PRODUCED);
        }

        // enable automated consumer group offset sync
        mm2Props.put("sync.group.offsets.enabled", "true");
        mm2Props.put("sync.group.offsets.interval.seconds", "1");
        mm2Props.put("offset.lag.max", Integer.toString(offsetLagMax));
        // one way replication from primary to backup
        mm2Props.put(BACKUP_CLUSTER_ALIAS + "->" + PRIMARY_CLUSTER_ALIAS + ".enabled", "false");


        mm2Config = new MirrorMakerConfig(mm2Props);

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

        // make sure the topic is created in the primary cluster only
        String reverseTopic1 = remoteTopicName("test-topic-1", BACKUP_CLUSTER_ALIAS);
        if (!"test-topic-1".equals(reverseTopic1)) {
            topicShouldNotBeCreated(primary, reverseTopic1);
        }
        waitForTopicCreated(backup, backupTopic1);
        // create a consumer at backup cluster with same consumer group Id to consume 1 topic
        try (Consumer<byte[], byte[]> backupConsumer = backup.kafka().createConsumerAndSubscribeTo(
            consumerProps, backupTopic1)) {

            waitForConsumerGroupFullSync(backup, Collections.singletonList(backupTopic1),
                    consumerGroupName, NUM_RECORDS_PRODUCED, offsetLagMax);
            assertDownstreamRedeliveriesBoundedByMaxLag(backupConsumer, offsetLagMax);
        }

        // now create a new topic in primary cluster
        primary.kafka().createTopic("test-topic-2", NUM_PARTITIONS);
        // make sure the topic is created in backup cluster
        String remoteTopic2 = remoteTopicName("test-topic-2", PRIMARY_CLUSTER_ALIAS);
        waitForTopicCreated(backup, remoteTopic2);

        // produce some records to the new topic in primary cluster
        produceMessages(primaryProducer, "test-topic-2");

        // create a consumer at primary cluster to consume the new topic
        try (Consumer<byte[], byte[]> consumer1 = primary.kafka().createConsumerAndSubscribeTo(Collections.singletonMap(
                "group.id", consumerGroupName), "test-topic-2")) {
            // we need to wait for consuming all the records for MM2 replicating the expected offsets
            waitForConsumingAllRecords(consumer1, NUM_RECORDS_PRODUCED);
        }

        // create a consumer at backup cluster with same consumer group ID to consume old and new topic
        try (Consumer<byte[], byte[]> backupConsumer = backup.kafka().createConsumerAndSubscribeTo(Collections.singletonMap(
            "group.id", consumerGroupName), backupTopic1, remoteTopic2)) {

            waitForConsumerGroupFullSync(backup, Arrays.asList(backupTopic1, remoteTopic2),
                    consumerGroupName, NUM_RECORDS_PRODUCED, offsetLagMax);
            assertDownstreamRedeliveriesBoundedByMaxLag(backupConsumer, offsetLagMax);
        }

        assertMonotonicCheckpoints(backup, PRIMARY_CLUSTER_ALIAS + ".checkpoints.internal");
    }
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