void doShutdown() {
        try {
            TargetState preEmptedState = pendingTargetStateChange.getAndSet(null);
            Callback<TargetState> stateChangeCallback = pendingStateChangeCallback.getAndSet(null);
            if (stateChangeCallback != null) {
                stateChangeCallback.onCompletion(
                        new ConnectException(
                                "Could not begin changing connector state to " + preEmptedState.name()
                                    + " as the connector has been scheduled for shutdown"),
                        null);
            }
            if (state == State.STARTED)
                connector.stop();
            this.state = State.STOPPED;
            statusListener.onShutdown(connName);
            log.info("Completed shutdown for {}", this);
        } catch (Throwable t) {
            log.error("{} Error while shutting down connector", this, t);
            state = State.FAILED;
            statusListener.onFailure(connName, t);
        } finally {
            Utils.closeQuietly(ctx, "connector context for " + connName);
            Utils.closeQuietly(metrics, "connector metrics for " + connName);
            Utils.closeQuietly(offsetStorageReader, "offset reader for " + connName);
            if (offsetStore != null) {
                Utils.closeQuietly(offsetStore::stop, "offset backing store for " + connName);
            }
        }
    }
Stop this connector. This method does not block, it only triggers shutdown. Use
#{@link #awaitShutdown} to block until completion.
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConnector.java