protected static void resetSomeMirrorMakerSourceConnectorOffsets(EmbeddedConnectCluster connectCluster, String... topics) {
        Set<String> topicsSet = new HashSet<>(Arrays.asList(topics));
        String connectorName = MirrorSourceConnector.class.getSimpleName();

        ConnectorOffsets currentOffsets = connectCluster.connectorOffsets(connectorName);
        List<ConnectorOffset> alteredOffsetContents = currentOffsets.offsets().stream()
                .map(connectorOffset -> {
                    TopicPartition topicPartition = MirrorUtils.unwrapPartition(connectorOffset.partition());
                    if (!topicsSet.contains(topicPartition.topic())) {
                        return null;
                    }

                    return new ConnectorOffset(connectorOffset.partition(), null);
                }).filter(Objects::nonNull)
                .collect(Collectors.toList());

        connectCluster.alterConnectorOffsets(connectorName, new ConnectorOffsets(alteredOffsetContents));
    }
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