boolean shouldReplicateTopic(String topic) {
        return (topicFilter.shouldReplicateTopic(topic) || replicationPolicy.isHeartbeatsTopic(topic))
            && !replicationPolicy.isInternalTopic(topic) && !isCycle(topic);
    }
@Test
    public void testIdentityReplication() {
        MirrorSourceConnector connector = new MirrorSourceConnector(new SourceAndTarget("source", "target"),
            new IdentityReplicationPolicy(), x -> true, getConfigPropertyFilter());
        assertTrue(connector.shouldReplicateTopic("target.topic1"), "should allow cycles");
        assertTrue(connector.shouldReplicateTopic("target.source.topic1"), "should allow cycles");
        assertTrue(connector.shouldReplicateTopic("source.target.topic1"), "should allow cycles");
        assertTrue(connector.shouldReplicateTopic("target.source.target.topic1"), "should allow cycles");
        assertTrue(connector.shouldReplicateTopic("source.target.source.topic1"), "should allow cycles");
        assertTrue(connector.shouldReplicateTopic("topic1"), "should allow normal topics");
        assertTrue(connector.shouldReplicateTopic("othersource.topic1"), "should allow normal topics");
        assertFalse(connector.shouldReplicateTopic("target.heartbeats"), "should not allow heartbeat cycles");
        assertFalse(connector.shouldReplicateTopic("target.source.heartbeats"), "should not allow heartbeat cycles");
        assertFalse(connector.shouldReplicateTopic("source.target.heartbeats"), "should not allow heartbeat cycles");
        assertFalse(connector.shouldReplicateTopic("target.source.target.heartbeats"), "should not allow heartbeat cycles");
        assertFalse(connector.shouldReplicateTopic("source.target.source.heartbeats"), "should not allow heartbeat cycles");
        assertTrue(connector.shouldReplicateTopic("heartbeats"), "should allow heartbeat topics");
        assertTrue(connector.shouldReplicateTopic("othersource.heartbeats"), "should allow heartbeat topics");
    }