@Override
    public boolean alterOffsets(Map<String, String> config, Map<Map<String, ?>, Map<String, ?>> offsets) {
        for (Map.Entry<Map<String, ?>, Map<String, ?>> offsetEntry : offsets.entrySet()) {
            Map<String, ?> sourceOffset = offsetEntry.getValue();
            if (sourceOffset == null) {
                // We allow tombstones for anything; if there's garbage in the offsets for the connector, we don't
                // want to prevent users from being able to clean it up using the REST API
                continue;
            }

            Map<String, ?> sourcePartition = offsetEntry.getKey();
            if (sourcePartition == null) {
                throw new ConnectException("Source partitions may not be null");
            }

            MirrorUtils.validateSourcePartitionString(sourcePartition, SOURCE_CLUSTER_ALIAS_KEY);
            MirrorUtils.validateSourcePartitionString(sourcePartition, TARGET_CLUSTER_ALIAS_KEY);

            MirrorUtils.validateSourceOffset(sourcePartition, sourceOffset, true);
        }

        // We don't actually use these offsets in the task class, so no additional effort is required beyond just validating
        // the format of the user-supplied offsets
        return true;
    }
@Test
    public void testAlterOffsetsTombstones() {
        MirrorHeartbeatConnector connector = new MirrorHeartbeatConnector();

        Function<Map<String, ?>, Boolean> alterOffsets = partition -> connector.alterOffsets(
                null,
                Collections.singletonMap(partition, null)
        );

        Map<String, Object> partition = sourcePartition("src", "bak");
        assertTrue(() -> alterOffsets.apply(partition));
        partition.put(SOURCE_CLUSTER_ALIAS_KEY, 618);
        assertTrue(() -> alterOffsets.apply(partition));
        partition.remove(SOURCE_CLUSTER_ALIAS_KEY);
        assertTrue(() -> alterOffsets.apply(partition));

        assertTrue(() -> alterOffsets.apply(null));
        assertTrue(() -> alterOffsets.apply(Collections.emptyMap()));
        assertTrue(() -> alterOffsets.apply(Collections.singletonMap("unused_partition_key", "unused_partition_value")));
    }