@Override
    public Set<Map<String, Object>> connectorPartitions(String connectorName) {
        return connectorPartitions.getOrDefault(connectorName, Collections.emptySet());
    }
@Test
    public void testConnectorPartitions() throws Exception {
        JsonConverter jsonConverter = new JsonConverter();
        jsonConverter.configure(Collections.singletonMap(JsonConverterConfig.SCHEMAS_ENABLE_CONFIG, "false"), true);
        store = spy(new KafkaOffsetBackingStore(() -> {
            fail("Should not attempt to instantiate admin in these tests");
            return null;
        }, () -> CLIENT_ID_BASE, jsonConverter));

        doReturn(storeLog).when(store).createKafkaBasedLog(capturedTopic.capture(), capturedProducerProps.capture(),
                capturedConsumerProps.capture(), capturedConsumedCallback.capture(),
                capturedNewTopic.capture(), capturedAdminSupplier.capture(), any(), any());

        store.configure(mockConfig(props));
        store.start();

        verify(storeLog).start();

        doAnswer(invocation -> {
            capturedConsumedCallback.getValue().onCompletion(null,
                    new ConsumerRecord<>(TOPIC, 0, 0, 0L, TimestampType.CREATE_TIME, 0, 0,
                            jsonConverter.fromConnectData("", null, Arrays.asList("connector1",
                                    Collections.singletonMap("partitionKey", "partitionValue1"))), TP0_VALUE.array(),
                            new RecordHeaders(), Optional.empty()));
            capturedConsumedCallback.getValue().onCompletion(null,
                    new ConsumerRecord<>(TOPIC, 0, 1, 0L, TimestampType.CREATE_TIME, 0, 0,
                            jsonConverter.fromConnectData("", null, Arrays.asList("connector1",
                                    Collections.singletonMap("partitionKey", "partitionValue1"))), TP1_VALUE.array(),
                            new RecordHeaders(), Optional.empty()));
            capturedConsumedCallback.getValue().onCompletion(null,
                    new ConsumerRecord<>(TOPIC, 0, 2, 0L, TimestampType.CREATE_TIME, 0, 0,
                            jsonConverter.fromConnectData("", null, Arrays.asList("connector1",
                                    Collections.singletonMap("partitionKey", "partitionValue2"))), TP2_VALUE.array(),
                            new RecordHeaders(), Optional.empty()));
            capturedConsumedCallback.getValue().onCompletion(null,
                    new ConsumerRecord<>(TOPIC, 0, 3, 0L, TimestampType.CREATE_TIME, 0, 0,
                            jsonConverter.fromConnectData("", null, Arrays.asList("connector2",
                                    Collections.singletonMap("partitionKey", "partitionValue"))), TP1_VALUE.array(),
                            new RecordHeaders(), Optional.empty()));
            storeLogCallbackArgumentCaptor.getValue().onCompletion(null, null);
            return null;
        }).when(storeLog).readToEnd(storeLogCallbackArgumentCaptor.capture());

        // Trigger a read to the end of the log
        store.get(Collections.emptyList()).get(10000, TimeUnit.MILLISECONDS);

        Set<Map<String, Object>> connectorPartitions1 = store.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 = store.connectorPartitions("connector2");
        Set<Map<String, Object>> expectedConnectorPartition2 = Collections.singleton(Collections.singletonMap("partitionKey", "partitionValue"));
        assertEquals(expectedConnectorPartition2, connectorPartitions2);

        doAnswer(invocation -> {
            capturedConsumedCallback.getValue().onCompletion(null,
                    new ConsumerRecord<>(TOPIC, 0, 4, 0L, TimestampType.CREATE_TIME, 0, 0,
                            jsonConverter.fromConnectData("", null, Arrays.asList("connector1",
                                    Collections.singletonMap("partitionKey", "partitionValue1"))), null,
                            new RecordHeaders(), Optional.empty()));
            storeLogCallbackArgumentCaptor.getValue().onCompletion(null, null);
            return null;
        }).when(storeLog).readToEnd(storeLogCallbackArgumentCaptor.capture());

        // Trigger a read to the end of the log
        store.get(Collections.emptyList()).get(10000, TimeUnit.MILLISECONDS);

        // Null valued offset for a partition key should remove that partition for the connector
        connectorPartitions1 = store.connectorPartitions("connector1");
        assertEquals(Collections.singleton(Collections.singletonMap("partitionKey", "partitionValue2")), connectorPartitions1);

        store.stop();
        verify(storeLog).stop();
    }