@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 testCreateConnectorWithStoppedInitialState() throws Exception {
        initialize(true);
        Map<String, String> config = connectorConfig(SourceSink.SINK);
        expectConfigValidation(SourceSink.SINK, config);

        // Only the connector should be created; we expect no tasks to be spawned for a connector created with a paused or stopped initial state
        mockStartConnector(config, null, TargetState.STOPPED, null);

        when(worker.isRunning(CONNECTOR_NAME)).thenReturn(true);
        when(herder.connectorType(any())).thenReturn(ConnectorType.SINK);

        herder.putConnectorConfig(CONNECTOR_NAME, config, TargetState.STOPPED, false, createCallback);
        Herder.Created<ConnectorInfo> connectorInfo = createCallback.get(WAIT_TIME_MS, TimeUnit.MILLISECONDS);
        assertEquals(
            new ConnectorInfo(CONNECTOR_NAME, connectorConfig(SourceSink.SINK), Collections.emptyList(), ConnectorType.SINK),
            connectorInfo.result()
        );
        verify(loaderSwap).close();
    }