private void suspend(boolean paused) {
        State newState = paused ? State.PAUSED : State.STOPPED;
        try {
            if (state == newState) {
                // Already in the desired state
                return;
            }

            if (state == State.STARTED) {
                connector.stop();
            }

            if (state == State.FAILED && newState != State.STOPPED) {
                throw new IllegalArgumentException("Cannot transition to non-stopped state when connector has already failed");
            }

            if (paused) {
                statusListener.onPause(connName);
            } else {
                statusListener.onStop(connName);
            }

            this.state = newState;
        } catch (Throwable t) {
            log.error("{} Error while {} connector", this, paused ? "pausing" : "stopping", t);
            if (paused) {
                statusListener.onFailure(connName, t);
                this.state = State.FAILED;
            } else {
                // We say the connector is STOPPED even if it fails at this point
                this.state = State.STOPPED;
                // One more try to make sure the status is updated correctly
                statusListener.onStop(connName);
            }
        }
    }
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