class MirrorSourceConnector extends SourceConnector {

    void refreshTopicPartitions()
            throws InterruptedException, ExecutionException {
        knownSourceTopicPartitions = findSourceTopicPartitions();
        knownTargetTopicPartitions = findTargetTopicPartitions();
        List<TopicPartition> upstreamTargetTopicPartitions = knownTargetTopicPartitions.stream()
                .map(x -> new TopicPartition(replicationPolicy.upstreamTopic(x.topic()), x.partition()))
                .collect(Collectors.toList());

        Set<TopicPartition> newTopicPartitions = new HashSet<>(knownSourceTopicPartitions);
        newTopicPartitions.removeAll(upstreamTargetTopicPartitions);
        Set<TopicPartition> deadTopicPartitions = new HashSet<>(upstreamTargetTopicPartitions);
        deadTopicPartitions.removeAll(knownSourceTopicPartitions);
        if (!newTopicPartitions.isEmpty() || !deadTopicPartitions.isEmpty()) {
            log.info("Found {} topic-partitions on {}. {} are new. {} were removed. Previously had {}.",
                    knownSourceTopicPartitions.size(), sourceAndTarget.source(), newTopicPartitions.size(),
                    deadTopicPartitions.size(), knownSourceTopicPartitions.size());
            log.trace("Found new topic-partitions: {}", newTopicPartitions);
            computeAndCreateTopicPartitions();
            context.requestTaskReconfiguration();
        }
    }

    public  MirrorSourceConnector();
      MirrorSourceConnector(List<TopicPartition> knownSourceTopicPartitions, MirrorConnectorConfig config);
      MirrorSourceConnector(SourceAndTarget sourceAndTarget, ReplicationPolicy replicationPolicy,
            TopicFilter topicFilter, ConfigPropertyFilter configPropertyFilter);

}