private void doRun() throws InterruptedException {
        try {
            synchronized (this) {
                if (isStopping())
                    return;

                if (targetState == TargetState.PAUSED) {
                    onPause();
                    if (!awaitUnpause()) return;
                }
            }

            doStart();
            execute();
        } catch (Throwable t) {
            failed = true;
            if (cancelled) {
                log.warn("{} After being scheduled for shutdown, the orphan task threw an uncaught exception. A newer instance of this task might be already running", this, t);
            } else if (isStopping()) {
                log.warn("{} After being scheduled for shutdown, task threw an uncaught exception.", this, t);
            } else {
                log.error("{} Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted", this, t);
                throw t;
            }
        } finally {
            doClose();
        }
    }
Remove all metrics published by this task.
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerTask.java