You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "gharris1727 (via GitHub)" <gi...@apache.org> on 2023/11/14 18:48:20 UTC
[PR] KAFKA-15826: Close consumer when sink task is cancelled [kafka]
gharris1727 opened a new pull request, #14762:
URL: https://github.com/apache/kafka/pull/14762
Currently the only place that the consumer is closed is on the task thread, which may be blocked indefinitely by the task plugin. Similar to AbstractWorkerSourceTask, we should close the consumer here to allow for cleanup to take place.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15826: Close consumer when sink task is cancelled [kafka]
Posted by "gharris1727 (via GitHub)" <gi...@apache.org>.
gharris1727 commented on PR #14762:
URL: https://github.com/apache/kafka/pull/14762#issuecomment-1850986158
I found this same leak again when debugging some failures of the `kafka.utils.TestUtils.verifyNoUnexpectedThreads` assertion in core tests. Whenever we leak one of these clients in connect:runtime, all of the core tests using this assertion will fail. In particular, the BlockingConnectorTest causes this failure:
```
Found 1 unexpected threads during @BeforeAll: `kafka-coordinator-heartbeat-thread | connect-blocking-connector` ==> expected: <true> but was: <false>
Expected :true
Actual :false
<Click to see difference>
org.opentest4j.AssertionFailedError: Found 1 unexpected threads during @BeforeAll: `kafka-coordinator-heartbeat-thread | connect-blocking-connector` ==> expected: <true> but was: <false>
at app//org.junit.jupiter.api.AssertionFailureBuilder.build(AssertionFailureBuilder.java:151)
at app//org.junit.jupiter.api.AssertionFailureBuilder.buildAndThrow(AssertionFailureBuilder.java:132)
at app//org.junit.jupiter.api.AssertTrue.failNotTrue(AssertTrue.java:63)
at app//org.junit.jupiter.api.AssertTrue.assertTrue(AssertTrue.java:36)
at app//org.junit.jupiter.api.Assertions.assertTrue(Assertions.java:210)
at app//kafka.utils.TestUtils$.verifyNoUnexpectedThreads(TestUtils.scala:2412)
at app//kafka.utils.TestUtils.verifyNoUnexpectedThreads(TestUtils.scala)
```
This made me realize that the BlockingConnectorTest may be changed so that the operations block while the test is ongoing, and then unblock during the cleanup phase. This would stabilize the tests and resolve the leaks without changing the user-facing cancellation behavior or experiencing CMEs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15826: Close consumer when sink task is cancelled [kafka]
Posted by "gharris1727 (via GitHub)" <gi...@apache.org>.
gharris1727 closed pull request #14762: KAFKA-15826: Close consumer when sink task is cancelled
URL: https://github.com/apache/kafka/pull/14762
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15826: Close consumer when sink task is cancelled [kafka]
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya commented on PR #14762:
URL: https://github.com/apache/kafka/pull/14762#issuecomment-1814288871
> Since other workers in the cluster may not have this fix applied, I'll leave the error message unchanged and just update the assertions.
Hm, the original purpose of those tests were to verify that the admin client error handling [here](https://github.com/apache/kafka/blob/e7f4f5dfe7ef2c95f3e9a1dbbeded8502d3e1e05/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java#L1427-L1438) and [here](https://github.com/apache/kafka/blob/e7f4f5dfe7ef2c95f3e9a1dbbeded8502d3e1e05/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java#L1470-L1481) is correct and to safeguard against any regressions (from unexpected changes in the admin client or consumer group coordinator for instance) in that path. Considering that, these assertion changes defeat the purpose of those tests. Perhaps we could bump up the value for the `task.shutdown.graceful.timeout.ms` worker config (which would delay the task cancellation) in those particular tests in order to retain the original intent?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15826: Close consumer when sink task is cancelled [kafka]
Posted by "gharris1727 (via GitHub)" <gi...@apache.org>.
gharris1727 commented on PR #14762:
URL: https://github.com/apache/kafka/pull/14762#issuecomment-1813403690
The only problem I have with this approach is that the [official documentation](https://kafka.apache.org/36/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#multithreaded) states that the KafkaConsumer is not thread-safe, except for wakeup(). They recommend calling wakeup() on a different thread, and then close() on the primary thread. In the blocked connector case we're unlikely to be within a consumer call, but we technically don't have exclusive access to the consumer and could receive a ConcurrentModificationException on any of the consumer methods.
I think if we want cancellation to close the consumer, we'd also need to "take ownership" of the consumer and prevent the main thread from using the consumer. We already have isCancelled() guards on offset commit, but maybe that isn't sufficient to prevent ConcurrentModificationExceptions. I'll have to look into this more.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15826: Close consumer when sink task is cancelled [kafka]
Posted by "gharris1727 (via GitHub)" <gi...@apache.org>.
gharris1727 commented on PR #14762:
URL: https://github.com/apache/kafka/pull/14762#issuecomment-1812953889
@yashmayya I noticed those failures and wasn't sure they're affected, thanks for explaining exactly what is going on!
Since other workers in the cluster may not have this fix applied, I'll leave the error message unchanged and just update the assertions. There could also be unpredictable zombie conditions where cancel() is never called, and this fix is ineffective.
The change to the offsets API behavior was unintended, but I think it is beneficial, and indicates that this change is moving in the right direction. Now there are fewer situations in which a user needs to restart the workers to use the REST API 🙃.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15826: Close consumer when sink task is cancelled [kafka]
Posted by "gharris1727 (via GitHub)" <gi...@apache.org>.
gharris1727 commented on PR #14762:
URL: https://github.com/apache/kafka/pull/14762#issuecomment-1815495283
Hey @yashmayya Thanks for letting me know the original intent of the test. I agree, changing the test to expect the call to succeed was not in the spirit of the original test, so i've reverted back to the old assertions.
I tried to increase the graceful shutdown timeout, but I wasn't able to get the calls to fail. I think that the graceful shutdown blocks the herder thread, and prevents the herder from servicing the alter/reset requests, so they always wait to execute until after the cancellation has been applied.
Rather than trying to get the requests to target a different worker and dealing with that nondeterminism, I elected to just simulate the zombie tasks with an out-of-band consumer that I create in the test. That should also test the case when a user has an external consumer in the group.
PTAL, thanks so much for the review!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15826: Close consumer when sink task is cancelled [kafka]
Posted by "gharris1727 (via GitHub)" <gi...@apache.org>.
gharris1727 commented on PR #14762:
URL: https://github.com/apache/kafka/pull/14762#issuecomment-1896474948
I think due to the locking concerns I raised earlier, and that we can resolve this resource leak in our tests, this PR is not viable to merge. We can revisit this in the future if the locking model of the consumer can be changed, or if this becomes a demonstrable problem in real situations and not just tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15826: Close consumer when sink task is cancelled [kafka]
Posted by "gharris1727 (via GitHub)" <gi...@apache.org>.
gharris1727 commented on PR #14762:
URL: https://github.com/apache/kafka/pull/14762#issuecomment-1816905171
> What was the conclusion here? Is triggering wakeup on a separate thread (through WorkerSinkTask::stop) before also eventually closing the consumer on a separate thread (through WorkerSinkTask::cancel) safe? Looks like it probably might be, but I was only able to take a cursory look.
I don't think it is. For example, it looks like doCommitSync catches the WakeupException and then makes another call to the consumer, and that call could be ongoing when cancel()/close() is called. I think also that after wakeup() exits, there's no guarantee that the other thread has actually thrown the WakeupException and released the consumer lock. Since the plugin has access to the WorkerSinkTaskContext, it can also perform an infinite loop and catch all of the WakeupExceptions.
I think that preventing ConcurrentModificationExceptions completely will require heavy synchronization, which gives more opportunities for the task thread to block the herder thread. If we assume that framework has no infinite loops (which may not be the case given the behavior of doCommitSync), we could make cancellation only happen if the task thread is in plugin code, and the consumer is unlikely to be in use. This could be done with an AtomicReference, which is the same synchronization primitive that the Consumer is using to fire the ConcurrentModificationExceptions.
I'll look into it more, I don't think we should merge this change until we have a handle on the CMEs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15826: Close consumer when sink task is cancelled [kafka]
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya commented on code in PR #14762:
URL: https://github.com/apache/kafka/pull/14762#discussion_r1397042412
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java:
##########
@@ -163,6 +163,12 @@ public void initialize(TaskConfig taskConfig) {
}
}
+ @Override
+ public void cancel() {
+ super.cancel();
+ Utils.closeQuietly(consumer, "consumer");
Review Comment:
```suggestion
Utils.closeQuietly(consumer, "consumer for sink task: " + id);
```
nit
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java:
##########
@@ -767,11 +775,7 @@ public void testResetSinkConnectorOffsetsZombieSinkTasks() throws Exception {
connect.kafka().produce(TOPIC, 0, "key", "value");
}
- // Configure a sink connector whose sink task blocks in its stop method
- Map<String, String> connectorConfigs = new HashMap<>();
- connectorConfigs.put(CONNECTOR_CLASS_CONFIG, BlockingConnectorTest.BlockingSinkConnector.class.getName());
- connectorConfigs.put(TOPICS_CONFIG, TOPIC);
- connectorConfigs.put("block", "Task::stop");
+ Map<String, String> connectorConfigs = baseSinkConnectorConfigs();
Review Comment:
Same as above, `baseSinkConnectorConfigs()` can be used directly in the call to `configureConnector`
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java:
##########
@@ -408,27 +412,30 @@ public void testAlterSinkConnectorOffsetsZombieSinkTasks() throws Exception {
connect.kafka().produce(TOPIC, 0, "key", "value");
}
- // Configure a sink connector whose sink task blocks in its stop method
- Map<String, String> connectorConfigs = new HashMap<>();
- connectorConfigs.put(CONNECTOR_CLASS_CONFIG, BlockingConnectorTest.BlockingSinkConnector.class.getName());
- connectorConfigs.put(TOPICS_CONFIG, TOPIC);
- connectorConfigs.put("block", "Task::stop");
+ Map<String, String> connectorConfigs = baseSinkConnectorConfigs();
Review Comment:
This can be used directly now (i.e. without defining a separate local variable)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org