@Override
    public void putConnectorConfig(String connector, Map<String, String> properties, TargetState targetState) {
        log.debug("Writing connector configuration for connector '{}'", connector);
        Struct connectConfig = new Struct(CONNECTOR_CONFIGURATION_V0);
        connectConfig.put("properties", properties);
        byte[] serializedConfig = converter.fromConnectData(topic, CONNECTOR_CONFIGURATION_V0, connectConfig);
        try {
            Timer timer = time.timer(READ_WRITE_TOTAL_TIMEOUT_MS);
            List<ProducerKeyValue> keyValues = new ArrayList<>();
            if (targetState != null) {
                log.debug("Writing target state {} for connector {}", targetState, connector);
                keyValues.add(new ProducerKeyValue(TARGET_STATE_KEY(connector), serializeTargetState(targetState)));
            }
            keyValues.add(new ProducerKeyValue(CONNECTOR_KEY(connector), serializedConfig));
            sendPrivileged(keyValues, timer);
            configLog.readToEnd().get(timer.remainingMs(), TimeUnit.MILLISECONDS);
        } catch (InterruptedException | ExecutionException | TimeoutException e) {
            log.error("Failed to write connector configuration to Kafka: ", e);
            throw new ConnectException("Error writing connector configuration to Kafka", e);
        }
    }
@Test
    public void testPutConnectorConfig() throws Exception {
        when(configLog.partitionCount()).thenReturn(1);

        configStorage.setupAndCreateKafkaBasedLog(TOPIC, config);
        verifyConfigure();
        configStorage.start();

        // Null before writing
        ClusterConfigState configState = configStorage.snapshot();
        assertEquals(-1, configState.offset());
        assertNull(configState.connectorConfig(CONNECTOR_IDS.get(0)));
        assertNull(configState.connectorConfig(CONNECTOR_IDS.get(1)));

        String configKey = CONNECTOR_CONFIG_KEYS.get(1);
        String targetStateKey = TARGET_STATE_KEYS.get(1);

        doAnswer(expectReadToEnd(Collections.singletonMap(CONNECTOR_CONFIG_KEYS.get(0), CONFIGS_SERIALIZED.get(0))))
                .doAnswer(expectReadToEnd(Collections.singletonMap(CONNECTOR_CONFIG_KEYS.get(1), CONFIGS_SERIALIZED.get(1))))
                // Config deletion
                .doAnswer(expectReadToEnd(new LinkedHashMap<String, byte[]>() {{
                            put(configKey, null);
                            put(targetStateKey, null);
                        }})
                ).when(configLog).readToEnd();

        // Writing should block until it is written and read back from Kafka
        expectConvertWriteRead(
                CONNECTOR_CONFIG_KEYS.get(0), KafkaConfigBackingStore.CONNECTOR_CONFIGURATION_V0, CONFIGS_SERIALIZED.get(0),
                "properties", SAMPLE_CONFIGS.get(0));

        configStorage.putConnectorConfig(CONNECTOR_IDS.get(0), SAMPLE_CONFIGS.get(0), null);
        configState = configStorage.snapshot();

        assertEquals(1, configState.offset());
        assertEquals(SAMPLE_CONFIGS.get(0), configState.connectorConfig(CONNECTOR_IDS.get(0)));
        assertNull(configState.connectorConfig(CONNECTOR_IDS.get(1)));
        verify(configUpdateListener).onConnectorConfigUpdate(CONNECTOR_IDS.get(0));

        // Second should also block and all configs should still be available
        expectConvertWriteRead(
                CONNECTOR_CONFIG_KEYS.get(1), KafkaConfigBackingStore.CONNECTOR_CONFIGURATION_V0, CONFIGS_SERIALIZED.get(1),
                "properties", SAMPLE_CONFIGS.get(1));

        configStorage.putConnectorConfig(CONNECTOR_IDS.get(1), SAMPLE_CONFIGS.get(1), null);
        configState = configStorage.snapshot();

        assertEquals(2, configState.offset());
        assertEquals(SAMPLE_CONFIGS.get(0), configState.connectorConfig(CONNECTOR_IDS.get(0)));
        assertEquals(SAMPLE_CONFIGS.get(1), configState.connectorConfig(CONNECTOR_IDS.get(1)));
        verify(configUpdateListener).onConnectorConfigUpdate(CONNECTOR_IDS.get(1));

        // Config deletion
        when(producerFuture.get(anyLong(), any(TimeUnit.class))).thenReturn(null);
        when(converter.toConnectData(TOPIC, null)).thenReturn(new SchemaAndValue(null, null));
        when(configLog.sendWithReceipt(AdditionalMatchers.or(Mockito.eq(configKey), Mockito.eq(targetStateKey)),
                Mockito.isNull())).thenReturn(producerFuture);

        // Deletion should remove the second one we added
        configStorage.removeConnectorConfig(CONNECTOR_IDS.get(1));
        configState = configStorage.snapshot();

        assertEquals(4, configState.offset());
        assertEquals(SAMPLE_CONFIGS.get(0), configState.connectorConfig(CONNECTOR_IDS.get(0)));
        assertNull(configState.connectorConfig(CONNECTOR_IDS.get(1)));
        assertNull(configState.targetState(CONNECTOR_IDS.get(1)));
        verify(configUpdateListener).onConnectorConfigRemove(CONNECTOR_IDS.get(1));

        configStorage.stop();
        verify(configLog).stop();
    }