@Test
public void testPutConnectorConfig() throws Exception {

    expectConfigure();
    expectStart(Collections.EMPTY_LIST, Collections.EMPTY_MAP);
    expectConvertWriteAndRead(
            CONNECTOR_CONFIG_KEYS.get(0), KafkaConfigBackingStore.CONNECTOR_CONFIGURATION_V0, CONFIGS_SERIALIZED.get(0),
            "properties", SAMPLE_CONFIGS.get(0));
    configUpdateListener.onConnectorConfigUpdate(CONNECTOR_IDS.get(0));
    EasyMock.expectLastCall();
    expectConvertWriteAndRead(
            CONNECTOR_CONFIG_KEYS.get(1), KafkaConfigBackingStore.CONNECTOR_CONFIGURATION_V0, CONFIGS_SERIALIZED.get(1),
            "properties", SAMPLE_CONFIGS.get(1));
    configUpdateListener.onConnectorConfigUpdate(CONNECTOR_IDS.get(1));
    EasyMock.expectLastCall();
    // Config deletion
    expectConnectorRemoval(CONNECTOR_CONFIG_KEYS.get(1), TARGET_STATE_KEYS.get(1));
    configUpdateListener.onConnectorConfigRemove(CONNECTOR_IDS.get(1));
    EasyMock.expectLastCall();
    expectStop();
    PowerMock.replayAll();
    configStorage.configure(DEFAULT_DISTRIBUTED_CONFIG);
    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)));
    // Writing should block until it is written and read back from Kafka
    configStorage.putConnectorConfig(CONNECTOR_IDS.get(0), SAMPLE_CONFIGS.get(0));
    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)));
    // Second should also block and all configs should still be available
    configStorage.putConnectorConfig(CONNECTOR_IDS.get(1), SAMPLE_CONFIGS.get(1));
    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)));
    // 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)));
    configStorage.stop();
    PowerMock.verifyAll();
}