@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 testCreateConnectorFailedValidation() {
        initialize(false);
        // Basic validation should be performed and return an error, but should still evaluate the connector's config

        Map<String, String> config = connectorConfig(SourceSink.SOURCE);
        config.remove(ConnectorConfig.NAME_CONFIG);

        Connector connectorMock = mock(SourceConnector.class);
        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.newConnector(anyString())).thenReturn(connectorMock);
        when(plugins.connectorLoader(anyString())).thenReturn(pluginLoader);
        when(plugins.withClassLoader(pluginLoader)).thenReturn(loaderSwap);

        when(connectorMock.config()).thenReturn(new ConfigDef());

        ConfigValue validatedValue = new ConfigValue("foo.bar");
        when(connectorMock.validate(config)).thenReturn(new Config(singletonList(validatedValue)));

        herder.putConnectorConfig(CONNECTOR_NAME, config, false, createCallback);

        ExecutionException exception = assertThrows(ExecutionException.class, () -> createCallback.get(WAIT_TIME_MS, TimeUnit.MILLISECONDS));
        if (BadRequestException.class != exception.getCause().getClass()) {
            throw new AssertionError(exception.getCause());
        }
        verify(loaderSwap).close();
    }