void doRun() {
        initialize();
        while (!stopping) {
            Throwable failure = externalFailure;
            if (failure != null)
                onFailure(failure);

            TargetState newTargetState;
            Callback<TargetState> stateChangeCallback;
            synchronized (this) {
                newTargetState = pendingTargetStateChange.getAndSet(null);
                stateChangeCallback = pendingStateChangeCallback.getAndSet(null);
            }
            if (newTargetState != null && !stopping) {
                doTransitionTo(newTargetState, stateChangeCallback);
            }
            synchronized (this) {
                if (pendingTargetStateChange.get() != null
                        || (!State.FAILED.equals(state) && externalFailure != null)
                        || stopping
                ) {
                    // An update occurred before we entered the synchronized block; no big deal,
                    // just start the loop again until we've handled everything
                } else {
                    try {
                        wait();
                    } catch (InterruptedException e) {
                        // We'll pick up any potential state changes at the top of the loop
                    }
                }
            }
        }
        doShutdown();
    }
Fail the connector.
@param cause the cause of the failure; if null, the connector will not be failed
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConnector.java