private Optional<OffsetSync> latestOffsetSync(TopicPartition topicPartition, long upstreamOffset) {
        return Optional.ofNullable(offsetSyncs.get(topicPartition))
                .map(syncs -> lookupLatestSync(syncs, upstreamOffset));
    }
Start the OffsetSyncStore, blocking until all previous Offset Syncs have been read from backing storage.
connect/mirror/src/main/java/org/apache/kafka/connect/mirror/OffsetSyncStore.java