@Test
public void testPutConnectorConfig() throws Exception {

    Map<String, String> connConfig = connectorConfig(CONNECTOR_NAME, BogusSourceConnector.class, false);
    Map<String, String> newConnConfig = new HashMap<>(connConfig);
    newConnConfig.put("foo", "bar");
    Callback<Map<String, String>> connectorConfigCb = PowerMock.createMock(Callback.class);
    Callback<Herder.Created<ConnectorInfo>> putConnectorConfigCb = PowerMock.createMock(Callback.class);
    // Create
    connector = PowerMock.createMock(BogusSourceConnector.class);
    expectAdd(CONNECTOR_NAME, BogusSourceConnector.class, BogusSourceTask.class, false);
    // Should get first config
    connectorConfigCb.onCompletion(null, connConfig);
    EasyMock.expectLastCall();
    // Update config, which requires stopping and restarting
    worker.stopConnector(CONNECTOR_NAME);
    EasyMock.expectLastCall();
    Capture<ConnectorConfig> capturedConfig = EasyMock.newCapture();
    worker.startConnector(EasyMock.capture(capturedConfig), EasyMock.<ConnectorContext>anyObject(),
            EasyMock.eq(herder), EasyMock.eq(TargetState.STARTED));
    EasyMock.expectLastCall();
    EasyMock.expect(worker.isRunning(CONNECTOR_NAME)).andReturn(true);
    // Generate same task config, which should result in no additional action to restart tasks
    EasyMock.expect(worker.connectorTaskConfigs(CONNECTOR_NAME, DEFAULT_MAX_TASKS, null))
            .andReturn(Collections.singletonList(taskConfig(BogusSourceTask.class, false)));
    worker.isSinkConnector(CONNECTOR_NAME);
    EasyMock.expectLastCall().andReturn(false);
    ConnectorInfo newConnInfo = new ConnectorInfo(CONNECTOR_NAME, newConnConfig, Arrays.asList(new ConnectorTaskId(CONNECTOR_NAME, 0)));
    putConnectorConfigCb.onCompletion(null, new Herder.Created<>(false, newConnInfo));
    EasyMock.expectLastCall();
    // Should get new config
    connectorConfigCb.onCompletion(null, newConnConfig);
    EasyMock.expectLastCall();
    PowerMock.replayAll();
    herder.putConnectorConfig(CONNECTOR_NAME, connConfig, false, createCallback);
    herder.connectorConfig(CONNECTOR_NAME, connectorConfigCb);
    herder.putConnectorConfig(CONNECTOR_NAME, newConnConfig, true, putConnectorConfigCb);
    assertEquals("bar", capturedConfig.getValue().originals().get("foo"));
    herder.connectorConfig(CONNECTOR_NAME, connectorConfigCb);
    PowerMock.verifyAll();
}