You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "yashmayya (via GitHub)" <gi...@apache.org> on 2023/07/24 07:24:24 UTC
[GitHub] [kafka] yashmayya opened a new pull request, #14079: KAFKA-14238: Move DLQ reporter setup from the DistributedHerder's tick thread to the sink task thread
yashmayya opened a new pull request, #14079:
URL: https://github.com/apache/kafka/pull/14079
From https://issues.apache.org/jira/browse/KAFKA-15238:
> When Kafka Connect is run in distributed mode - if a sink connector's task is restarted (via a worker's REST API), the following sequence of steps will occur (on the DistributedHerder's thread):
>
> 1. The existing sink task will be stopped ([ref](https://github.com/apache/kafka/blob/4981fa939d588645401619bfc3e321dc523d10e7/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java#L1367))
> 2. A new sink task will be started ([ref](https://github.com/apache/kafka/blob/4981fa939d588645401619bfc3e321dc523d10e7/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java#L1867C40-L1867C40))
> 3. As a part of the above step, a new WorkerSinkTask will be instantiated ([ref](https://github.com/apache/kafka/blob/4981fa939d588645401619bfc3e321dc523d10e7/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java#L656-L663))
> 4. The DLQ reporter (see [KIP-298](https://cwiki.apache.org/confluence/display/KAFKA/KIP-298%3A+Error+Handling+in+Connect)) for the sink task is also instantiated and configured as a part of this ([ref](https://github.com/apache/kafka/blob/4981fa939d588645401619bfc3e321dc523d10e7/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java#L1800))
> 5. The DLQ reporter setup involves two synchronous admin client calls to list topics and create the DLQ topic if it isn't already created ([ref](https://github.com/apache/kafka/blob/4981fa939d588645401619bfc3e321dc523d10e7/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/DeadLetterQueueReporter.java#L84-L87))
>
> All of these are occurring synchronously on the herder's tick thread - in this portion [here](https://github.com/apache/kafka/blob/4981fa939d588645401619bfc3e321dc523d10e7/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java#L457-L469) where external requests are run. If the admin client call in the DLQ reporter setup step blocks for some time (due to auth failures and retries or network issues or whatever other reason), this can cause the Connect worker to become non-functional (REST API requests will timeout) and even fall out of the Connect cluster and become a zombie (since the tick thread also drives group membership functions - see [here](https://github.com/apache/kafka/blob/4981fa939d588645401619bfc3e321dc523d10e7/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java#L403), [here](https://github.com/apache/kafka/blob/4981fa939d588645401619bfc3e321dc523d10e7/connect/runtime/src/main/ja
va/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java#L535)).
- This patch fixes the issue by moving the DLQ reporter setup from the `DistributedHerder`'s tick thread to the sink task's thread itself (in the [initializeAndStart](https://github.com/apache/kafka/blob/4981fa939d588645401619bfc3e321dc523d10e7/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerTask.java#L202) phase).
- This way, any blocking in the admin client calls during DLQ setup will only affect the sink task for which the DLQ is being setup (by delaying its startup) instead of affecting the entire Connect worker.
- In case the DLQ setup fails, the sink task's status will still be updated with the failure just like it would prior to this patch.
- A small integration test is also added to verify the DLQ topic functionality.
- Another thing to note is that no explicit timeout is being used for the blocking admin client calls which isn't ideal. We should probably add a reasonable timeout value (the default admin client timeout values are fairly large) for the overall `listTopics` + (potential) `createTopics` operation in either this patch or a separate one.
### 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
[GitHub] [kafka] C0urante merged pull request #14079: KAFKA-15238: Move DLQ reporter setup from the DistributedHerder's tick thread to the sink task thread
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante merged PR #14079:
URL: https://github.com/apache/kafka/pull/14079
--
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
[GitHub] [kafka] yashmayya commented on a diff in pull request #14079: KAFKA-15238: Move DLQ reporter setup from the DistributedHerder's tick thread to the sink task thread
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya commented on code in PR #14079:
URL: https://github.com/apache/kafka/pull/14079#discussion_r1272243105
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java:
##########
@@ -1797,7 +1797,6 @@ public WorkerTask doBuild(Task task,
TransformationChain<SinkRecord> transformationChain = new TransformationChain<>(connectorConfig.<SinkRecord>transformationStages(), retryWithToleranceOperator);
log.info("Initializing: {}", transformationChain);
SinkConnectorConfig sinkConfig = new SinkConnectorConfig(plugins, connectorConfig.originalsStrings());
- retryWithToleranceOperator.reporters(sinkTaskReporters(id, sinkConfig, errorHandlingMetrics, connectorClass));
Review Comment:
I guess we could do something similar for source tasks as well to maintain symmetry even though there's no real need to do so (no blocking calls there).
--
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
[GitHub] [kafka] C0urante commented on a diff in pull request #14079: KAFKA-15238: Move DLQ reporter setup from the DistributedHerder's tick thread to the sink task thread
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on code in PR #14079:
URL: https://github.com/apache/kafka/pull/14079#discussion_r1272633816
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/SinkConnectorsIntegrationTest.java:
##########
@@ -309,6 +319,48 @@ public void testCooperativeConsumerPartitionAssignment() throws Exception {
assertEquals(0, task.timesCommitted(tp3));
}
+ @Test
Review Comment:
Do we need a new integration test case for the DLQ topic? It looks like most of this is covered in the existing [ErrorHandlingIntegrationTest::testSkipRetryAndDLQWithHeaders case](https://github.com/apache/kafka/blob/84691b11f64e85a7f3f6fdbafd0f8fb2f8dd630c/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ErrorHandlingIntegrationTest.java#L107-L184).
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java:
##########
@@ -1797,7 +1797,6 @@ public WorkerTask doBuild(Task task,
TransformationChain<SinkRecord> transformationChain = new TransformationChain<>(connectorConfig.<SinkRecord>transformationStages(), retryWithToleranceOperator);
log.info("Initializing: {}", transformationChain);
SinkConnectorConfig sinkConfig = new SinkConnectorConfig(plugins, connectorConfig.originalsStrings());
- retryWithToleranceOperator.reporters(sinkTaskReporters(id, sinkConfig, errorHandlingMetrics, connectorClass));
Review Comment:
I'm fine either way, whatever you think would be easier to maintain/read in the future.
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskTest.java:
##########
@@ -1951,6 +1960,27 @@ public void testOriginalTopicWithTopicMutatingTransformations() {
PowerMock.verifyAll();
}
+ @Test
+ public void testErrorReportersConfigured() {
+ RetryWithToleranceOperator retryWithToleranceOperator = createMock(RetryWithToleranceOperator.class);
+ createTask(initialState, keyConverter, valueConverter, headerConverter, retryWithToleranceOperator,
+ () -> Collections.singletonList(createMock(ErrorReporter.class)));
+
+ expectInitializeTask();
+ Capture<List<ErrorReporter>> errorReporters = EasyMock.newCapture();
+ retryWithToleranceOperator.reporters(EasyMock.capture(errorReporters));
+ PowerMock.expectLastCall();
+
+ PowerMock.replayAll(retryWithToleranceOperator);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+
+ assertEquals(1, errorReporters.getValue().size());
Review Comment:
Nit: we can get a little more granular about our assertions here by using `assertEquals` to compare collections instead of just an expected vs. actual collection's size.
```suggestion
List<ErrorReporter> errorReporters = Collections.singletonList(createMock(ErrorReporter.class));
createTask(initialState, keyConverter, valueConverter, headerConverter, retryWithToleranceOperator,
() -> errorReporters);
expectInitializeTask();
Capture<List<ErrorReporter>> errorReportersCapture = EasyMock.newCapture();
retryWithToleranceOperator.reporters(EasyMock.capture(errorReportersCapture));
PowerMock.expectLastCall();
PowerMock.replayAll(retryWithToleranceOperator);
workerTask.initialize(TASK_CONFIG);
workerTask.initializeAndStart();
assertEquals(errorReporters, errorReportersCapture.getValue());
```
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskTest.java:
##########
@@ -1951,6 +1960,27 @@ public void testOriginalTopicWithTopicMutatingTransformations() {
PowerMock.verifyAll();
}
+ @Test
+ public void testErrorReportersConfigured() {
Review Comment:
Can we also add a test case to cover when the `Supplier<List<ErrorReporter>>` throws an exception, which verifies that that causes an exception to also be thrown from `WorkerSinkTask::initializeAndStart`?
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/ErrorHandlingTaskTest.java:
##########
@@ -494,7 +492,8 @@ private void expectTopicCreation(String topic) {
}
}
- private void createSinkTask(TargetState initialState, RetryWithToleranceOperator retryWithToleranceOperator) {
+ private void createSinkTask(TargetState initialState, RetryWithToleranceOperator retryWithToleranceOperator,
+ Supplier<List<ErrorReporter>> errorReportersSupplier) {
Review Comment:
Nit: do we have to accept a `Supplier` here or can we change the last parameter to `List<ErrorReporter> errorReporters` and pass `() -> errorReporters` to the `WorkerSinkTask` constructor below?
--
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
[GitHub] [kafka] yashmayya commented on pull request #14079: KAFKA-15238: Move DLQ reporter setup from the DistributedHerder's tick thread to the sink task thread
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya commented on PR #14079:
URL: https://github.com/apache/kafka/pull/14079#issuecomment-1649820640
Thanks for the swift reviews Chris! And yeah, even I was surprised by how brief the changes ended up being 😄
--
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
[GitHub] [kafka] yashmayya commented on pull request #14079: KAFKA-15238: Move DLQ reporter setup from the DistributedHerder's tick thread to the sink task thread
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya commented on PR #14079:
URL: https://github.com/apache/kafka/pull/14079#issuecomment-1647416167
Hi Chris, would you be able to take a look at this whenever you get a chance?
--
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
[GitHub] [kafka] yashmayya commented on a diff in pull request #14079: KAFKA-15238: Move DLQ reporter setup from the DistributedHerder's tick thread to the sink task thread
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya commented on code in PR #14079:
URL: https://github.com/apache/kafka/pull/14079#discussion_r1273081952
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java:
##########
@@ -1797,7 +1797,6 @@ public WorkerTask doBuild(Task task,
TransformationChain<SinkRecord> transformationChain = new TransformationChain<>(connectorConfig.<SinkRecord>transformationStages(), retryWithToleranceOperator);
log.info("Initializing: {}", transformationChain);
SinkConnectorConfig sinkConfig = new SinkConnectorConfig(plugins, connectorConfig.originalsStrings());
- retryWithToleranceOperator.reporters(sinkTaskReporters(id, sinkConfig, errorHandlingMetrics, connectorClass));
Review Comment:
It wasn't too much of a change so I've gone ahead and made it to maintain symmetry.
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskTest.java:
##########
@@ -1951,6 +1960,27 @@ public void testOriginalTopicWithTopicMutatingTransformations() {
PowerMock.verifyAll();
}
+ @Test
+ public void testErrorReportersConfigured() {
+ RetryWithToleranceOperator retryWithToleranceOperator = createMock(RetryWithToleranceOperator.class);
+ createTask(initialState, keyConverter, valueConverter, headerConverter, retryWithToleranceOperator,
+ () -> Collections.singletonList(createMock(ErrorReporter.class)));
+
+ expectInitializeTask();
+ Capture<List<ErrorReporter>> errorReporters = EasyMock.newCapture();
+ retryWithToleranceOperator.reporters(EasyMock.capture(errorReporters));
+ PowerMock.expectLastCall();
+
+ PowerMock.replayAll(retryWithToleranceOperator);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+
+ assertEquals(1, errorReporters.getValue().size());
Review Comment:
Makes sense, done.
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/SinkConnectorsIntegrationTest.java:
##########
@@ -309,6 +319,48 @@ public void testCooperativeConsumerPartitionAssignment() throws Exception {
assertEquals(0, task.timesCommitted(tp3));
}
+ @Test
Review Comment:
Whoops, you're right, I've removed this redundant IT.
--
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