KAFKA-14783 (KIP-875): New STOPPED state for connectors

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java

@@ -171,6 +171,12 @@ public void onStartup(String connector) {
                 workerId, generation()));
     }
 
+    @Override
+    public void onStop(String connector) {
+        statusBackingStore.put(new ConnectorStatus(connector, AbstractStatus.State.STOPPED,
+                workerId, generation()));
+    }
+
     @Override
     public void onPause(String connector) {
         statusBackingStore.put(new ConnectorStatus(connector, ConnectorStatus.State.PAUSED,

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractStatus.java

@@ -25,8 +25,9 @@ public enum State {
         RUNNING,
         PAUSED,
         FAILED,
-        DESTROYED,
+        DESTROYED, // Never visible to users; destroyed Connector and Task instances are not shown
         RESTARTING,
+        STOPPED, // Only ever visible to users for Connector instances; never for Task instances
     }
 
     private final T id;

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ConnectorStatus.java

@@ -45,6 +45,12 @@ public interface Listener {
          */
         void onFailure(String connector, Throwable cause);
 
+        /**
+         * Invoked when the connector is stopped through the REST API
+         * @param connector The connector name
+         */
+        void onStop(String connector);
+
         /**
          * Invoked when the connector is paused through the REST API
          * @param connector The connector name

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Herder.java

@@ -246,9 +246,23 @@ default void validateConnectorConfig(Map<String, String> connectorConfig, Callba
      */
     void restartConnectorAndTasks(RestartRequest request, Callback<ConnectorStateInfo> cb);
 
+    /**
+     * Stop the connector. This call will asynchronously suspend processing by the connector and
+     * shut down all of its tasks.
+     * @param connector name of the connector
+     * @param cb callback to invoke upon completion
+     */
+    void stopConnector(String connector, Callback<Void> cb);
+
     /**
      * Pause the connector. This call will asynchronously suspend processing by the connector and all
      * of its tasks.
+     * <p>
+     * Note that, unlike {@link #stopConnector(String, Callback)}, tasks for this connector will not
+     * be shut down and none of their resources will be de-allocated. Instead, they will be left in an
+     * "idling" state where no data is polled from them (if source tasks) or given to them (if sink tasks),
+     * but all internal state kept by the tasks and their resources is left intact and ready to begin
+     * processing records again as soon as the connector is {@link #resumeConnector(String) resumed}.
      * @param connector name of the connector
      */
     void pauseConnector(String connector);

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/StateTracker.java

@@ -73,6 +73,7 @@ private static final class StateChange {
         private final long unassignedTotalTimeMs;
         private final long runningTotalTimeMs;
         private final long pausedTotalTimeMs;
+        private final long stoppedTotalTimeMs;
         private final long failedTotalTimeMs;
         private final long destroyedTotalTimeMs;
         private final long restartingTotalTimeMs;
@@ -81,16 +82,17 @@ private static final class StateChange {
          * The initial StateChange instance before any state has changed.
          */
         StateChange() {
-            this(null, 0L, 0L, 0L, 0L, 0L, 0L, 0L);
+            this(null, 0L, 0L, 0L, 0L, 0L, 0L, 0L, 0L);
         }
 
-        StateChange(State state, long startTime, long unassignedTotalTimeMs, long runningTotalTimeMs,
-                            long pausedTotalTimeMs, long failedTotalTimeMs, long destroyedTotalTimeMs, long restartingTotalTimeMs) {
+        StateChange(State state, long startTime, long unassignedTotalTimeMs, long runningTotalTimeMs, long pausedTotalTimeMs,
+                            long stoppedTotalTimeMs, long failedTotalTimeMs, long destroyedTotalTimeMs, long restartingTotalTimeMs) {
             this.state = state;
             this.startTime = startTime;
             this.unassignedTotalTimeMs = unassignedTotalTimeMs;
             this.runningTotalTimeMs = runningTotalTimeMs;
             this.pausedTotalTimeMs = pausedTotalTimeMs;
+            this.stoppedTotalTimeMs  = stoppedTotalTimeMs;
             this.failedTotalTimeMs = failedTotalTimeMs;
             this.destroyedTotalTimeMs = destroyedTotalTimeMs;
             this.restartingTotalTimeMs = restartingTotalTimeMs;
@@ -106,14 +108,15 @@ private static final class StateChange {
          */
         public StateChange newState(State state, long now) {
             if (this.state == null) {
-                return new StateChange(state, now, 0L, 0L, 0L, 0L, 0L, 0L);
+                return new StateChange(state, now, 0L, 0L, 0L, 0L, 0L, 0L, 0L);
             }
             if (state == this.state) {
                 return this;
             }
             long unassignedTime = this.unassignedTotalTimeMs;
             long runningTime = this.runningTotalTimeMs;
             long pausedTime = this.pausedTotalTimeMs;
+            long stoppedTime = this.stoppedTotalTimeMs;
             long failedTime = this.failedTotalTimeMs;
             long destroyedTime = this.destroyedTotalTimeMs;
             long restartingTime = this.restartingTotalTimeMs;
@@ -128,6 +131,9 @@ public StateChange newState(State state, long now) {
                 case PAUSED:
                     pausedTime += duration;
                     break;
+                case STOPPED:
+                    stoppedTime += duration;
+                    break;
                 case FAILED:
                     failedTime += duration;
                     break;
@@ -138,7 +144,7 @@ public StateChange newState(State state, long now) {
                     restartingTime += duration;
                     break;
             }
-            return new StateChange(state, now, unassignedTime, runningTime, pausedTime, failedTime, destroyedTime, restartingTime);
+            return new StateChange(state, now, unassignedTime, runningTime, pausedTime, stoppedTime, failedTime, destroyedTime, restartingTime);
         }
 
         /**
@@ -164,6 +170,9 @@ public double durationRatio(State ratioState, long now) {
                 case PAUSED:
                     durationDesired += pausedTotalTimeMs;
                     break;
+                case STOPPED:
+                    durationDesired += stoppedTotalTimeMs;
+                    break;
                 case FAILED:
                     durationDesired += failedTotalTimeMs;
                     break;

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/TargetState.java

@@ -22,7 +22,10 @@
  * target state is "STARTED." This does not mean it has actually started, just that
  * the Connect framework will attempt to start it after its tasks have been assigned.
  * After the connector has been paused, the target state will change to PAUSED,
- * and all the tasks will stop doing work.
+ * and all the tasks will stop doing work. A target state of STOPPED is similar to
+ * PAUSED, but is also accompanied by a full shutdown of the connector's tasks,
+ * including deallocation of any Kafka clients, SMTs, and other resources brought
+ * up for or by that task.
  * <p>
  * Target states are persisted in the config topic, which is read by all of the
  * workers in the group. When a worker sees a new target state for a connector which
@@ -33,4 +36,5 @@
 public enum TargetState {
     STARTED,
     PAUSED,
+    STOPPED,
 }

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConnector.java

@@ -57,7 +57,8 @@ public class WorkerConnector implements Runnable {
 
     private enum State {
         INIT,    // initial state before startup
-        STOPPED, // the connector has been stopped/paused.
+        PAUSED,  // The connector has been paused.
+        STOPPED, // the connector has been stopped.
         STARTED, // the connector has been started/resumed.
         FAILED,  // the connector has failed (no further transitions are possible after this state)
     }
@@ -186,6 +187,7 @@ private boolean doStart() throws Throwable {
                     return false;
 
                 case INIT:
+                case PAUSED:
                 case STOPPED:
                     connector.start(config);
                     this.state = State.STARTED;
@@ -220,29 +222,40 @@ public boolean isRunning() {
         return state == State.STARTED;
     }
 
-    @SuppressWarnings("fallthrough")
-    private void pause() {
+    private void suspend(boolean paused) {
+        State newState = paused ? State.PAUSED : State.STOPPED;
         try {
-            switch (state) {
-                case STOPPED:
-                    return;
+            if (state == newState) {
+                // Already in the desired state
+                return;
+            }
 
-                case STARTED:
-                    connector.stop();
-                    // fall through
+            if (state == State.STARTED) {
+                connector.stop();
+            }
 
-                case INIT:
-                    statusListener.onPause(connName);
-                    this.state = State.STOPPED;
-                    break;
+            if (state == State.FAILED && newState != State.STOPPED) {
+                throw new IllegalArgumentException("Cannot transition to non-stopped state when connector has already failed");
+            }
 
-                default:
-                    throw new IllegalArgumentException("Cannot pause connector in state " + state);
+            if (paused) {
+                statusListener.onPause(connName);
+            } else {
+                statusListener.onStop(connName);
             }
+
+            this.state = newState;
         } catch (Throwable t) {
-            log.error("{} Error while shutting down connector", this, t);
-            statusListener.onFailure(connName, t);
-            this.state = State.FAILED;
+            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);
+            }
         }
     }
 
@@ -332,7 +345,8 @@ public void transitionTo(TargetState targetState, Callback<TargetState> stateCha
     }
 
     void doTransitionTo(TargetState targetState, Callback<TargetState> stateChangeCallback) {
-        if (state == State.FAILED) {
+        // Edge case: we don't do transitions most of the time if we've already failed, but for the STOPPED state, it's fine
+        if (state == State.FAILED && targetState != TargetState.STOPPED) {
             stateChangeCallback.onCompletion(
                     new ConnectException(this + " Cannot transition connector to " + targetState + " since it has failed"),
                     null);
@@ -354,7 +368,9 @@ void doTransitionTo(TargetState targetState, Callback<TargetState> stateChangeCa
     private void doTransitionTo(TargetState targetState) throws Throwable {
         log.debug("{} Transition connector to {}", this, targetState);
         if (targetState == TargetState.PAUSED) {
-            pause();
+            suspend(true);
+        } else if (targetState == TargetState.STOPPED) {
+            suspend(false);
         } else if (targetState == TargetState.STARTED) {
             if (state == State.INIT)
                 start();
@@ -448,6 +464,16 @@ public void onShutdown(String connector) {
             }
         }
 
+        @Override
+        public void onStop(String connector) {
+            state = AbstractStatus.State.STOPPED;
+            synchronized (this) {
+                if (!cancelled) {
+                    delegate.onStop(connector);
+                }
+            }
+        }
+
         @Override
         public void onPause(String connector) {
             state = AbstractStatus.State.PAUSED;
@@ -502,6 +528,10 @@ boolean isPaused() {
             return state == AbstractStatus.State.PAUSED;
         }
 
+        boolean isStopped() {
+            return state == AbstractStatus.State.STOPPED;
+        }
+
         boolean isFailed() {
             return state == AbstractStatus.State.FAILED;
         }

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerMetricsGroup.java

@@ -128,6 +128,11 @@ public void onStartup(final String connector) {
             delegateListener.onStartup(connector);
         }
 
+        @Override
+        public void onStop(final String connector) {
+            delegateListener.onStop(connector);
+        }
+
         @Override
         public void onPause(final String connector) {
             delegateListener.onPause(connector);

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerTask.java

@@ -169,7 +169,9 @@ protected boolean isFailed() {
     }
 
     protected boolean isStopping() {
-        return stopping;
+        // The target state should never be STOPPED, but if things go wrong and it somehow is,
+        // we handle that identically to a request to shut down the task
+        return stopping || targetState == TargetState.STOPPED;
     }
 
     protected boolean isCancelled() {
@@ -188,7 +190,7 @@ private void doClose() {
     private void doRun() throws InterruptedException {
         try {
             synchronized (this) {
-                if (stopping)
+                if (isStopping())
                     return;
 
                 if (targetState == TargetState.PAUSED) {
@@ -204,7 +206,7 @@ private void doRun() throws InterruptedException {
             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 (stopping) {
+            } 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);
@@ -281,7 +283,7 @@ public boolean shouldPause() {
     protected boolean awaitUnpause() throws InterruptedException {
         synchronized (this) {
             while (targetState == TargetState.PAUSED) {
-                if (stopping)
+                if (isStopping())
                     return false;
                 this.wait();
             }
@@ -291,9 +293,21 @@ protected boolean awaitUnpause() throws InterruptedException {
 
     public void transitionTo(TargetState state) {
         synchronized (this) {
-            // ignore the state change if we are stopping
-            if (stopping)
+            // Ignore the state change if we are stopping.
+            // This has the consequence that, if we ever transition to the STOPPED target state (which
+            // should never happen since whole point of that state is that it comes with a complete
+            // shutdown of all the tasks for the connector), we will never be able to transition out of it.
+            // Since part of transitioning to the STOPPED state is that we shut down the task and all of
+            // its resources (Kafka clients, SMTs, etc.), this is a reasonable way to do things; otherwise,
+            // we'd have to re-instantiate all of those resources to be able to resume (or even just pause)
+            // the task .
+            if (isStopping()) {
+                log.debug("{} Ignoring request to transition stopped task {} to state {}", this, id, state);
                 return;
+            }
+
+            if (targetState == TargetState.STOPPED)
+                log.warn("{} Received unexpected request to transition task {} to state {}; will shut down in response", this, id, TargetState.STOPPED);
 
             this.targetState = state;
             this.notifyAll();