diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java index 62e5011ca6ee2..122383695ff22 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java +++ b/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, diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractStatus.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractStatus.java index c5e07029a424d..76036d610d738 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractStatus.java +++ b/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; diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ConnectorStatus.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ConnectorStatus.java index 6da4a7db0fea8..10ed188cdf883 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ConnectorStatus.java +++ b/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 diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Herder.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Herder.java index 007b0d190c22d..099a012be3fa1 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Herder.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Herder.java @@ -246,9 +246,23 @@ default void validateConnectorConfig(Map connectorConfig, Callba */ void restartConnectorAndTasks(RestartRequest request, Callback 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 cb); + /** * Pause the connector. This call will asynchronously suspend processing by the connector and all * of its tasks. + *

+ * 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); diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/StateTracker.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/StateTracker.java index 297d47384c7fd..23ac3eac32fd7 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/StateTracker.java +++ b/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,7 +108,7 @@ 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; @@ -114,6 +116,7 @@ public StateChange newState(State state, long now) { 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; diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/TargetState.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/TargetState.java index f2486a6a0eb64..07f8372e969e2 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/TargetState.java +++ b/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. *

* 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, } diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConnector.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConnector.java index 50bf3e672c23e..680cb7a6f2910 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConnector.java +++ b/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 stateCha } void doTransitionTo(TargetState targetState, Callback 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 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; } diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerMetricsGroup.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerMetricsGroup.java index f03bc4f2b79b7..35f2f6340590a 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerMetricsGroup.java +++ b/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); diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerTask.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerTask.java index a669f6a040fc2..ce40814a5753f 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerTask.java +++ b/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(); diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java index 89d2bbbefed2b..b5f92386ab053 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java @@ -1094,6 +1094,38 @@ public void putConnectorConfig(final String connName, final Map ); } + @Override + public void stopConnector(final String connName, final Callback callback) { + log.trace("Submitting request to transition connector {} to STOPPED state", connName); + + addRequest( + () -> { + if (!configState.contains(connName)) + throw new NotFoundException("Unknown connector " + connName); + + // We only allow the leader to handle this request since it involves writing task configs to the config topic + if (!isLeader()) { + callback.onCompletion(new NotLeaderException("Only the leader can transition connectors to the STOPPED state.", leaderUrl()), null); + return null; + } + + // We write the task configs first since, if we fail between then and writing the target state, the + // cluster is still kept in a healthy state. A RUNNING connector with zero tasks is acceptable (although, + // if the connector is reassigned during the ensuing rebalance, it is likely that it will immediately generate + // a non-empty set of task configs). A STOPPED connector with a non-empty set of tasks is less acceptable + // and likely to confuse users. + writeTaskConfigs(connName, Collections.emptyList()); + configBackingStore.putTargetState(connName, TargetState.STOPPED); + // Force a read of the new target state for the connector + refreshConfigSnapshot(workerSyncTimeoutMs); + + callback.onCompletion(null, null); + return null; + }, + forwardErrorCallback(callback) + ); + } + @Override public void requestTaskReconfiguration(final String connName) { log.trace("Submitting connector task reconfiguration request {}", connName); @@ -1151,7 +1183,7 @@ public void putTaskConfigs(final String connName, final List else if (!configState.contains(connName)) callback.onCompletion(new NotFoundException("Connector " + connName + " not found"), null); else { - writeToConfigTopicAsLeader(() -> configBackingStore.putTaskConfigs(connName, configs)); + writeTaskConfigs(connName, configs); callback.onCompletion(null, null); } return null; @@ -1963,6 +1995,9 @@ private void reconfigureConnector(final String connName, final Callback cb if (!worker.isRunning(connName)) { log.info("Skipping reconfiguration of connector {} since it is not running", connName); return; + } else if (configState.targetState(connName) != TargetState.STARTED) { + log.info("Skipping reconfiguration of connector {} since its target state is {}", connName, configState.targetState(connName)); + return; } Map configs = configState.connectorConfig(connName); @@ -1975,52 +2010,66 @@ private void reconfigureConnector(final String connName, final Callback cb } final List> taskProps = worker.connectorTaskConfigs(connName, connConfig); - if (taskConfigsChanged(configState, connName, taskProps)) { - List> rawTaskProps = reverseTransform(connName, configState, taskProps); - if (isLeader()) { - writeToConfigTopicAsLeader(() -> configBackingStore.putTaskConfigs(connName, rawTaskProps)); - cb.onCompletion(null, null); - } else if (restClient == null) { - throw new NotLeaderException("This worker is not able to communicate with the leader of the cluster, " - + "which is required for dynamically-reconfiguring connectors. If running MirrorMaker 2 " - + "in dedicated mode, consider enabling inter-worker communication via the " - + "'dedicated.mode.enable.internal.rest' property.", - leaderUrl() - ); - } else { - // We cannot forward the request on the same thread because this reconfiguration can happen as a result of connector - // addition or removal. If we blocked waiting for the response from leader, we may be kicked out of the worker group. - forwardRequestExecutor.submit(() -> { - try { - String leaderUrl = leaderUrl(); - if (Utils.isBlank(leaderUrl)) { - cb.onCompletion(new ConnectException("Request to leader to " + - "reconfigure connector tasks failed " + - "because the URL of the leader's REST interface is empty!"), null); - return; - } - String reconfigUrl = namespacedUrl(leaderUrl) - .path("connectors") - .path(connName) - .path("tasks") - .build() - .toString(); - log.trace("Forwarding task configurations for connector {} to leader", connName); - restClient.httpRequest(reconfigUrl, "POST", null, rawTaskProps, null, sessionKey, requestSignatureAlgorithm); - cb.onCompletion(null, null); - log.trace("Request to leader to reconfigure connector tasks succeeded"); - } catch (ConnectException e) { - log.error("Request to leader to reconfigure connector tasks failed", e); - cb.onCompletion(e, null); - } - }); - } - } + publishConnectorTaskConfigs(connName, taskProps, cb); } catch (Throwable t) { cb.onCompletion(t, null); } } + private void publishConnectorTaskConfigs(String connName, List> taskProps, Callback cb) { + if (!taskConfigsChanged(configState, connName, taskProps)) { + return; + } + + List> rawTaskProps = reverseTransform(connName, configState, taskProps); + if (isLeader()) { + writeTaskConfigs(connName, rawTaskProps); + cb.onCompletion(null, null); + } else if (restClient == null) { + throw new NotLeaderException("This worker is not able to communicate with the leader of the cluster, " + + "which is required for dynamically-reconfiguring connectors. If running MirrorMaker 2 " + + "in dedicated mode, consider enabling inter-worker communication via the " + + "'dedicated.mode.enable.internal.rest' property.", + leaderUrl() + ); + } else { + // We cannot forward the request on the same thread because this reconfiguration can happen as a result of connector + // addition or removal. If we blocked waiting for the response from leader, we may be kicked out of the worker group. + forwardRequestExecutor.submit(() -> { + try { + String leaderUrl = leaderUrl(); + if (Utils.isBlank(leaderUrl)) { + cb.onCompletion(new ConnectException("Request to leader to " + + "reconfigure connector tasks failed " + + "because the URL of the leader's REST interface is empty!"), null); + return; + } + String reconfigUrl = UriBuilder.fromUri(leaderUrl) + .path("connectors") + .path(connName) + .path("tasks") + .build() + .toString(); + log.trace("Forwarding task configurations for connector {} to leader", connName); + restClient.httpRequest(reconfigUrl, "POST", null, rawTaskProps, null, sessionKey, requestSignatureAlgorithm); + cb.onCompletion(null, null); + } catch (ConnectException e) { + log.error("Request to leader to reconfigure connector tasks failed", e); + cb.onCompletion(e, null); + } + }); + } + } + + private void writeTaskConfigs(String connName, List> taskConfigs) { + if (!taskConfigs.isEmpty()) { + if (configState.targetState(connName) == TargetState.STOPPED) + throw new BadRequestException("Cannot submit non-empty set of task configs for stopped connector " + connName); + } + + writeToConfigTopicAsLeader(() -> configBackingStore.putTaskConfigs(connName, taskConfigs)); + } + // Invoked by exactly-once worker source tasks after they have successfully initialized their transactional // producer to ensure that it is still safe to bring up the task private void verifyTaskGenerationAndOwnership(ConnectorTaskId id, int initialTaskGen) { diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorsResource.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorsResource.java index 086b98dc653b9..fd53f5bb1decc 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorsResource.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorsResource.java @@ -266,6 +266,19 @@ public Response restartConnector(final @PathParam("connector") String connector, return Response.accepted().entity(stateInfo).build(); } + @PUT + @Path("/{connector}/stop") + @Operation(summary = "Stop the specified connector", + description = "This operation is idempotent and has no effects if the connector is already stopped") + public void stopConnector( + @PathParam("connector") String connector, + final @Context HttpHeaders headers, + final @Parameter(hidden = true) @QueryParam("forward") Boolean forward) throws Throwable { + FutureCallback cb = new FutureCallback<>(); + herder.stopConnector(connector, cb); + requestHandler.completeOrForwardRequest(cb, "/connectors/" + connector + "/stop", "PUT", headers, null, forward); + } + @PUT @Path("/{connector}/pause") @Operation(summary = "Pause the specified connector", diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/standalone/StandaloneHerder.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/standalone/StandaloneHerder.java index 3b3c3ced06f39..b3c83e3dd789a 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/standalone/StandaloneHerder.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/standalone/StandaloneHerder.java @@ -238,6 +238,17 @@ private synchronized void putConnectorConfig(String connName, } } + @Override + public synchronized void stopConnector(String connName, Callback callback) { + try { + removeConnectorTasks(connName); + configBackingStore.putTargetState(connName, TargetState.STOPPED); + callback.onCompletion(null, null); + } catch (Throwable t) { + callback.onCompletion(t, null); + } + } + @Override public synchronized void requestTaskReconfiguration(String connName) { if (!worker.connectorNames().contains(connName)) { @@ -429,7 +440,10 @@ private void removeConnectorTasks(String connName) { private void updateConnectorTasks(String connName) { if (!worker.isRunning(connName)) { - log.info("Skipping update of connector {} since it is not running", connName); + log.info("Skipping update of tasks for connector {} since it is not running", connName); + return; + } else if (configState.targetState(connName) != TargetState.STARTED) { + log.info("Skipping update of tasks for connector {} since its target state is {}", connName, configState.targetState(connName)); return; } diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java index 9952ef9d66bfb..36b39bdb58cb0 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java @@ -79,6 +79,8 @@ import java.util.concurrent.TimeoutException; import java.util.function.Supplier; +import static org.apache.kafka.connect.runtime.TargetState.PAUSED; +import static org.apache.kafka.connect.runtime.TargetState.STOPPED; import static org.apache.kafka.connect.util.ConnectUtils.className; /** @@ -240,6 +242,10 @@ public static String TASK_COUNT_RECORD_KEY(String connectorName) { public static final Schema TARGET_STATE_V0 = SchemaBuilder.struct() .field("state", Schema.STRING_SCHEMA) .build(); + public static final Schema TARGET_STATE_V1 = SchemaBuilder.struct() + .field("state", Schema.STRING_SCHEMA) + .field("state.v2", Schema.OPTIONAL_STRING_SCHEMA) + .build(); public static final Schema TASK_COUNT_RECORD_V0 = SchemaBuilder.struct() .field("task-count", Schema.INT32_SCHEMA) .build(); @@ -633,9 +639,11 @@ public void refresh(long timeout, TimeUnit unit) throws TimeoutException { */ @Override public void putTargetState(String connector, TargetState state) { - Struct connectTargetState = new Struct(TARGET_STATE_V0); - connectTargetState.put("state", state.name()); - byte[] serializedTargetState = converter.fromConnectData(topic, TARGET_STATE_V0, connectTargetState); + Struct connectTargetState = new Struct(TARGET_STATE_V1); + // Older workers don't support the STOPPED state; fall back on PAUSED + connectTargetState.put("state", state == STOPPED ? PAUSED.name() : state.name()); + connectTargetState.put("state.v2", state.name()); + byte[] serializedTargetState = converter.fromConnectData(topic, TARGET_STATE_V1, connectTargetState); log.debug("Writing target state {} for connector {}", state, connector); try { configLog.send(TARGET_STATE_KEY(connector), serializedTargetState).get(READ_WRITE_TOTAL_TIMEOUT_MS, TimeUnit.MILLISECONDS); @@ -928,11 +936,22 @@ private void processTargetStateRecord(String connectorName, SchemaAndValue value return; } @SuppressWarnings("unchecked") - Object targetState = ((Map) value.value()).get("state"); - if (!(targetState instanceof String)) { - log.error("Invalid data for target state for connector '{}': 'state' field should be a String but is {}", + Map valueMap = (Map) value.value(); + Object targetState = valueMap.get("state.v2"); + if (targetState != null && !(targetState instanceof String)) { + log.error("Invalid data for target state for connector '{}': 'state.v2' field should be a String but is {}", connectorName, className(targetState)); - return; + // We don't return here; it's still possible that there's a value we can use in the older state field + targetState = null; + } + if (targetState == null) { + // This record may have been written by an older worker; fall back on the older state field + targetState = valueMap.get("state"); + if (!(targetState instanceof String)) { + log.error("Invalid data for target state for connector '{}': 'state' field should be a String but is {}", + connectorName, className(targetState)); + return; + } } try { diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectWorkerIntegrationTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectWorkerIntegrationTest.java index 5cd794e7c83b1..65ecaa2b2c810 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectWorkerIntegrationTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectWorkerIntegrationTest.java @@ -325,6 +325,181 @@ public void testSourceTaskNotBlockedOnShutdownWithNonExistentTopic() throws Exce assertTrue("Connector and all tasks were not stopped in time", stopCounter.await(1, TimeUnit.MINUTES)); } + /** + * Verify that the target state (started, paused, stopped) of a connector can be updated, with + * an emphasis on ensuring that the transitions between each state are correct. + *

+ * The transitions we need to cover are: + *

    + *
  1. RUNNING -> PAUSED
    2. + *
  2. RUNNING -> STOPPED
    3. + *
  3. PAUSED -> RUNNING
    4. + *
  4. PAUSED -> STOPPED
    5. + *
  5. STOPPED -> RUNNING
    6. + *
  6. STOPPED -> PAUSED
    7. + *
+ * With some reordering, we can perform each transition just once: + *
    + *
  • Start with RUNNING
    • + *
  • Transition to STOPPED (2)
    • + *
  • Transition to RUNNING (5)
    • + *
  • Transition to PAUSED (1)
    • + *
  • Transition to STOPPED (4)
    • + *
  • Transition to PAUSED (6)
    • + *
  • Transition to RUNNING (3)
    • + *
+ */ + @Test + public void testPauseStopResume() throws Exception { + connect = connectBuilder.build(); + // start the clusters + connect.start(); + + connect.assertions().assertAtLeastNumWorkersAreUp(NUM_WORKERS, + "Initial group of workers did not start in time."); + + // Want to make sure to use multiple tasks + final int numTasks = 4; + Map props = defaultSourceConnectorProps(TOPIC_NAME); + props.put(TASKS_MAX_CONFIG, Integer.toString(numTasks)); + + // Start with RUNNING + connect.configureConnector(CONNECTOR_NAME, props); + connect.assertions().assertConnectorAndExactlyNumTasksAreRunning( + CONNECTOR_NAME, + numTasks, + "Connector tasks did not start in time" + ); + + // Transition to STOPPED + connect.stopConnector(CONNECTOR_NAME); + // Issue a second request to ensure that this operation is idempotent + connect.stopConnector(CONNECTOR_NAME); + connect.assertions().assertConnectorIsStopped( + CONNECTOR_NAME, + "Connector did not stop in time" + ); + + // Transition to RUNNING + connect.resumeConnector(CONNECTOR_NAME); + // Issue a second request to ensure that this operation is idempotent + connect.resumeConnector(CONNECTOR_NAME); + connect.assertions().assertConnectorAndExactlyNumTasksAreRunning( + CONNECTOR_NAME, + numTasks, + "Connector tasks did not resume in time" + ); + + // Transition to PAUSED + connect.pauseConnector(CONNECTOR_NAME); + // Issue a second request to ensure that this operation is idempotent + connect.pauseConnector(CONNECTOR_NAME); + connect.assertions().assertConnectorAndExactlyNumTasksArePaused( + CONNECTOR_NAME, + numTasks, + "Connector did not pause in time" + ); + + // Transition to STOPPED + connect.stopConnector(CONNECTOR_NAME); + connect.assertions().assertConnectorIsStopped( + CONNECTOR_NAME, + "Connector did not stop in time" + ); + + // Transition to PAUSED + connect.pauseConnector(CONNECTOR_NAME); + connect.assertions().assertConnectorAndExactlyNumTasksArePaused( + CONNECTOR_NAME, + 0, + "Connector did not pause in time" + ); + + // Transition to RUNNING + connect.resumeConnector(CONNECTOR_NAME); + connect.assertions().assertConnectorAndExactlyNumTasksAreRunning( + CONNECTOR_NAME, + numTasks, + "Connector tasks did not resume in time" + ); + + // Delete the connector + connect.deleteConnector(CONNECTOR_NAME); + connect.assertions().assertConnectorAndTasksAreNotRunning( + CONNECTOR_NAME, + "Connector tasks were not destroyed in time" + ); + } + + /** + * Test out the {@code STOPPED} state introduced in + * KIP-875, + * with an emphasis on correctly handling errors thrown from the connector. + */ + @Test + public void testStoppedState() throws Exception { + connect = connectBuilder.build(); + // start the clusters + connect.start(); + + connect.assertions().assertAtLeastNumWorkersAreUp(NUM_WORKERS, + "Initial group of workers did not start in time."); + + Map props = defaultSourceConnectorProps(TOPIC_NAME); + // Fail the connector on startup + props.put("connector.start.inject.error", "true"); + + // Start the connector (should fail immediately and generate no tasks) + connect.configureConnector(CONNECTOR_NAME, props); + connect.assertions().assertConnectorIsFailedAndTasksHaveFailed( + CONNECTOR_NAME, + 0, + "Connector should have failed and not generated any tasks" + ); + + // Stopping a failed connector updates its state to STOPPED in the REST API + connect.stopConnector(CONNECTOR_NAME); + connect.assertions().assertConnectorIsStopped( + CONNECTOR_NAME, + "Connector did not stop in time" + ); + + // Can resume a connector after its Connector has failed before shutdown after receiving a stop request + props.remove("connector.start.inject.error"); + connect.configureConnector(CONNECTOR_NAME, props); + connect.resumeConnector(CONNECTOR_NAME); + connect.assertions().assertConnectorAndExactlyNumTasksAreRunning( + CONNECTOR_NAME, + NUM_TASKS, + "Connector or tasks did not start running healthily in time" + ); + + // Fail the connector on shutdown + props.put("connector.stop.inject.error", "true"); + // Stopping a connector that fails during shutdown after receiving a stop request updates its state to STOPPED in the REST API + connect.configureConnector(CONNECTOR_NAME, props); + connect.stopConnector(CONNECTOR_NAME); + connect.assertions().assertConnectorIsStopped( + CONNECTOR_NAME, + "Connector did not stop in time" + ); + + // Can resume a connector after its Connector has failed during shutdown after receiving a stop request + connect.resumeConnector(CONNECTOR_NAME); + connect.assertions().assertConnectorAndExactlyNumTasksAreRunning( + CONNECTOR_NAME, + NUM_TASKS, + "Connector or tasks did not start running healthily in time" + ); + + // Can delete a stopped connector + connect.deleteConnector(CONNECTOR_NAME); + connect.assertions().assertConnectorAndTasksAreNotRunning( + CONNECTOR_NAME, + "Connector and all of its tasks should no longer be running" + ); + } + private Map defaultSourceConnectorProps(String topic) { // setup up props for the source connector Map props = new HashMap<>(); diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectorTopicsIntegrationTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectorTopicsIntegrationTest.java index 8c4e156693121..a8b812f8c31e7 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectorTopicsIntegrationTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectorTopicsIntegrationTest.java @@ -150,7 +150,7 @@ public void testGetActiveTopics() throws InterruptedException { // deleting a connector resets its active topics connect.deleteConnector(BAR_CONNECTOR); - connect.assertions().assertConnectorAndTasksAreStopped(BAR_CONNECTOR, + connect.assertions().assertConnectorAndTasksAreNotRunning(BAR_CONNECTOR, "Connector tasks did not stop in time."); connect.assertions().assertConnectorActiveTopics(BAR_CONNECTOR, Collections.emptyList(), @@ -205,7 +205,7 @@ public void testTopicTrackingResetIsDisabled() throws InterruptedException { // deleting a connector resets its active topics connect.deleteConnector(FOO_CONNECTOR); - connect.assertions().assertConnectorAndTasksAreStopped(FOO_CONNECTOR, + connect.assertions().assertConnectorAndTasksAreNotRunning(FOO_CONNECTOR, "Connector tasks did not stop in time."); connect.assertions().assertConnectorActiveTopics(FOO_CONNECTOR, Collections.emptyList(), diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ErrorHandlingIntegrationTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ErrorHandlingIntegrationTest.java index 5bc5fcdbd255f..7fc038523635e 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ErrorHandlingIntegrationTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ErrorHandlingIntegrationTest.java @@ -178,7 +178,7 @@ public void testSkipRetryAndDLQWithHeaders() throws Exception { } connect.deleteConnector(CONNECTOR_NAME); - connect.assertions().assertConnectorAndTasksAreStopped(CONNECTOR_NAME, + connect.assertions().assertConnectorAndTasksAreNotRunning(CONNECTOR_NAME, "Connector tasks did not stop in time."); } @@ -247,7 +247,7 @@ public void testErrantRecordReporter() throws Exception { ConsumerRecords messages = connect.kafka().consume(EXPECTED_INCORRECT_RECORDS, CONSUME_MAX_DURATION_MS, DLQ_TOPIC); connect.deleteConnector(CONNECTOR_NAME); - connect.assertions().assertConnectorAndTasksAreStopped(CONNECTOR_NAME, + connect.assertions().assertConnectorAndTasksAreNotRunning(CONNECTOR_NAME, "Connector tasks did not stop in time."); } diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSourceConnector.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSourceConnector.java index 1bbff23e95336..cf7a1f408f102 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSourceConnector.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSourceConnector.java @@ -104,6 +104,9 @@ public List> taskConfigs(int maxTasks) { public void stop() { log.info("Stopped {} connector {}", this.getClass().getSimpleName(), connectorName); connectorHandle.recordConnectorStop(); + if (Boolean.parseBoolean(commonConfigs.getOrDefault("connector.stop.inject.error", "false"))) { + throw new RuntimeException("Injecting errors during connector stop"); + } } @Override diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/RebalanceSourceConnectorsIntegrationTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/RebalanceSourceConnectorsIntegrationTest.java index ebc53edee2dfd..04e12ea41e01f 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/RebalanceSourceConnectorsIntegrationTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/RebalanceSourceConnectorsIntegrationTest.java @@ -206,7 +206,7 @@ public void testDeleteConnector() throws Exception { // delete connector connect.deleteConnector(CONNECTOR_NAME + 3); - connect.assertions().assertConnectorAndTasksAreStopped(CONNECTOR_NAME + 3, + connect.assertions().assertConnectorAndTasksAreNotRunning(CONNECTOR_NAME + 3, "Connector tasks did not stop in time."); waitForCondition(this::assertConnectorAndTasksAreUniqueAndBalanced, diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerConnectorTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerConnectorTest.java index e716efc091df8..cf07d5d2e29cb 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerConnectorTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerConnectorTest.java @@ -103,7 +103,7 @@ public void testInitializeFailure() { assertFailedMetric(workerConnector); workerConnector.shutdown(); workerConnector.doShutdown(); - assertStoppedMetric(workerConnector); + assertDestroyedMetric(workerConnector); verifyInitialize(); verify(listener).onFailure(CONNECTOR, exception); @@ -127,7 +127,7 @@ public void testFailureIsFinalState() { assertFailedMetric(workerConnector); workerConnector.shutdown(); workerConnector.doShutdown(); - assertStoppedMetric(workerConnector); + assertDestroyedMetric(workerConnector); verifyInitialize(); verify(listener).onFailure(CONNECTOR, exception); @@ -153,7 +153,7 @@ public void testStartupAndShutdown() { assertRunningMetric(workerConnector); workerConnector.shutdown(); workerConnector.doShutdown(); - assertStoppedMetric(workerConnector); + assertDestroyedMetric(workerConnector); verifyInitialize(); verify(connector).start(CONFIG); @@ -181,7 +181,7 @@ public void testStartupAndPause() { assertPausedMetric(workerConnector); workerConnector.shutdown(); workerConnector.doShutdown(); - assertStoppedMetric(workerConnector); + assertDestroyedMetric(workerConnector); verifyInitialize(); verify(connector).start(CONFIG); @@ -195,6 +195,37 @@ public void testStartupAndPause() { verifyNoMoreInteractions(onStateChange); } + @Test + public void testStartupAndStop() { + connector = sinkConnector; + when(connector.version()).thenReturn(VERSION); + + Callback onStateChange = mockCallback(); + WorkerConnector workerConnector = new WorkerConnector(CONNECTOR, connector, connectorConfig, ctx, metrics, listener, null, null, classLoader); + + workerConnector.initialize(); + assertInitializedSinkMetric(workerConnector); + + workerConnector.doTransitionTo(TargetState.STARTED, onStateChange); + assertRunningMetric(workerConnector); + workerConnector.doTransitionTo(TargetState.STOPPED, onStateChange); + assertStoppedMetric(workerConnector); + workerConnector.shutdown(); + workerConnector.doShutdown(); + assertDestroyedMetric(workerConnector); + + verifyInitialize(); + verify(connector).start(CONFIG); + verify(listener).onStartup(CONNECTOR); + verify(listener).onStop(CONNECTOR); + verifyCleanShutdown(true); + + InOrder inOrder = inOrder(onStateChange); + inOrder.verify(onStateChange).onCompletion(isNull(), eq(TargetState.STARTED)); + inOrder.verify(onStateChange).onCompletion(isNull(), eq(TargetState.STOPPED)); + verifyNoMoreInteractions(onStateChange); + } + @Test public void testOnResume() { connector = sourceConnector; @@ -213,7 +244,7 @@ public void testOnResume() { assertRunningMetric(workerConnector); workerConnector.shutdown(); workerConnector.doShutdown(); - assertStoppedMetric(workerConnector); + assertDestroyedMetric(workerConnector); verifyInitialize(); verify(listener).onPause(CONNECTOR); @@ -241,7 +272,7 @@ public void testStartupPaused() { assertPausedMetric(workerConnector); workerConnector.shutdown(); workerConnector.doShutdown(); - assertStoppedMetric(workerConnector); + assertDestroyedMetric(workerConnector); verifyInitialize(); // connector never gets started @@ -252,6 +283,31 @@ public void testStartupPaused() { verifyNoMoreInteractions(onStateChange); } + @Test + public void testStartupStopped() { + connector = sinkConnector; + when(connector.version()).thenReturn(VERSION); + + Callback onStateChange = mockCallback(); + WorkerConnector workerConnector = new WorkerConnector(CONNECTOR, connector, connectorConfig, ctx, metrics, listener, null, null, classLoader); + + workerConnector.initialize(); + assertInitializedSinkMetric(workerConnector); + workerConnector.doTransitionTo(TargetState.STOPPED, onStateChange); + assertStoppedMetric(workerConnector); + workerConnector.shutdown(); + workerConnector.doShutdown(); + assertDestroyedMetric(workerConnector); + + verifyInitialize(); + // connector never gets started + verify(listener).onStop(CONNECTOR); + verifyCleanShutdown(false); + + verify(onStateChange).onCompletion(isNull(), eq(TargetState.STOPPED)); + verifyNoMoreInteractions(onStateChange); + } + @Test public void testStartupFailure() { RuntimeException exception = new RuntimeException(); @@ -269,7 +325,7 @@ public void testStartupFailure() { assertFailedMetric(workerConnector); workerConnector.shutdown(); workerConnector.doShutdown(); - assertStoppedMetric(workerConnector); + assertDestroyedMetric(workerConnector); verifyInitialize(); verify(connector).start(CONFIG); @@ -280,6 +336,49 @@ public void testStartupFailure() { verifyNoMoreInteractions(onStateChange); } + @Test + public void testStopFailure() { + RuntimeException exception = new RuntimeException(); + connector = sourceConnector; + + when(connector.version()).thenReturn(VERSION); + + // Fail during the first call to stop, then succeed for the next attempt + doThrow(exception).doNothing().when(connector).stop(); + + Callback onFirstStateChange = mockCallback(); + Callback onSecondStateChange = mockCallback(); + Callback onThirdStateChange = mockCallback(); + WorkerConnector workerConnector = new WorkerConnector(CONNECTOR, connector, connectorConfig, ctx, metrics, listener, offsetStorageReader, offsetStore, classLoader); + + workerConnector.initialize(); + assertInitializedSourceMetric(workerConnector); + workerConnector.doTransitionTo(TargetState.STARTED, onFirstStateChange); + assertRunningMetric(workerConnector); + workerConnector.doTransitionTo(TargetState.STOPPED, onSecondStateChange); + assertStoppedMetric(workerConnector); + workerConnector.doTransitionTo(TargetState.STARTED, onThirdStateChange); + assertRunningMetric(workerConnector); + workerConnector.shutdown(); + workerConnector.doShutdown(); + assertDestroyedMetric(workerConnector); + + verifyInitialize(); + verify(connector, times(2)).start(CONFIG); + verify(listener).onStartup(CONNECTOR); + verify(listener).onResume(CONNECTOR); + verify(listener).onStop(CONNECTOR); + verify(onFirstStateChange).onCompletion(isNull(), eq(TargetState.STARTED)); + verifyNoMoreInteractions(onFirstStateChange); + // We swallow failures when transitioning to the STOPPED state + verify(onSecondStateChange).onCompletion(isNull(), eq(TargetState.STOPPED)); + verifyNoMoreInteractions(onSecondStateChange); + verify(onThirdStateChange).onCompletion(isNull(), eq(TargetState.STARTED)); + verifyNoMoreInteractions(onThirdStateChange); + verifyShutdown(2, true, true); + verifyNoMoreInteractions(listener); + } + @Test public void testShutdownFailure() { RuntimeException exception = new RuntimeException(); @@ -327,7 +426,7 @@ public void testTransitionStartedToStarted() { assertRunningMetric(workerConnector); workerConnector.shutdown(); workerConnector.doShutdown(); - assertStoppedMetric(workerConnector); + assertDestroyedMetric(workerConnector); verifyInitialize(); verify(connector).start(CONFIG); @@ -356,7 +455,7 @@ public void testTransitionPausedToPaused() { assertPausedMetric(workerConnector); workerConnector.shutdown(); workerConnector.doShutdown(); - assertStoppedMetric(workerConnector); + assertDestroyedMetric(workerConnector); verifyInitialize(); verify(connector).start(CONFIG); @@ -370,6 +469,38 @@ public void testTransitionPausedToPaused() { verifyNoMoreInteractions(onStateChange); } + @Test + public void testTransitionStoppedToStopped() { + connector = sourceConnector; + when(connector.version()).thenReturn(VERSION); + + Callback onStateChange = mockCallback(); + WorkerConnector workerConnector = new WorkerConnector(CONNECTOR, connector, connectorConfig, ctx, metrics, listener, offsetStorageReader, offsetStore, classLoader); + + workerConnector.initialize(); + assertInitializedSourceMetric(workerConnector); + workerConnector.doTransitionTo(TargetState.STARTED, onStateChange); + assertRunningMetric(workerConnector); + workerConnector.doTransitionTo(TargetState.STOPPED, onStateChange); + assertStoppedMetric(workerConnector); + workerConnector.doTransitionTo(TargetState.STOPPED, onStateChange); + assertStoppedMetric(workerConnector); + workerConnector.shutdown(); + workerConnector.doShutdown(); + assertDestroyedMetric(workerConnector); + + verifyInitialize(); + verify(connector).start(CONFIG); + verify(listener).onStartup(CONNECTOR); + verify(listener).onStop(CONNECTOR); + verifyCleanShutdown(true); + + InOrder inOrder = inOrder(onStateChange); + inOrder.verify(onStateChange).onCompletion(isNull(), eq(TargetState.STARTED)); + inOrder.verify(onStateChange, times(2)).onCompletion(isNull(), eq(TargetState.STOPPED)); + verifyNoMoreInteractions(onStateChange); + } + @Test public void testFailConnectorThatIsNeitherSourceNorSink() { connector = mock(Connector.class); @@ -389,13 +520,23 @@ public void testFailConnectorThatIsNeitherSourceNorSink() { protected void assertFailedMetric(WorkerConnector workerConnector) { assertFalse(workerConnector.metrics().isUnassigned()); assertTrue(workerConnector.metrics().isFailed()); + assertFalse(workerConnector.metrics().isStopped()); + assertFalse(workerConnector.metrics().isPaused()); + assertFalse(workerConnector.metrics().isRunning()); + } + + protected void assertStoppedMetric(WorkerConnector workerConnector) { + assertFalse(workerConnector.metrics().isUnassigned()); + assertFalse(workerConnector.metrics().isFailed()); assertFalse(workerConnector.metrics().isPaused()); + assertTrue(workerConnector.metrics().isStopped()); assertFalse(workerConnector.metrics().isRunning()); } protected void assertPausedMetric(WorkerConnector workerConnector) { assertFalse(workerConnector.metrics().isUnassigned()); assertFalse(workerConnector.metrics().isFailed()); + assertFalse(workerConnector.metrics().isStopped()); assertTrue(workerConnector.metrics().isPaused()); assertFalse(workerConnector.metrics().isRunning()); } @@ -403,13 +544,15 @@ protected void assertPausedMetric(WorkerConnector workerConnector) { protected void assertRunningMetric(WorkerConnector workerConnector) { assertFalse(workerConnector.metrics().isUnassigned()); assertFalse(workerConnector.metrics().isFailed()); + assertFalse(workerConnector.metrics().isStopped()); assertFalse(workerConnector.metrics().isPaused()); assertTrue(workerConnector.metrics().isRunning()); } - protected void assertStoppedMetric(WorkerConnector workerConnector) { + protected void assertDestroyedMetric(WorkerConnector workerConnector) { assertTrue(workerConnector.metrics().isUnassigned()); assertFalse(workerConnector.metrics().isFailed()); + assertFalse(workerConnector.metrics().isStopped()); assertFalse(workerConnector.metrics().isPaused()); assertFalse(workerConnector.metrics().isRunning()); } @@ -425,6 +568,7 @@ protected void assertInitializedSourceMetric(WorkerConnector workerConnector) { protected void assertInitializedMetric(WorkerConnector workerConnector, String expectedType) { assertTrue(workerConnector.metrics().isUnassigned()); assertFalse(workerConnector.metrics().isFailed()); + assertFalse(workerConnector.metrics().isStopped()); assertFalse(workerConnector.metrics().isPaused()); assertFalse(workerConnector.metrics().isRunning()); MetricGroup metricGroup = workerConnector.metrics().metricGroup(); @@ -457,6 +601,10 @@ private void verifyCleanShutdown(boolean started) { } private void verifyShutdown(boolean clean, boolean started) { + verifyShutdown(1, clean, started); + } + + private void verifyShutdown(int connectorStops, boolean clean, boolean started) { verify(ctx).close(); if (connector instanceof SourceConnector) { verify(offsetStorageReader).close(); @@ -466,7 +614,7 @@ private void verifyShutdown(boolean clean, boolean started) { verify(listener).onShutdown(CONNECTOR); } if (started) { - verify(connector).stop(); + verify(connector, times(connectorStops)).stop(); } } diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java index 7ad75ae0b1b7f..33b8f966e2edf 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java @@ -217,6 +217,17 @@ public class DistributedHerderTest { Collections.emptyMap(), Collections.emptySet(), Collections.emptySet()); + private static final ClusterConfigState SNAPSHOT_STOPPED_CONN1 = new ClusterConfigState( + 1, + null, + Collections.singletonMap(CONN1, 3), + Collections.singletonMap(CONN1, CONN1_CONFIG), + Collections.singletonMap(CONN1, TargetState.STOPPED), + Collections.emptyMap(), // Stopped connectors should have an empty set of task configs + Collections.emptyMap(), + Collections.emptyMap(), + Collections.emptySet(), + Collections.emptySet()); private static final ClusterConfigState SNAPSHOT_UPDATED_CONN1_CONFIG = new ClusterConfigState( 1, null, @@ -2161,6 +2172,65 @@ public void testConnectorResumed() throws Exception { PowerMock.verifyAll(); } + @Test + public void testConnectorStopped() throws Exception { + // ensure that target state changes are propagated to the worker + + EasyMock.expect(member.memberId()).andStubReturn("member"); + EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0); + EasyMock.expect(worker.connectorNames()).andStubReturn(Collections.singleton(CONN1)); + + // join + expectRebalance(1, Arrays.asList(CONN1), Collections.emptyList()); + expectConfigRefreshAndSnapshot(SNAPSHOT); + EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0); + Capture> onStart = newCapture(); + worker.startConnector(EasyMock.eq(CONN1), EasyMock.anyObject(), EasyMock.anyObject(), + EasyMock.eq(herder), EasyMock.eq(TargetState.STARTED), capture(onStart)); + PowerMock.expectLastCall().andAnswer(() -> { + onStart.getValue().onCompletion(null, TargetState.STARTED); + return true; + }); + member.wakeup(); + PowerMock.expectLastCall(); + expectExecuteTaskReconfiguration(true, conn1SinkConfig, () -> TASK_CONFIGS); + member.poll(EasyMock.anyInt()); + PowerMock.expectLastCall(); + + // handle the state change + member.wakeup(); + PowerMock.expectLastCall(); + member.ensureActive(); + PowerMock.expectLastCall(); + + EasyMock.expect(configBackingStore.snapshot()).andReturn(SNAPSHOT_STOPPED_CONN1); + + Capture> onStop = newCapture(); + worker.setTargetState(EasyMock.eq(CONN1), EasyMock.eq(TargetState.STOPPED), capture(onStop)); + PowerMock.expectLastCall().andAnswer(() -> { + onStart.getValue().onCompletion(null, TargetState.STOPPED); + return null; + }); + + member.poll(EasyMock.anyInt()); + PowerMock.expectLastCall(); + + // These will occur just before/during the third tick + member.ensureActive(); + PowerMock.expectLastCall(); + member.poll(EasyMock.anyInt()); + PowerMock.expectLastCall(); + + PowerMock.replayAll(); + + herder.tick(); // join + configUpdateListener.onConnectorTargetStateChange(CONN1); // state changes to stopped + herder.tick(); // worker should apply the state change + herder.tick(); + + PowerMock.verifyAll(); + } + @Test public void testUnknownConnectorPaused() throws Exception { EasyMock.expect(member.memberId()).andStubReturn("member"); @@ -2197,6 +2267,139 @@ public void testUnknownConnectorPaused() throws Exception { PowerMock.verifyAll(); } + @Test + public void testStopConnector() throws Exception { + EasyMock.expect(member.memberId()).andStubReturn("leader"); + EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0); + EasyMock.expect(worker.connectorNames()).andStubReturn(Collections.singleton(CONN1)); + + // join as leader + expectRebalance(1, Collections.emptyList(), singletonList(TASK0), true); + expectConfigRefreshAndSnapshot(SNAPSHOT); + worker.startSourceTask(EasyMock.eq(TASK0), EasyMock.anyObject(), EasyMock.anyObject(), EasyMock.anyObject(), + EasyMock.eq(herder), EasyMock.eq(TargetState.STARTED)); + PowerMock.expectLastCall().andReturn(true); + member.poll(EasyMock.anyInt()); + PowerMock.expectLastCall(); + + // handle stop request + member.wakeup(); + PowerMock.expectLastCall(); + member.ensureActive(); + PowerMock.expectLastCall(); + expectConfigRefreshAndSnapshot(SNAPSHOT); + configBackingStore.putTaskConfigs(CONN1, Collections.emptyList()); + PowerMock.expectLastCall(); + configBackingStore.putTargetState(CONN1, TargetState.STOPPED); + PowerMock.expectLastCall(); + member.poll(EasyMock.anyInt()); + PowerMock.expectLastCall(); + + PowerMock.replayAll(); + + FutureCallback cb = new FutureCallback<>(); + + herder.tick(); // join + herder.stopConnector(CONN1, cb); // external request + herder.tick(); // continue + + assertTrue("Callback should already have been invoked by herder", cb.isDone()); + cb.get(0, TimeUnit.MILLISECONDS); + + PowerMock.verifyAll(); + } + + @Test + public void testStopConnectorNotLeader() throws Exception { + EasyMock.expect(member.memberId()).andStubReturn("member"); + EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0); + EasyMock.expect(worker.connectorNames()).andStubReturn(Collections.singleton(CONN1)); + + // join as leader + expectRebalance(1, Collections.emptyList(), singletonList(TASK0)); + expectConfigRefreshAndSnapshot(SNAPSHOT); + worker.startSourceTask(EasyMock.eq(TASK0), EasyMock.anyObject(), EasyMock.anyObject(), EasyMock.anyObject(), + EasyMock.eq(herder), EasyMock.eq(TargetState.STARTED)); + PowerMock.expectLastCall().andReturn(true); + member.poll(EasyMock.anyInt()); + PowerMock.expectLastCall(); + + // handle stop request + member.wakeup(); + PowerMock.expectLastCall(); + member.ensureActive(); + PowerMock.expectLastCall(); + member.poll(EasyMock.anyInt()); + PowerMock.expectLastCall(); + + PowerMock.replayAll(); + + FutureCallback cb = new FutureCallback<>(); + + herder.tick(); // join + herder.stopConnector(CONN1, cb); // external request + herder.tick(); // continue + + assertTrue("Callback should already have been invoked by herder", cb.isDone()); + ExecutionException e = assertThrows( + "Should not be able to handle request to stop connector when not leader", + ExecutionException.class, + () -> cb.get(0, TimeUnit.SECONDS) + ); + assertTrue(e.getCause() instanceof NotLeaderException); + + PowerMock.verifyAll(); + } + + @Test + public void testStopConnectorFailToWriteTaskConfigs() throws Exception { + EasyMock.expect(member.memberId()).andStubReturn("leader"); + EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0); + EasyMock.expect(worker.connectorNames()).andStubReturn(Collections.singleton(CONN1)); + + // join as leader + expectRebalance(1, Collections.emptyList(), singletonList(TASK0), true); + expectConfigRefreshAndSnapshot(SNAPSHOT); + worker.startSourceTask(EasyMock.eq(TASK0), EasyMock.anyObject(), EasyMock.anyObject(), EasyMock.anyObject(), + EasyMock.eq(herder), EasyMock.eq(TargetState.STARTED)); + PowerMock.expectLastCall().andReturn(true); + member.poll(EasyMock.anyInt()); + PowerMock.expectLastCall(); + + ConnectException taskConfigsWriteException = new ConnectException("Could not write task configs to config topic"); + + // handle stop request + member.wakeup(); + PowerMock.expectLastCall(); + member.ensureActive(); + PowerMock.expectLastCall(); + configBackingStore.putTaskConfigs(CONN1, Collections.emptyList()); + // We do not expect configBackingStore::putTargetState to be invoked, which + // is intentional since that call should only take place if we are first able to + // successfully write the empty list of task configs + PowerMock.expectLastCall().andThrow(taskConfigsWriteException); + member.poll(EasyMock.anyInt()); + PowerMock.expectLastCall(); + + PowerMock.replayAll(); + + FutureCallback cb = new FutureCallback<>(); + + herder.tick(); // join + herder.stopConnector(CONN1, cb); // external request + herder.tick(); // continue + + assertTrue("Callback should already have been invoked by herder", cb.isDone()); + ExecutionException e = assertThrows( + "Should not be able to handle request to stop connector when not leader", + ExecutionException.class, + () -> cb.get(0, TimeUnit.SECONDS) + ); + assertEquals(e.getCause(), taskConfigsWriteException); + + PowerMock.verifyAll(); + } + @Test public void testConnectorPausedRunningTaskOnly() throws Exception { // even if we don't own the connector, we should still propagate target state diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/standalone/StandaloneHerderTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/standalone/StandaloneHerderTest.java index b76d8f8382431..9eb29d5de5345 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/standalone/StandaloneHerderTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/standalone/StandaloneHerderTest.java @@ -86,6 +86,7 @@ import static org.apache.kafka.connect.runtime.TopicCreationConfig.DEFAULT_TOPIC_CREATION_PREFIX; import static org.apache.kafka.connect.runtime.TopicCreationConfig.PARTITIONS_CONFIG; import static org.apache.kafka.connect.runtime.TopicCreationConfig.REPLICATION_FACTOR_CONFIG; +import static org.easymock.EasyMock.capture; import static org.easymock.EasyMock.eq; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; @@ -733,6 +734,7 @@ public void testCreateAndStop() throws Exception { expectStop(); statusBackingStore.put(new TaskStatus(new ConnectorTaskId(CONNECTOR_NAME, 0), AbstractStatus.State.DESTROYED, WORKER_ID, 0)); + EasyMock.expectLastCall(); statusBackingStore.stop(); EasyMock.expectLastCall(); @@ -937,6 +939,50 @@ public void testCorruptConfig() throws Throwable { PowerMock.verifyAll(); } + @Test + public void testTargetStates() throws Exception { + connector = PowerMock.createMock(BogusSourceConnector.class); + expectAdd(SourceSink.SOURCE); + + Map connectorConfig = connectorConfig(SourceSink.SOURCE); + Connector connectorMock = PowerMock.createMock(SourceConnector.class); + expectConfigValidation(connectorMock, true, connectorConfig); + + // We pause, then stop, the connector + expectTargetState(CONNECTOR_NAME, TargetState.PAUSED); + expectTargetState(CONNECTOR_NAME, TargetState.STOPPED); + + // herder.stop() should stop any running connectors and tasks even if destroyConnector was not invoked + expectStop(); + + statusBackingStore.put(new TaskStatus(new ConnectorTaskId(CONNECTOR_NAME, 0), AbstractStatus.State.DESTROYED, WORKER_ID, 0)); + EasyMock.expectLastCall(); + + statusBackingStore.stop(); + EasyMock.expectLastCall(); + worker.stop(); + EasyMock.expectLastCall(); + + PowerMock.replayAll(); + + FutureCallback stopCallback = new FutureCallback<>(); + FutureCallback> taskConfigsCallback = new FutureCallback<>(); + + herder.putConnectorConfig(CONNECTOR_NAME, connectorConfig, false, createCallback); + Herder.Created connectorInfo = createCallback.get(1000L, TimeUnit.SECONDS); + assertEquals(createdInfo(SourceSink.SOURCE), connectorInfo.result()); + herder.pauseConnector(CONNECTOR_NAME); + herder.stopConnector(CONNECTOR_NAME, stopCallback); + stopCallback.get(10L, TimeUnit.SECONDS); + herder.taskConfigs(CONNECTOR_NAME, taskConfigsCallback); + assertEquals(Collections.emptyList(), taskConfigsCallback.get(1, TimeUnit.SECONDS)); + + herder.stop(); + assertTrue(noneConnectorClientConfigOverridePolicy.isClosed()); + + PowerMock.verifyAll(); + } + private void expectAdd(SourceSink sourceSink) { Map connectorProps = connectorConfig(sourceSink); ConnectorConfig connConfig = sourceSink == SourceSink.SOURCE ? @@ -996,6 +1042,15 @@ private void expectAdd(SourceSink sourceSink) { PowerMock.expectLastCall().andReturn(sourceSink == SourceSink.SINK).anyTimes(); } + private void expectTargetState(String connector, TargetState state) { + Capture> stateChangeCallback = Capture.newInstance(); + worker.setTargetState(eq(connector), eq(state), capture(stateChangeCallback)); + EasyMock.expectLastCall().andAnswer(() -> { + stateChangeCallback.getValue().onCompletion(null, state); + return null; + }); + } + private ConnectorInfo createdInfo(SourceSink sourceSink) { return new ConnectorInfo(CONNECTOR_NAME, connectorConfig(sourceSink), Arrays.asList(new ConnectorTaskId(CONNECTOR_NAME, 0)), diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaConfigBackingStoreTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaConfigBackingStoreTest.java index 63fe460685f85..f06bdee904106 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaConfigBackingStoreTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaConfigBackingStoreTest.java @@ -61,6 +61,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.LinkedHashMap; import java.util.List; import java.util.Locale; @@ -145,8 +146,15 @@ public class KafkaConfigBackingStoreTest { new Struct(KafkaConfigBackingStore.TASK_COUNT_RECORD_V0).put("task-count", 6), new Struct(KafkaConfigBackingStore.TASK_COUNT_RECORD_V0).put("task-count", 9) ); - private static final Struct TARGET_STATE_PAUSED = new Struct(KafkaConfigBackingStore.TARGET_STATE_V0).put("state", "PAUSED"); private static final Struct TARGET_STATE_STARTED = new Struct(KafkaConfigBackingStore.TARGET_STATE_V0).put("state", "STARTED"); + private static final Struct TARGET_STATE_PAUSED_LEGACY = new Struct(KafkaConfigBackingStore.TARGET_STATE_V0) + .put("state", "PAUSED"); + private static final Struct TARGET_STATE_PAUSED = new Struct(KafkaConfigBackingStore.TARGET_STATE_V1) + .put("state", "PAUSED") + .put("state.v2", "PAUSED"); + private static final Struct TARGET_STATE_STOPPED = new Struct(KafkaConfigBackingStore.TARGET_STATE_V1) + .put("state", "PAUSED") + .put("state.v2", "STOPPED"); private static final Struct TASKS_COMMIT_STRUCT_TWO_TASK_CONNECTOR = new Struct(KafkaConfigBackingStore.CONNECTOR_TASKS_COMMIT_V0).put("tasks", 2); @@ -460,7 +468,7 @@ public void testWritePrivileges() throws Exception { expectConvert(KafkaConfigBackingStore.CONNECTOR_CONFIGURATION_V0, CONNECTOR_CONFIG_STRUCTS.get(0), CONFIGS_SERIALIZED.get(1)); // In the meantime, write a target state (which doesn't require write privileges) - expectConvert(KafkaConfigBackingStore.TARGET_STATE_V0, TARGET_STATE_PAUSED, CONFIGS_SERIALIZED.get(1)); + expectConvert(KafkaConfigBackingStore.TARGET_STATE_V1, TARGET_STATE_PAUSED, CONFIGS_SERIALIZED.get(1)); storeLog.send("target-state-" + CONNECTOR_IDS.get(1), CONFIGS_SERIALIZED.get(1)); EasyMock.expectLastCall().andReturn(producerFuture); producerFuture.get(EasyMock.anyLong(), EasyMock.anyObject()); @@ -811,14 +819,18 @@ public void testRestoreTargetState() throws Exception { new ConsumerRecord<>(TOPIC, 0, 3, 0L, TimestampType.CREATE_TIME, 0, 0, TARGET_STATE_KEYS.get(0), CONFIGS_SERIALIZED.get(3), new RecordHeaders(), Optional.empty()), new ConsumerRecord<>(TOPIC, 0, 4, 0L, TimestampType.CREATE_TIME, 0, 0, COMMIT_TASKS_CONFIG_KEYS.get(0), - CONFIGS_SERIALIZED.get(4), new RecordHeaders(), Optional.empty())); + CONFIGS_SERIALIZED.get(4), new RecordHeaders(), Optional.empty()), + new ConsumerRecord<>(TOPIC, 0, 5, 0L, TimestampType.CREATE_TIME, 0, 0, TARGET_STATE_KEYS.get(1), + CONFIGS_SERIALIZED.get(5), new RecordHeaders(), Optional.empty())); LinkedHashMap deserialized = new LinkedHashMap<>(); deserialized.put(CONFIGS_SERIALIZED.get(0), CONNECTOR_CONFIG_STRUCTS.get(0)); deserialized.put(CONFIGS_SERIALIZED.get(1), TASK_CONFIG_STRUCTS.get(0)); deserialized.put(CONFIGS_SERIALIZED.get(2), TASK_CONFIG_STRUCTS.get(0)); - deserialized.put(CONFIGS_SERIALIZED.get(3), TARGET_STATE_PAUSED); + // A worker running an older version wrote this target state; make sure we can handle it correctly + deserialized.put(CONFIGS_SERIALIZED.get(3), TARGET_STATE_PAUSED_LEGACY); deserialized.put(CONFIGS_SERIALIZED.get(4), TASKS_COMMIT_STRUCT_TWO_TASK_CONNECTOR); - logOffset = 5; + deserialized.put(CONFIGS_SERIALIZED.get(5), TARGET_STATE_STOPPED); + logOffset = 6; expectStart(existingRecords, deserialized); @@ -834,9 +846,10 @@ public void testRestoreTargetState() throws Exception { // Should see a single connector with initial state paused ClusterConfigState configState = configStorage.snapshot(); - assertEquals(5, configState.offset()); // Should always be next to be read, even if uncommitted + assertEquals(6, configState.offset()); // Should always be next to be read, even if uncommitted assertEquals(Arrays.asList(CONNECTOR_IDS.get(0)), new ArrayList<>(configState.connectors())); assertEquals(TargetState.PAUSED, configState.targetState(CONNECTOR_IDS.get(0))); + assertEquals(TargetState.STOPPED, configState.targetState(CONNECTOR_IDS.get(1))); configStorage.stop(); @@ -857,18 +870,27 @@ public void testBackgroundUpdateTargetState() throws Exception { CONFIGS_SERIALIZED.get(2), new RecordHeaders(), Optional.empty()), new ConsumerRecord<>(TOPIC, 0, 3, 0L, TimestampType.CREATE_TIME, 0, 0, COMMIT_TASKS_CONFIG_KEYS.get(0), CONFIGS_SERIALIZED.get(3), new RecordHeaders(), Optional.empty())); - LinkedHashMap deserialized = new LinkedHashMap<>(); - deserialized.put(CONFIGS_SERIALIZED.get(0), CONNECTOR_CONFIG_STRUCTS.get(0)); - deserialized.put(CONFIGS_SERIALIZED.get(1), TASK_CONFIG_STRUCTS.get(0)); - deserialized.put(CONFIGS_SERIALIZED.get(2), TASK_CONFIG_STRUCTS.get(0)); - deserialized.put(CONFIGS_SERIALIZED.get(3), TASKS_COMMIT_STRUCT_TWO_TASK_CONNECTOR); + LinkedHashMap deserializedOnStartup = new LinkedHashMap<>(); + deserializedOnStartup.put(CONFIGS_SERIALIZED.get(0), CONNECTOR_CONFIG_STRUCTS.get(0)); + deserializedOnStartup.put(CONFIGS_SERIALIZED.get(1), TASK_CONFIG_STRUCTS.get(0)); + deserializedOnStartup.put(CONFIGS_SERIALIZED.get(2), TASK_CONFIG_STRUCTS.get(0)); + deserializedOnStartup.put(CONFIGS_SERIALIZED.get(3), TASKS_COMMIT_STRUCT_TWO_TASK_CONNECTOR); logOffset = 5; - expectStart(existingRecords, deserialized); + expectStart(existingRecords, deserializedOnStartup); + + LinkedHashMap serializedAfterStartup = new LinkedHashMap<>(); + serializedAfterStartup.put(TARGET_STATE_KEYS.get(0), CONFIGS_SERIALIZED.get(0)); + serializedAfterStartup.put(TARGET_STATE_KEYS.get(1), CONFIGS_SERIALIZED.get(1)); - expectRead(TARGET_STATE_KEYS.get(0), CONFIGS_SERIALIZED.get(0), TARGET_STATE_PAUSED); + Map deserializedAfterStartup = new HashMap<>(); + deserializedAfterStartup.put(TARGET_STATE_KEYS.get(0), TARGET_STATE_PAUSED); + deserializedAfterStartup.put(TARGET_STATE_KEYS.get(1), TARGET_STATE_STOPPED); + + expectRead(serializedAfterStartup, deserializedAfterStartup); configUpdateListener.onConnectorTargetStateChange(CONNECTOR_IDS.get(0)); + configUpdateListener.onConnectorTargetStateChange(CONNECTOR_IDS.get(1)); EasyMock.expectLastCall(); expectPartitionCount(1); @@ -879,11 +901,17 @@ public void testBackgroundUpdateTargetState() throws Exception { configStorage.setupAndCreateKafkaBasedLog(TOPIC, config); configStorage.start(); - // Should see a single connector with initial state paused + // Should see a single connector with initial state started ClusterConfigState configState = configStorage.snapshot(); + assertEquals(Collections.singleton(CONNECTOR_IDS.get(0)), configStorage.connectorTargetStates.keySet()); assertEquals(TargetState.STARTED, configState.targetState(CONNECTOR_IDS.get(0))); + // Should see two connectors now, one paused and one stopped configStorage.refresh(0, TimeUnit.SECONDS); + configState = configStorage.snapshot(); + assertEquals(new HashSet<>(CONNECTOR_IDS), configStorage.connectorTargetStates.keySet()); + assertEquals(TargetState.PAUSED, configState.targetState(CONNECTOR_IDS.get(0))); + assertEquals(TargetState.STOPPED, configState.targetState(CONNECTOR_IDS.get(1))); configStorage.stop(); diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedConnectCluster.java b/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedConnectCluster.java index 97f8581e9d999..e849476f457c8 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedConnectCluster.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedConnectCluster.java @@ -24,6 +24,7 @@ import org.apache.kafka.connect.runtime.rest.entities.ActiveTopicsInfo; import org.apache.kafka.connect.runtime.rest.entities.ConfigInfos; import org.apache.kafka.connect.runtime.rest.entities.ConnectorOffsets; +import org.apache.kafka.connect.runtime.rest.entities.ConnectorInfo; import org.apache.kafka.connect.runtime.rest.entities.ConnectorStateInfo; import org.apache.kafka.connect.runtime.rest.entities.ServerInfo; import org.apache.kafka.connect.runtime.rest.errors.ConnectRestException; @@ -391,6 +392,22 @@ public void deleteConnector(String connName) { } } + /** + * Stop an existing connector. + * + * @param connName name of the connector to be paused + * @throws ConnectRestException if the REST API returns error status + * @throws ConnectException for any other error. + */ + public void stopConnector(String connName) { + String url = endpointForResource(String.format("connectors/%s/stop", connName)); + Response response = requestPut(url, ""); + if (response.getStatus() >= Response.Status.BAD_REQUEST.getStatusCode()) { + throw new ConnectRestException(response.getStatus(), + "Could not execute PUT request. Error response: " + responseToString(response)); + } + } + /** * Pause an existing connector. * @@ -554,6 +571,54 @@ public ActiveTopicsInfo connectorTopics(String connectorName) { "Could not read connector state. Error response: " + responseToString(response)); } + /** + * Get the info of a connector running in this cluster (retrieved via the GET /connectors/{connector} endpoint). + + * @param connectorName name of the connector + * @return an instance of {@link ConnectorInfo} populated with state information of the connector and its tasks. + */ + public ConnectorInfo connectorInfo(String connectorName) { + ObjectMapper mapper = new ObjectMapper(); + String url = endpointForResource(String.format("connectors/%s", connectorName)); + Response response = requestGet(url); + try { + if (response.getStatus() < Response.Status.BAD_REQUEST.getStatusCode()) { + return mapper.readValue(responseToString(response), ConnectorInfo.class); + } + } catch (IOException e) { + log.error("Could not read connector info from response: {}", + responseToString(response), e); + throw new ConnectException("Could not not parse connector info", e); + } + throw new ConnectRestException(response.getStatus(), + "Could not read connector info. Error response: " + responseToString(response)); + } + + /** + * Get the task configs of a connector running in this cluster. + + * @param connectorName name of the connector + * @return a map from task ID (connector name + "-" + task number) to task config + */ + public Map> taskConfigs(String connectorName) { + ObjectMapper mapper = new ObjectMapper(); + String url = endpointForResource(String.format("connectors/%s/tasks-config", connectorName)); + Response response = requestGet(url); + try { + if (response.getStatus() < Response.Status.BAD_REQUEST.getStatusCode()) { + // We use String instead of ConnectorTaskId as the key here since the latter can't be automatically + // deserialized by Jackson when used as a JSON object key (i.e., when it's serialized as a JSON string) + return mapper.readValue(responseToString(response), new TypeReference>>() { }); + } + } catch (IOException e) { + log.error("Could not read task configs from response: {}", + responseToString(response), e); + throw new ConnectException("Could not not parse task configs", e); + } + throw new ConnectRestException(response.getStatus(), + "Could not read task configs. Error response: " + responseToString(response)); + } + /** * Reset the set of active topics of a connector running in this cluster. * diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedConnectClusterAssertions.java b/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedConnectClusterAssertions.java index c026cb72903da..d05185662203c 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedConnectClusterAssertions.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedConnectClusterAssertions.java @@ -27,6 +27,7 @@ import javax.ws.rs.core.Response; import java.util.Arrays; import java.util.Collection; +import java.util.Collections; import java.util.HashSet; import java.util.Map; import java.util.Optional; @@ -37,6 +38,7 @@ import java.util.stream.Collectors; import static org.apache.kafka.test.TestUtils.waitForCondition; +import static org.junit.Assert.assertEquals; /** * A set of common assertions that can be applied to a Connect cluster during integration testing @@ -301,7 +303,8 @@ public void assertConnectorAndAtLeastNumTasksAreRunning(String connectorName, in } /** - * Assert that a connector is running with at least the given number of tasks all in running state + * Assert that a connector is running, that it has a specific number of tasks, and that all of + * its tasks are in the RUNNING state. * * @param connectorName the connector name * @param numTasks the number of tasks @@ -326,6 +329,33 @@ public void assertConnectorAndExactlyNumTasksAreRunning(String connectorName, in } } + /** + * Assert that a connector is paused, that it has a specific number of tasks, and that all of + * its tasks are in the PAUSED state. + * + * @param connectorName the connector name + * @param numTasks the number of tasks + * @param detailMessage the assertion message + * @throws InterruptedException + */ + public void assertConnectorAndExactlyNumTasksArePaused(String connectorName, int numTasks, String detailMessage) + throws InterruptedException { + try { + waitForCondition( + () -> checkConnectorState( + connectorName, + AbstractStatus.State.PAUSED, + numTasks, + AbstractStatus.State.PAUSED, + Integer::equals + ).orElse(false), + CONNECTOR_SHUTDOWN_DURATION_MS, + "The connector or exactly " + numTasks + " tasks are not paused."); + } catch (AssertionError e) { + throw new AssertionError(detailMessage, e); + } + } + /** * Assert that a connector is running, that it has a specific number of tasks, and that all of * its tasks are in the FAILED state. @@ -415,11 +445,11 @@ public void assertConnectorIsFailedAndTasksHaveFailed(String connectorName, int * @param detailMessage the assertion message * @throws InterruptedException */ - public void assertConnectorAndTasksAreStopped(String connectorName, String detailMessage) + public void assertConnectorAndTasksAreNotRunning(String connectorName, String detailMessage) throws InterruptedException { try { waitForCondition( - () -> checkConnectorAndTasksAreStopped(connectorName), + () -> checkConnectorAndTasksAreNotRunning(connectorName), CONNECTOR_SETUP_DURATION_MS, "At least the connector or one of its tasks is still running"); } catch (AssertionError e) { @@ -433,7 +463,7 @@ public void assertConnectorAndTasksAreStopped(String connectorName, String detai * @param connectorName the connector * @return true if the connector and all the tasks are not in RUNNING state; false otherwise */ - protected boolean checkConnectorAndTasksAreStopped(String connectorName) { + protected boolean checkConnectorAndTasksAreNotRunning(String connectorName) { ConnectorStateInfo info; try { info = connect.connectorStatus(connectorName); @@ -450,6 +480,35 @@ protected boolean checkConnectorAndTasksAreStopped(String connectorName) { && info.tasks().stream().noneMatch(s -> s.state().equals(AbstractStatus.State.RUNNING.toString())); } + + /** + * Assert that a connector is in the stopped state and has no tasks. + * + * @param connectorName the connector name + * @param detailMessage the assertion message + * @throws InterruptedException + */ + public void assertConnectorIsStopped(String connectorName, String detailMessage) + throws InterruptedException { + try { + waitForCondition( + () -> checkConnectorState( + connectorName, + AbstractStatus.State.STOPPED, + 0, + null, + Integer::equals + ).orElse(false), + CONNECTOR_SHUTDOWN_DURATION_MS, + "At least the connector or one of its tasks is still running"); + // If the connector is truly stopped, we should also see an empty set of tasks and task configs + assertEquals(Collections.emptyList(), connect.connectorInfo(connectorName).tasks()); + assertEquals(Collections.emptyMap(), connect.taskConfigs(connectorName)); + } catch (AssertionError e) { + throw new AssertionError(detailMessage, e); + } + } + /** * Check whether the given connector state matches the current state of the connector and * whether it has at least the given number of tasks, with all the tasks matching the given diff --git a/gradle/spotbugs-exclude.xml b/gradle/spotbugs-exclude.xml index e519909b1d0cf..a659240da4147 100644 --- a/gradle/spotbugs-exclude.xml +++ b/gradle/spotbugs-exclude.xml @@ -303,7 +303,6 @@ For a detailed description of spotbugs bug categories, see https://spotbugs.read -