KAFKA-14783 (KIP-875): New STOPPED state for connectors by C0urante · Pull Request #13424 · apache/kafka

C0urante · 2023-03-20T21:05:49Z

Adds the new STOPPED target state for connectors, which causes all tasks for the connector to be shut down and for its status in the REST API to be updated to STOPPED.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

C0urante · 2023-03-20T21:07:25Z

cc @yashmayya; feel free to review and/or build off of this for KAFKA-14786, KAFKA-14368, and KAFKA-14784.

C0urante · 2023-03-20T21:08:04Z

@mimaison do you think you'll have time to take a look? Hoping to get this in before the 3.5.0 release but we can punt if there's not enough bandwidth for review.

mimaison

Yeah let's try to get this into 3.5. I took a quick look and left a few minor comments.

mimaison · 2023-03-21T09:52:23Z

    void restartConnectorAndTasks(RestartRequest request, Callback<ConnectorStateInfo> cb);

+    /**
+     * Stop the conector. This call will asynchronously suspend processing by the connector and all


connector

mimaison · 2023-03-21T13:02:40Z

-            switch (state) {
-                case STOPPED:
-                    return;
+            if ((state == State.STOPPED || state == State.PAUSED) && state == newState) {


As newState is either PAUSED or STOPPED, would state == newState be enough here?

Yep, good call 👍

mimaison · 2023-03-21T15:04:35Z

+                        return null;
+                    }
+
+                    // TODO: We may want to add a new ConfigBackingStore method for stopping a connector so that


Should we create a ticket for this TODO?

I'm worried that if we create a ticket, somebody might jump on that and make the change now, even though it's not really necessary yet. If it's too bothersome as a TODO I could just remove it altogether?

Usually I prefer TODO tags to be associated with a task. What's the drawback if someone was to make this change now?

Currently, the KafkaConfigBackingStore uses a transactional producer only when exactly once support is enabled on the worker. Maybe Chris' concern was that currently we'll only be able to do the tasks config + target state write atomically when exactly once support is enabled on the worker? 🤔

Although that is already the case today for other operations that require multiple writes to the config topic like putTaskConfigs and removeConnectorConfig, so I'm not sure whether or not that was his concern.

I guess this wasn't really a TODO so much as a MAYBE-DO. I've removed it since on second thought it's a DON'T-DO-FOR-NOW as there is little if any benefit from that change, except possibly blocking the herder thread for less time if we're having trouble writing to the config topic.

There's also the transactional producer logic that Yash mentioned, but since it's not guaranteed that we'll have access to one (if exactly-once source support is disabled), that's not worth considering right now.

mimaison · 2023-03-21T15:29:58Z

        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 start");


during connector start -> during connector stop

yashmayya

Thanks Chris! I haven't had a chance to go through all the tests yet, but I hope to do so in a subsequent review.

yashmayya · 2023-03-22T12:26:41Z

+     */
+    public Map<String, Map<String, String>> taskConfigs(String connectorName) {
+        ObjectMapper mapper = new ObjectMapper();
+        String url = endpointForResource(String.format("connectors/%s/tasks-config", connectorName));


I'm a little confused as to why we have a GET /connectors/{connector}/tasks as well as a GET /connectors/{connector}/tasks-config API? Looks like the only difference between them is that the former returns "raw" task configurations (i.e. before externalized secrets are replaced using config transformers) - in which case the names of the two APIs don't really help to distinguish between them much (same with the Herder methods - Herder::taskConfigs and Herder::tasksConfig).

Good question... I know the tasks-config endpoint was introduced in KIP-661, but now that you mention it, it does seem a little redundant given the /tasks endpoint's content.

Maybe @mimaison could shed some light here?

I thought there was a difference but checking now they seem pretty similar. I also wonder if GET /connectors/{connector}/tasks-config should avoid returning the raw configs to avoid leaking values returned by providers.

I also wonder if GET /connectors/{connector}/tasks-config should avoid returning the raw configs to avoid leaking values returned by providers.

Yeah that's a good point as well - it's possible that with certain config providers, Connect operators won't be able to access the raw secrets directly but will have access to the worker's REST API (and the worker obviously will have access to the secrets from the provider) right?

Also, if we choose to make that change (I assume it'll require a KIP because it's changing the response of a public REST API), the two APIs would essentially become identical. Would it be possible to deprecate / remove one of the APIs?

C0urante · 2023-03-23T14:43:25Z

Thanks all for the reviews! I believe I've addressed every comment with either a code change or a response. This should be ready for another pass.

yashmayya

Thanks Chris, I just had a couple more minor comments, but none blocking. LGTM!

yashmayya · 2023-03-24T06:50:47Z

This calls TaskStatus.Listener::onDeletion after the tasks are stopped by the worker which updates their status to DESTROYED whereas this doesn't seem to happen in the DistributedHerder implementation for stopConnector so the task statuses will be UNASSIGNED post shutdown. I guess this shouldn't really matter anyway because we're publishing an empty set of task configs?

I guess this shouldn't really matter anyway because we're publishing an empty set of task configs?

Exactly--and thanks to the logic introduced in the fix for KAFKA-9472, once the rebalance caused by the empty set of task configs has completed, the DistributedHerder class will automatically wipe the deleted tasks from the status store.

gharris1727

LGTM!

Co-authored-by: Yash Mayya <yash.mayya@gmail.com>

…r::suspend, fix MonitorableSourceConnector exception message

…nector

…uspend catch block

… skipped

…en in DistributedHerder::stopConnector

…edHerderTest Co-authored-by: Yash Mayya <yash.mayya@gmail.com>

Co-authored-by: Yash Mayya <yash.mayya@gmail.com>

…HerderTest

C0urante · 2023-04-10T15:16:14Z

@mimaison thanks for the reviews. Do you think you'll have time to take another look at this before the April 12th feature freeze?

mimaison

LGTM

C0urante · 2023-04-11T13:37:33Z

Thanks Mickael!

… w/ conflicts resolved Reviewers: Mickael Maison <mickael.maison@gmail.com>, Yash Mayya <yash.mayya@gmail.com>, Greg Harris <gharris1727@gmail.com>

C0urante changed the title ~~KAFKA-14783: New STOPPED state for connectors~~ KAFKA-14783 (KIP-875): New STOPPED state for connectors Mar 20, 2023

mimaison reviewed Mar 21, 2023

View reviewed changes

yashmayya reviewed Mar 22, 2023

View reviewed changes

C0urante added connect kip Requires or implements a KIP labels Mar 22, 2023

gharris1727 reviewed Mar 22, 2023

View reviewed changes

Comment thread connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractStatus.java Outdated

Comment thread connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConnector.java Outdated

yashmayya reviewed Mar 23, 2023

View reviewed changes

yashmayya approved these changes Mar 24, 2023

View reviewed changes

gharris1727 approved these changes Mar 27, 2023

View reviewed changes

C0urante force-pushed the kafka-14783 branch from 9cf7b4b to d05029a Compare March 27, 2023 20:58

yashmayya mentioned this pull request Mar 28, 2023

KAFKA-14368: Connect offset write REST API #13465

Merged

3 tasks

C0urante and others added 14 commits April 2, 2023 13:22

KAFKA-14783: New STOPPED state for connectors

cf2befa

Remove unnecessary "fallthrough" warning suppression

b34f3a3

Co-authored-by: Yash Mayya <yash.mayya@gmail.com>

Fix typo in ConnectWorkerIntegrationTest comment

7ee1465

Co-authored-by: Yash Mayya <yash.mayya@gmail.com>

Move new STOPPED value for AbstractStatus.State enum to end

93ffb44

Rename WorkerConnector::stop to WorkerConnector::suspend

5d5f9aa

Fix typo in 'connector', simplify boolean condition in WorkerConnecto…

cc03d90

…r::suspend, fix MonitorableSourceConnector exception message

Remove unnecessary SpotBugs exclusion

f83c4e3

Add detail to Javadocs for Herder::stopConnector and Herder::pauseCon…

6412a32

…nector

Remove try/catch around call to status listener in WorkerConnector::s…

4c6a9e1

…uspend catch block

Improve log messages in StandaloneHerder when task config updates are…

4739c9d

… skipped

Move read-to-end of config topic to after target state has been writt…

56258fd

…en in DistributedHerder::stopConnector

Remove redundant invocation of PowerMock::expectLastCall in Distribut…

15e5766

…edHerderTest Co-authored-by: Yash Mayya <yash.mayya@gmail.com>

Fix copy+paste mistake in DistributedHerderTest::testConnectorStopped

d39dff2

Co-authored-by: Yash Mayya <yash.mayya@gmail.com>

Fix task configs for stopped connector config snapshot in Distributed…

40efa62

…HerderTest

C0urante force-pushed the kafka-14783 branch from d05029a to 40efa62 Compare April 2, 2023 17:23

Remove TODO from DistributedHerder

a210847

mimaison approved these changes Apr 11, 2023

View reviewed changes

C0urante merged commit e49a5a2 into apache:trunk Apr 11, 2023

C0urante deleted the kafka-14783 branch April 11, 2023 13:37

gharris1727 mentioned this pull request Apr 12, 2023

MINOR: Fix regression in MM2 task forwarding introduced by KAFKA-14783 #13548

Merged

3 tasks

yashmayya mentioned this pull request May 2, 2023

KAFKA-14876: Document the new 'PUT /connectors/{name}/stop' REST API for Connect #13657

Merged

3 tasks

yashmayya mentioned this pull request Jun 6, 2023

KAFKA-14784: Connect offset reset REST API #13818

Merged

3 tasks

yashmayya mentioned this pull request Jul 6, 2023

MINOR: Update connector status metric description to include 'stopped' as a potential value #13967

Merged

3 tasks

anleib mentioned this pull request Oct 4, 2024

KafkaConnectClient not supporting stop operation SourceLabOrg/kafka-connect-client#86

Closed

Conversation

C0urante commented Mar 20, 2023

Committer Checklist (excluded from commit message)

Uh oh!

C0urante commented Mar 20, 2023

Uh oh!

C0urante commented Mar 20, 2023

Uh oh!

mimaison left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

C0urante Apr 10, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

yashmayya left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

yashmayya Mar 22, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

C0urante commented Mar 23, 2023

Uh oh!

yashmayya left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

gharris1727 left a comment

Choose a reason for hiding this comment

Uh oh!

C0urante commented Apr 10, 2023

Uh oh!

mimaison left a comment

Choose a reason for hiding this comment

Uh oh!

C0urante commented Apr 11, 2023

Uh oh!

C0urante Apr 10, 2023 •

edited

Loading

yashmayya Mar 22, 2023 •

edited

Loading