@Test
    public void testReplicateFromLatest() throws Exception {
        // populate topic with records that should not be replicated
        String topic = "test-topic-1";
        produceMessages(primaryProducer, topic, NUM_PARTITIONS);

        String sentinelTopic = "test-topic-sentinel";
        primary.kafka().createTopic(sentinelTopic);

        // consume from the ends of topics when no committed offsets are found
        mm2Props.put(PRIMARY_CLUSTER_ALIAS + ".consumer." + AUTO_OFFSET_RESET_CONFIG, "latest");
        // 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);

        String backupSentinelTopic = remoteTopicName(sentinelTopic, PRIMARY_CLUSTER_ALIAS);
        waitForTopicCreated(backup, backupSentinelTopic);

        // wait for proof that the task has managed to poll its consumers at least once;
        // this should also mean that it knows the proper end offset of the other test topic,
        // and will consume exactly the expected number of records that we produce after
        // this assertion passes
        // NOTE: this assumes that there is a single MirrorSourceTask instance running;
        // if there are multiple tasks, the logic will need to be updated to ensure that each
        // task has managed to poll its consumer before continuing
        waitForCondition(
                () -> {
                    primary.kafka().produce(sentinelTopic, "sentinel-value");
                    int sentinelValues = backup.kafka().consumeAll(1_000, backupSentinelTopic).count();
                    return sentinelValues > 0;
                },
                RECORD_TRANSFER_DURATION_MS,
                "Records were not produced to sentinel topic in time"
        );

        // produce some more messages to the topic, now that MM2 is running and replication should be taking place
        produceMessages(primaryProducer, topic, NUM_PARTITIONS);

        String backupTopic = remoteTopicName(topic, PRIMARY_CLUSTER_ALIAS);
        // wait for at least the expected number of records to be replicated to the backup cluster
        backup.kafka().consume(NUM_RECORDS_PRODUCED, RECORD_TRANSFER_DURATION_MS, backupTopic);

        // consume all records from backup cluster
        ConsumerRecords<byte[], byte[]> replicatedRecords = backup.kafka().consumeAll(RECORD_TRANSFER_DURATION_MS, backupTopic);

        // ensure that we only replicated the records produced after startup
        replicatedRecords.partitions().forEach(topicPartition -> {
            int replicatedCount = replicatedRecords.records(topicPartition).size();
            assertEquals(
                    NUM_RECORDS_PER_PARTITION,
                    replicatedCount,
                    "Unexpected number of replicated records for partition " + topicPartition.partition()
            );
        });
    }
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