@Override
    public synchronized void putConnectorConfig(String connName,
                                                final Map<String, String> config,
                                                boolean allowReplace,
                                                final Callback<Created<ConnectorInfo>> callback) {
        putConnectorConfig(connName, config, null, allowReplace, callback);
    }
@Test
    public void testCorruptConfig() {
        initialize(false);
        Map<String, String> config = new HashMap<>();
        config.put(ConnectorConfig.NAME_CONFIG, CONNECTOR_NAME);
        config.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, BogusSinkConnector.class.getName());
        config.put(SinkConnectorConfig.TOPICS_CONFIG, TOPICS_LIST_STR);
        Connector connectorMock = mock(SinkConnector.class);
        String error = "This is an error in your config!";
        List<String> errors = new ArrayList<>(singletonList(error));
        String key = "foo.invalid.key";
        when(connectorMock.validate(config)).thenReturn(
            new Config(
                singletonList(new ConfigValue(key, null, Collections.emptyList(), errors))
            )
        );
        ConfigDef configDef = new ConfigDef();
        configDef.define(key, ConfigDef.Type.STRING, ConfigDef.Importance.HIGH, "");
        when(worker.configTransformer()).thenReturn(transformer);
        final ArgumentCaptor<Map<String, String>> configCapture = ArgumentCaptor.forClass(Map.class);
        when(transformer.transform(configCapture.capture())).thenAnswer(invocation -> configCapture.getValue());
        when(worker.getPlugins()).thenReturn(plugins);
        when(plugins.connectorLoader(anyString())).thenReturn(pluginLoader);
        when(plugins.withClassLoader(pluginLoader)).thenReturn(loaderSwap);
        when(plugins.newConnector(anyString())).thenReturn(connectorMock);
        when(connectorMock.config()).thenReturn(configDef);

        herder.putConnectorConfig(CONNECTOR_NAME, config, true, createCallback);
        ExecutionException e = assertThrows(
            ExecutionException.class,
            () -> createCallback.get(WAIT_TIME_MS, TimeUnit.MILLISECONDS),
            "Should have failed to configure connector"
        );
        assertNotNull(e.getCause());
        Throwable cause = e.getCause();
        assertInstanceOf(BadRequestException.class, cause);
        assertEquals(
            cause.getMessage(),
            "Connector configuration is invalid and contains the following 1 error(s):\n" +
                error + "\n" +
                "You can also find the above list of errors at the endpoint `/connector-plugins/{connectorType}/config/validate`"
        );
        verify(loaderSwap).close();
    }