From f3aa1df3c32cb6a10e0dfeaeea5061625660926d Mon Sep 17 00:00:00 2001 From: Chris Egerton Date: Tue, 11 Feb 2020 13:30:17 -0800 Subject: [PATCH 1/4] KAFKA-9472: Add failing integration test --- .../ConnectWorkerIntegrationTest.java | 39 ++++++++++++++++++- 1 file changed, 37 insertions(+), 2 deletions(-) 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 b8e04970f610d..b3f3020205528 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 @@ -102,7 +102,7 @@ public void testAddAndRemoveWorker() throws Exception { // create test topic connect.kafka().createTopic("test-topic", NUM_TOPIC_PARTITIONS); - // setup up props for the sink connector + // set up props for the source connector Map props = new HashMap<>(); props.put(CONNECTOR_CLASS_CONFIG, MonitorableSourceConnector.class.getSimpleName()); props.put(TASKS_MAX_CONFIG, String.valueOf(numTasks)); @@ -193,7 +193,7 @@ public void testBrokerCoordinator() throws Exception { // create test topic connect.kafka().createTopic("test-topic", NUM_TOPIC_PARTITIONS); - // setup up props for the sink connector + // set up props for the source connector Map props = new HashMap<>(); props.put(CONNECTOR_CLASS_CONFIG, MonitorableSourceConnector.class.getSimpleName()); props.put(TASKS_MAX_CONFIG, String.valueOf(numTasks)); @@ -236,4 +236,39 @@ public void testBrokerCoordinator() throws Exception { "Connector tasks did not start in time."); } + /** + * Verify that the number of tasks listed in the REST API is updated correctly after changes to + * the "tasks.max" connector configuration. + */ + @Test + public void testTaskStatuses() throws Exception { + connect = connectBuilder.build(); + // start the clusters + connect.start(); + + connect.assertions().assertAtLeastNumWorkersAreUp(NUM_WORKERS, + "Initial group of workers did not start in time."); + + // base connector props + Map connectorProps = new HashMap<>(); + connectorProps.put(CONNECTOR_CLASS_CONFIG, MonitorableSourceConnector.class.getSimpleName()); + + // start the connector with only one task + final int initialNumTasks = 1; + connectorProps.put(TASKS_MAX_CONFIG, String.valueOf(initialNumTasks)); + connect.configureConnector(CONNECTOR_NAME, connectorProps); + connect.assertions().assertConnectorAndExactlyNumTasksAreRunning(CONNECTOR_NAME, initialNumTasks, "Connector tasks did not start in time"); + + // then reconfigure it to use more tasks + final int increasedNumTasks = 5; + connectorProps.put(TASKS_MAX_CONFIG, String.valueOf(increasedNumTasks)); + connect.configureConnector(CONNECTOR_NAME, connectorProps); + connect.assertions().assertConnectorAndExactlyNumTasksAreRunning(CONNECTOR_NAME, increasedNumTasks, "Connector task statuses did not update in time."); + + // then reconfigure it to use fewer tasks + final int decreasedNumTasks = 3; + connectorProps.put(TASKS_MAX_CONFIG, String.valueOf(decreasedNumTasks)); + connect.configureConnector(CONNECTOR_NAME, connectorProps); + connect.assertions().assertConnectorAndExactlyNumTasksAreRunning(CONNECTOR_NAME, decreasedNumTasks, "Connector task statuses did not update in time."); + } } From 930561a05ef26ec3fcf090df7f64f0ed59b13746 Mon Sep 17 00:00:00 2001 From: Chris Egerton Date: Thu, 13 Feb 2020 17:00:27 -0800 Subject: [PATCH 2/4] KAFKA-9472: Delete statuses of no-longer-needed tasks in distributed mode --- .../kafka/connect/runtime/AbstractHerder.java | 7 +++++- .../kafka/connect/runtime/TaskStatus.java | 11 +++++++++ .../kafka/connect/runtime/WorkerTask.java | 6 +++++ .../distributed/DistributedHerder.java | 23 +++++++++++++++---- .../distributed/DistributedHerderTest.java | 3 ++- 5 files changed, 44 insertions(+), 6 deletions(-) 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 f0315b6eb7f39..cb6033365ec47 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 @@ -189,10 +189,15 @@ public void onPause(ConnectorTaskId id) { @Override public void onDeletion(String connector) { for (TaskStatus status : statusBackingStore.getAll(connector)) - statusBackingStore.put(new TaskStatus(status.id(), TaskStatus.State.DESTROYED, workerId, generation())); + onDeletion(status.id()); statusBackingStore.put(new ConnectorStatus(connector, ConnectorStatus.State.DESTROYED, workerId, generation())); } + @Override + public void onDeletion(ConnectorTaskId id) { + statusBackingStore.put(new TaskStatus(id, TaskStatus.State.DESTROYED, workerId, generation())); + } + @Override public void pauseConnector(String connector) { if (!configBackingStore.contains(connector)) diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/TaskStatus.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/TaskStatus.java index 5ee90415b9660..561c742e4dc2b 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/TaskStatus.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/TaskStatus.java @@ -61,5 +61,16 @@ public interface Listener { */ void onShutdown(ConnectorTaskId id); + /** + * Invoked after the task is no longer needed. This differs from + * {@link #onShutdown(ConnectorTaskId)} in that a shut down task may be expected to restart + * soon (as in the case of a rebalance), whereas a destroyed task will not be restarted + * until and unless a reconfiguration of its connector occurs. + * + * This may occur after the number of tasks for a connector is reduced. + * + * @param id The id of the task + */ + void onDeletion(ConnectorTaskId id); } } 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 c40dc90012e33..b0a6a0cb1da69 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 @@ -436,6 +436,12 @@ public void onShutdown(ConnectorTaskId id) { delegateListener.onShutdown(id); } + @Override + public void onDeletion(ConnectorTaskId id) { + taskStateTimer.changeState(State.DESTROYED, time.milliseconds()); + delegateListener.onDeletion(id); + } + public void recordState(TargetState state) { switch (state) { case STARTED: 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 bfed9ecfd3424..ce099fd9a1361 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 @@ -46,6 +46,7 @@ import org.apache.kafka.connect.runtime.SinkConnectorConfig; import org.apache.kafka.connect.runtime.SourceConnectorConfig; import org.apache.kafka.connect.runtime.TargetState; +import org.apache.kafka.connect.runtime.TaskStatus; import org.apache.kafka.connect.runtime.Worker; import org.apache.kafka.connect.runtime.rest.InternalRequestSignature; import org.apache.kafka.connect.runtime.rest.RestClient; @@ -1522,6 +1523,18 @@ private void updateDeletedConnectorStatus() { } } + private void updateDeletedTaskStatus() { + ClusterConfigState snapshot = configBackingStore.snapshot(); + for (String connector : statusBackingStore.connectors()) { + Set remainingTasks = new HashSet<>(snapshot.tasks(connector)); + + statusBackingStore.getAll(connector).stream() + .map(TaskStatus::id) + .filter(task -> !remainingTasks.contains(task)) + .forEach(this::onDeletion); + } + } + protected HerderMetrics herderMetrics() { return herderMetrics; } @@ -1584,11 +1597,13 @@ public void onAssigned(ExtendedAssignment assignment, int generation) { herderMetrics.rebalanceStarted(time.milliseconds()); } - // Delete the statuses of all connectors removed prior to the start of this rebalance. This has to - // be done after the rebalance completes to avoid race conditions as the previous generation attempts - // to change the state to UNASSIGNED after tasks have been stopped. - if (isLeader()) + // Delete the statuses of all connectors and tasks removed prior to the start of this rebalance. This + // has to be done after the rebalance completes to avoid race conditions as the previous generation + // attempts to change the state to UNASSIGNED after tasks have been stopped. + if (isLeader()) { updateDeletedConnectorStatus(); + updateDeletedTaskStatus(); + } // We *must* interrupt any poll() call since this could occur when the poll starts, and we might then // sleep in the poll() for a long time. Forcing a wakeup ensures we'll get to process this event in the 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 74693f890e2c1..4b73f9def4b2e 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 @@ -203,7 +203,7 @@ public void setUp() throws Exception { connectProtocolVersion = CONNECT_PROTOCOL_V0; herder = PowerMock.createPartialMock(DistributedHerder.class, - new String[]{"backoff", "connectorTypeForClass", "updateDeletedConnectorStatus"}, + new String[]{"backoff", "connectorTypeForClass", "updateDeletedConnectorStatus", "updateDeletedTaskStatus"}, new DistributedConfig(HERDER_CONFIG), worker, WORKER_ID, KAFKA_CLUSTER_ID, statusBackingStore, configBackingStore, member, MEMBER_URL, metrics, time, noneConnectorClientConfigOverridePolicy); @@ -217,6 +217,7 @@ public void setUp() throws Exception { delegatingLoader = PowerMock.createMock(DelegatingClassLoader.class); PowerMock.mockStatic(Plugins.class); PowerMock.expectPrivate(herder, "updateDeletedConnectorStatus").andVoid().anyTimes(); + PowerMock.expectPrivate(herder, "updateDeletedTaskStatus").andVoid().anyTimes(); } @After From 02d97ec9e6126a2a8230b2e287143e762ea02c16 Mon Sep 17 00:00:00 2001 From: Chris Egerton Date: Fri, 14 Feb 2020 12:52:52 -0800 Subject: [PATCH 3/4] KAFKA-9472: Delete statuses of no-longer-needed tasks in standalone mode --- .../kafka/connect/runtime/standalone/StandaloneHerder.java | 1 + .../kafka/connect/runtime/standalone/StandaloneHerderTest.java | 3 +++ 2 files changed, 4 insertions(+) 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 a3e75b5829b5f..6c5398f8e4573 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 @@ -319,6 +319,7 @@ private void removeConnectorTasks(String connName) { if (!tasks.isEmpty()) { worker.stopAndAwaitTasks(tasks); configBackingStore.removeTaskConfigs(connName); + tasks.forEach(this::onDeletion); } } 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 3018677b82a9c..befca3227c2b6 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 @@ -266,6 +266,7 @@ public void testDestroyConnector() throws Exception { EasyMock.expect(statusBackingStore.getAll(CONNECTOR_NAME)).andReturn(Collections.emptyList()); statusBackingStore.put(new ConnectorStatus(CONNECTOR_NAME, AbstractStatus.State.DESTROYED, WORKER_ID, 0)); + statusBackingStore.put(new TaskStatus(new ConnectorTaskId(CONNECTOR_NAME, 0), TaskStatus.State.DESTROYED, WORKER_ID, 0)); expectDestroy(); @@ -434,6 +435,8 @@ public void testCreateAndStop() throws Exception { // 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)); + statusBackingStore.stop(); EasyMock.expectLastCall(); worker.stop(); From 4bd4bc32249c50608039b95ffe8ab338a9c2126b Mon Sep 17 00:00:00 2001 From: Chris Egerton Date: Thu, 21 May 2020 13:30:54 -0700 Subject: [PATCH 4/4] KAFKA-9472: Make javadoc for added method conform to style of other methods --- .../org/apache/kafka/connect/runtime/TaskStatus.java | 10 +++------- 1 file changed, 3 insertions(+), 7 deletions(-) diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/TaskStatus.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/TaskStatus.java index 561c742e4dc2b..62bb1c717112b 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/TaskStatus.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/TaskStatus.java @@ -62,13 +62,9 @@ public interface Listener { void onShutdown(ConnectorTaskId id); /** - * Invoked after the task is no longer needed. This differs from - * {@link #onShutdown(ConnectorTaskId)} in that a shut down task may be expected to restart - * soon (as in the case of a rebalance), whereas a destroyed task will not be restarted - * until and unless a reconfiguration of its connector occurs. - * - * This may occur after the number of tasks for a connector is reduced. - * + * Invoked after the task has been deleted. Can be called if the + * connector tasks have been reduced, or if the connector itself has + * been deleted. * @param id The id of the task */ void onDeletion(ConnectorTaskId id);