public List<SourceAndTarget> clusterPairs() {
        List<SourceAndTarget> pairs = new ArrayList<>();
        Set<String> clusters = clusters();
        Map<String, String> originalStrings = originalsStrings();
        boolean globalHeartbeatsEnabled = MirrorHeartbeatConfig.EMIT_HEARTBEATS_ENABLED_DEFAULT;
        if (originalStrings.containsKey(MirrorHeartbeatConfig.EMIT_HEARTBEATS_ENABLED)) {
            globalHeartbeatsEnabled = Boolean.parseBoolean(originalStrings.get(MirrorHeartbeatConfig.EMIT_HEARTBEATS_ENABLED));
        }

        for (String source : clusters) {
            for (String target : clusters) {
                if (!source.equals(target)) {
                    String clusterPairConfigPrefix = source + "->" + target + ".";
                    boolean clusterPairEnabled = Boolean.parseBoolean(originalStrings.get(clusterPairConfigPrefix + "enabled"));
                    boolean clusterPairHeartbeatsEnabled = globalHeartbeatsEnabled;
                    if (originalStrings.containsKey(clusterPairConfigPrefix + MirrorHeartbeatConfig.EMIT_HEARTBEATS_ENABLED)) {
                        clusterPairHeartbeatsEnabled = Boolean.parseBoolean(originalStrings.get(clusterPairConfigPrefix + MirrorHeartbeatConfig.EMIT_HEARTBEATS_ENABLED));
                    }

                    // By default, all source->target Herder combinations are created even if `x->y.enabled=false`
                    // Unless `emit.heartbeats.enabled=false` or `x->y.emit.heartbeats.enabled=false`
                    // Reason for this behavior: for a given replication flow A->B with heartbeats, 2 herders are required :
                    // B->A for the MirrorHeartbeatConnector (emits heartbeats into A for monitoring replication health)
                    // A->B for the MirrorSourceConnector (actual replication flow)
                    if (clusterPairEnabled || clusterPairHeartbeatsEnabled) {
                        pairs.add(new SourceAndTarget(source, target));
                    }
                }
            }
        }
        return pairs;
    }
@Test
    public void testClusterPairsWithGloballyDisabledHeartbeats() {
        MirrorMakerConfig mirrorConfig = new MirrorMakerConfig(makeProps(
                "clusters", "a, b, c, d, e, f",
                "emit.heartbeats.enabled", "false",
                "a->b.enabled", "true",
                "a->c.enabled", "true",
                "a->d.enabled", "true",
                "a->e.enabled", "false",
                "a->f.enabled", "false"));
        List<SourceAndTarget> clusterPairs = mirrorConfig.clusterPairs();
        assertEquals(3, clusterPairs.size(),
            "clusterPairs count should match (x->y.enabled=true or x->y.emit.heartbeats.enabled=true) count");

        // Link b->a.enabled doesn't exist therefore it must not be in clusterPairs
        SourceAndTarget sourceAndTarget = new SourceAndTarget("b", "a");
        assertFalse(clusterPairs.contains(sourceAndTarget), "disabled/unset link x->y should not be in clusterPairs");
    }