@Override
    public Set<Map<String, Object>> connectorPartitions(String connectorName) {
        return connectorPartitions.getOrDefault(connectorName, Collections.emptySet());
    }
@Test
    public void testConnectorPartitions() throws Exception {
        @SuppressWarnings("unchecked")
        Callback<Void> setCallback = mock(Callback.class);

        // This test actually requires the offset store to track deserialized source partitions, so we can't use the member variable mock converter
        JsonConverter jsonConverter = new JsonConverter();
        jsonConverter.configure(Collections.singletonMap(JsonConverterConfig.SCHEMAS_ENABLE_CONFIG, "false"), true);

        Map<ByteBuffer, ByteBuffer> serializedPartitionOffsets = new HashMap<>();
        serializedPartitionOffsets.put(
                serializeKey(jsonConverter, "connector1", Collections.singletonMap("partitionKey", "partitionValue1")),
                serialize(jsonConverter, Collections.singletonMap("offsetKey", "offsetValue"))
        );
        store.set(serializedPartitionOffsets, setCallback).get();

        serializedPartitionOffsets.put(
                serializeKey(jsonConverter, "connector1", Collections.singletonMap("partitionKey", "partitionValue1")),
                serialize(jsonConverter, Collections.singletonMap("offsetKey", "offsetValue2"))
        );
        serializedPartitionOffsets.put(
                serializeKey(jsonConverter, "connector1", Collections.singletonMap("partitionKey", "partitionValue2")),
                serialize(jsonConverter, Collections.singletonMap("offsetKey", "offsetValue"))
        );
        serializedPartitionOffsets.put(
                serializeKey(jsonConverter, "connector2", Collections.singletonMap("partitionKey", "partitionValue")),
                serialize(jsonConverter, Collections.singletonMap("offsetKey", "offsetValue"))
        );

        store.set(serializedPartitionOffsets, setCallback).get();
        store.stop();

        // Restore into a new store to ensure correct reload from scratch
        FileOffsetBackingStore restore = new FileOffsetBackingStore(jsonConverter);
        restore.configure(config);
        restore.start();

        Set<Map<String, Object>> connectorPartitions1 = restore.connectorPartitions("connector1");
        Set<Map<String, Object>> expectedConnectorPartition1 = new HashSet<>();
        expectedConnectorPartition1.add(Collections.singletonMap("partitionKey", "partitionValue1"));
        expectedConnectorPartition1.add(Collections.singletonMap("partitionKey", "partitionValue2"));
        assertEquals(expectedConnectorPartition1, connectorPartitions1);

        Set<Map<String, Object>> connectorPartitions2 = restore.connectorPartitions("connector2");
        Set<Map<String, Object>> expectedConnectorPartition2 = Collections.singleton(Collections.singletonMap("partitionKey", "partitionValue"));
        assertEquals(expectedConnectorPartition2, connectorPartitions2);

        serializedPartitionOffsets.clear();
        // Null valued offset for a partition key should remove that partition for the connector
        serializedPartitionOffsets.put(
                serializeKey(jsonConverter, "connector1", Collections.singletonMap("partitionKey", "partitionValue1")),
                null
        );
        restore.set(serializedPartitionOffsets, setCallback).get();
        connectorPartitions1 = restore.connectorPartitions("connector1");
        assertEquals(Collections.singleton(Collections.singletonMap("partitionKey", "partitionValue2")), connectorPartitions1);

        verify(setCallback, times(3)).onCompletion(isNull(), isNull());
    }