@Test
    public void testOffsetTranslationBehindReplicationFlow() throws InterruptedException {
        String consumerGroupName = "consumer-group-lagging-behind";
        Map<String, Object> consumerProps = Collections.singletonMap("group.id", consumerGroupName);
        String remoteTopic = remoteTopicName("test-topic-1", PRIMARY_CLUSTER_ALIAS);
        warmUpConsumer(consumerProps);
        mm2Props.put("sync.group.offsets.enabled", "true");
        mm2Props.put("sync.group.offsets.interval.seconds", "1");
        mm2Props.put("offset.lag.max", Integer.toString(OFFSET_LAG_MAX));
        mm2Config = new MirrorMakerConfig(mm2Props);
        waitUntilMirrorMakerIsRunning(backup, CONNECTOR_LIST, mm2Config, PRIMARY_CLUSTER_ALIAS, BACKUP_CLUSTER_ALIAS);
        // Produce and consume an initial batch of records to establish an initial checkpoint
        produceMessages(primaryProducer, "test-topic-1");
        warmUpConsumer(consumerProps);
        MirrorClient backupClient = new MirrorClient(mm2Config.clientConfig(BACKUP_CLUSTER_ALIAS));
        Map<TopicPartition, OffsetAndMetadata> initialCheckpoints = waitForCheckpointOnAllPartitions(
                backupClient, consumerGroupName, PRIMARY_CLUSTER_ALIAS, remoteTopic);
        // Produce a large number of records to the topic, all replicated within one MM2 lifetime.
        int iterations = 100;
        for (int i = 1; i < iterations; i++) {
            produceMessages(primaryProducer, "test-topic-1");
        }
        waitForTopicCreated(backup, remoteTopic);
        assertEquals(iterations * NUM_RECORDS_PRODUCED, backup.kafka().consume(iterations * NUM_RECORDS_PRODUCED, RECORD_TRANSFER_DURATION_MS, remoteTopic).count(),
                "Records were not replicated to backup cluster.");
        // Once the replication has finished, we spin up the upstream consumer again and start consuming records
        ConsumerRecords<byte[], byte[]> allRecords = primary.kafka().consume(iterations * NUM_RECORDS_PRODUCED, RECORD_CONSUME_DURATION_MS, "test-topic-1");
        Map<TopicPartition, OffsetAndMetadata> partialCheckpoints;
        log.info("Initial checkpoints: {}", initialCheckpoints);
        try (Consumer<byte[], byte[]> primaryConsumer = primary.kafka().createConsumerAndSubscribeTo(consumerProps, "test-topic-1")) {
            primaryConsumer.poll(CONSUMER_POLL_TIMEOUT_MS);
            primaryConsumer.commitSync(partialOffsets(allRecords, 0.9f));
            partialCheckpoints = waitForNewCheckpointOnAllPartitions(
                    backupClient, consumerGroupName, PRIMARY_CLUSTER_ALIAS, remoteTopic, initialCheckpoints);
            log.info("Partial checkpoints: {}", partialCheckpoints);
        }

        for (TopicPartition tp : initialCheckpoints.keySet()) {
            assertTrue(initialCheckpoints.get(tp).offset() < partialCheckpoints.get(tp).offset(),
                    "Checkpoints should advance when the upstream consumer group advances");
        }

        assertMonotonicCheckpoints(backup, PRIMARY_CLUSTER_ALIAS + ".checkpoints.internal");

        Map<TopicPartition, OffsetAndMetadata> finalCheckpoints;
        try (Consumer<byte[], byte[]> primaryConsumer = primary.kafka().createConsumerAndSubscribeTo(consumerProps, "test-topic-1")) {
            primaryConsumer.poll(CONSUMER_POLL_TIMEOUT_MS);
            primaryConsumer.commitSync(partialOffsets(allRecords, 0.1f));
            finalCheckpoints = waitForNewCheckpointOnAllPartitions(
                    backupClient, consumerGroupName, PRIMARY_CLUSTER_ALIAS, remoteTopic, partialCheckpoints);
            log.info("Final checkpoints: {}", finalCheckpoints);
        }

        backupClient.close();

        for (TopicPartition tp : partialCheckpoints.keySet()) {
            assertTrue(finalCheckpoints.get(tp).offset() < partialCheckpoints.get(tp).offset(),
                    "Checkpoints should rewind when the upstream consumer group rewinds");
        }
    }
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