@Test
    public void testOffsetSyncsTopicsOnTarget() throws Exception {
        // move offset-syncs topics to target
        mm2Props.put(PRIMARY_CLUSTER_ALIAS + "->" + BACKUP_CLUSTER_ALIAS + ".offset-syncs.topic.location", "target");
        // 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);

        // Ensure the offset syncs topic is created in the target cluster
        waitForTopicCreated(backup, "mm2-offset-syncs." + PRIMARY_CLUSTER_ALIAS + ".internal");

        String consumerGroupName = "consumer-group-syncs-on-target";
        Map<String, Object> consumerProps = Collections.singletonMap("group.id", consumerGroupName);

        produceMessages(primaryProducer, "test-topic-1");

        warmUpConsumer(consumerProps);

        String remoteTopic = remoteTopicName("test-topic-1", PRIMARY_CLUSTER_ALIAS);

        // Check offsets are pushed to the checkpoint topic
        try (Consumer<byte[], byte[]> backupConsumer = backup.kafka().createConsumerAndSubscribeTo(Collections.singletonMap(
                "auto.offset.reset", "earliest"), PRIMARY_CLUSTER_ALIAS + ".checkpoints.internal")) {
            waitForCondition(() -> {
                ConsumerRecords<byte[], byte[]> records = backupConsumer.poll(Duration.ofSeconds(1L));
                for (ConsumerRecord<byte[], byte[]> record : records) {
                    Checkpoint checkpoint = Checkpoint.deserializeRecord(record);
                    if (remoteTopic.equals(checkpoint.topicPartition().topic())) {
                        return true;
                    }
                }
                return false;
            }, 30_000,
                "Unable to find checkpoints for " + PRIMARY_CLUSTER_ALIAS + ".test-topic-1"
            );
        }

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

        // Ensure no offset-syncs topics have been created on the primary cluster
        try (Admin adminClient = primary.kafka().createAdminClient()) {
            Set<String> primaryTopics = adminClient.listTopics().names().get();
            assertFalse(primaryTopics.contains("mm2-offset-syncs." + PRIMARY_CLUSTER_ALIAS + ".internal"));
            assertFalse(primaryTopics.contains("mm2-offset-syncs." + BACKUP_CLUSTER_ALIAS + ".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