You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "vamossagar12 (via GitHub)" <gi...@apache.org> on 2023/04/05 16:41:35 UTC
[GitHub] [kafka] vamossagar12 opened a new pull request, #13504: Kafka 14750
vamossagar12 opened a new pull request, #13504:
URL: https://github.com/apache/kafka/pull/13504
*More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.*
*Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.*
### 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] vamossagar12 commented on a diff in pull request #13504: KAFKA-14750: Check if topic exists in WorkerSinkTask when committing offsets.
Posted by "vamossagar12 (via GitHub)" <gi...@apache.org>.
vamossagar12 commented on code in PR #13504:
URL: https://github.com/apache/kafka/pull/13504#discussion_r1207922909
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java:
##########
@@ -700,9 +714,28 @@ private class HandleRebalance implements ConsumerRebalanceListener {
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
log.debug("{} Partitions assigned {}", WorkerSinkTask.this, partitions);
-
+ Set<String> deletedTopics = new HashSet<>();
for (TopicPartition tp : partitions) {
- long pos = consumer.position(tp);
+ if (deletedTopics.contains(tp.topic())) {
Review Comment:
> It's unlikely that Kafka Connect is the only consumer application that is impacted by this scenario
I understand that Connect is not the only consumer application that would be impacted by this behaviour. But ATM, it does impact the users in certain scenarios as described in the ticket. Nonetheless, I would involve somebody from the Client team.
--
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] vamossagar12 commented on pull request #13504: KAFKA-14750: Check if topic exists in WorkerSinkTask when committing offsets.
Posted by "vamossagar12 (via GitHub)" <gi...@apache.org>.
vamossagar12 commented on PR #13504:
URL: https://github.com/apache/kafka/pull/13504#issuecomment-1708163762
hey Chris, this one got marked as stale. I had posted an update [here](https://github.com/apache/kafka/pull/13504#issuecomment-1569680586) based on Deng's suggestion. Let me know if there's anything in that which can be considered as a workaround for this issue.
--
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] vamossagar12 commented on a diff in pull request #13504: KAFKA-14750: Check if topic exists in WorkerSinkTask when committing offsets.
Posted by "vamossagar12 (via GitHub)" <gi...@apache.org>.
vamossagar12 commented on code in PR #13504:
URL: https://github.com/apache/kafka/pull/13504#discussion_r1192983825
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java:
##########
@@ -700,9 +714,28 @@ private class HandleRebalance implements ConsumerRebalanceListener {
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
log.debug("{} Partitions assigned {}", WorkerSinkTask.this, partitions);
-
+ Set<String> deletedTopics = new HashSet<>();
for (TopicPartition tp : partitions) {
- long pos = consumer.position(tp);
+ if (deletedTopics.contains(tp.topic())) {
+ log.debug("Not Committing offsets for topic-partition {} since the topic {} has been deleted", tp, tp.topic());
Review Comment:
Changed this line.
--
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] vamossagar12 commented on a diff in pull request #13504: KAFKA-14750: Check if topic exists in WorkerSinkTask when committing offsets.
Posted by "vamossagar12 (via GitHub)" <gi...@apache.org>.
vamossagar12 commented on code in PR #13504:
URL: https://github.com/apache/kafka/pull/13504#discussion_r1169769282
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java:
##########
@@ -145,6 +154,10 @@ public WorkerSinkTask(ConnectorTaskId id,
this.isTopicTrackingEnabled = workerConfig.getBoolean(TOPIC_TRACKING_ENABLE_CONFIG);
this.taskStopped = false;
this.workerErrantRecordReporter = workerErrantRecordReporter;
+ Map<String, Object> adminProps = new HashMap<>();
+ adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, workerConfig.bootstrapServers());
Review Comment:
Ok. Thanks for pointing this out. Will make the changes as suggested.
--
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] vamossagar12 commented on a diff in pull request #13504: KAFKA-14750: Check if topic exists in WorkerSinkTask when committing offsets.
Posted by "vamossagar12 (via GitHub)" <gi...@apache.org>.
vamossagar12 commented on code in PR #13504:
URL: https://github.com/apache/kafka/pull/13504#discussion_r1207962014
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java:
##########
@@ -695,9 +705,28 @@ private class HandleRebalance implements ConsumerRebalanceListener {
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
log.debug("{} Partitions assigned {}", WorkerSinkTask.this, partitions);
-
+ Set<String> deletedTopics = new HashSet<>();
for (TopicPartition tp : partitions) {
- long pos = consumer.position(tp);
+ if (deletedTopics.contains(tp.topic())) {
+ log.debug("Not assigning offsets for topic-partition {} since the topic {} has been deleted", tp, tp.topic());
+ continue;
+ }
+ long pos;
+ try {
+ pos = consumer.position(tp);
+ } catch (TimeoutException e) {
+ log.error("TimeoutException occurred when fetching position for topic partition {}. " +
+ "Checking if the topic {} exists", tp, tp.topic());
+ Map<String, TopicDescription> topic = topicAdmin.describeTopics(tp.topic());
Review Comment:
> This adds new ACL requirements for sink connectors' admin clients,
That's a good point and something I missed.
> Probably safest to assume that the topic exists if we fail to describe it.
Yeah.. In the case when let's say `TopicAuthorizationException` is thrown, it again becomes difficult to say if the topic exists or not. The `TopicAdmin#describeTopics` throws a `ConnectException` in this case which could be a false positive. And since this PR throws any exceptions from `TopicAdmin#describeTopics` the connector might now fail with ConnectException, albeit a different one. So, as you said might be safe to assume that topic exists then.
--
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-14750: Check if topic exists in WorkerSinkTask when committing offsets. [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13504:
URL: https://github.com/apache/kafka/pull/13504#issuecomment-2629226244
This PR has been closed since it has not had any activity in 120 days. If you feel like this
was a mistake, or you would like to continue working on it, please feel free to re-open the
PR and ask for a 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
[GitHub] [kafka] vamossagar12 commented on pull request #13504: KAFKA-14750: Check if topic exists in WorkerSinkTask when committing offsets.
Posted by "vamossagar12 (via GitHub)" <gi...@apache.org>.
vamossagar12 commented on PR #13504:
URL: https://github.com/apache/kafka/pull/13504#issuecomment-1565417056
@dengziming , I am looping you in here since I see you wrote the Admin APIs that this PR is trying to use and need your inputs for this PR. This is the description of the problem at hand :
```
Currently, the WorkerSinkTask fails when a subscribed topic is deleted and the RebalanceListener tries to fetch the position of such a topic. Note that this doesn't happen whenever a topic is deleted but instead under extreme scenarios like mass deletion of topics. To get around this, Admin Client has been added to WorkerSinkTask. When RebalanceListener is invoked, and a TimeoutException is thrown upon fetching position, it is checked if the topic exists. If it doesn't, then the offset of that TopicPartition isn't committed and the deleted topic name is cached so that any future position fetch for other partitions of the deleted topic doesn't invoke AdminClient again.
```
This PR tries to ascertain that the topic has already been deleted using the Admin API describeTopic. However, none of us here have the expertise on the clients side of things. More specifically =>
1) Is the approach taken in this PR even a viable one?
2) More importantly, is this something that should be fixed on the Clients side of things? If yes, I can create a JIRA ticket for clients and take a call if we need an interim fix for this meanwhile it gets fixed?
Plz let me know.
--
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] vamossagar12 commented on a diff in pull request #13504: KAFKA-14750: Check if topic exists in WorkerSinkTask when committing offsets.
Posted by "vamossagar12 (via GitHub)" <gi...@apache.org>.
vamossagar12 commented on code in PR #13504:
URL: https://github.com/apache/kafka/pull/13504#discussion_r1192983808
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java:
##########
@@ -145,6 +154,10 @@ public WorkerSinkTask(ConnectorTaskId id,
this.isTopicTrackingEnabled = workerConfig.getBoolean(TOPIC_TRACKING_ENABLE_CONFIG);
this.taskStopped = false;
this.workerErrantRecordReporter = workerErrantRecordReporter;
+ Map<String, Object> adminProps = new HashMap<>();
+ adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, workerConfig.bootstrapServers());
Review Comment:
I added the logic to create Admin client in Worker and pass it along to WorkerSinkTask.
--
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] vamossagar12 commented on pull request #13504: KAFKA-14750: Check if topic exists in WorkerSinkTask when committing offsets.
Posted by "vamossagar12 (via GitHub)" <gi...@apache.org>.
vamossagar12 commented on PR #13504:
URL: https://github.com/apache/kafka/pull/13504#issuecomment-1510837804
The tests that have failed belong to either `MirrorConnectorsWithCustomForwardingAdminIntegrationTest` which seems to be failing and is being investigated separately. Also, it seems unrelated to this PR. @yashmayya , can you review this PR whenever you get the chance? Thanks!
--
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] vamossagar12 commented on a diff in pull request #13504: KAFKA-14750: Check if topic exists in WorkerSinkTask when committing offsets.
Posted by "vamossagar12 (via GitHub)" <gi...@apache.org>.
vamossagar12 commented on code in PR #13504:
URL: https://github.com/apache/kafka/pull/13504#discussion_r1169768087
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java:
##########
@@ -700,9 +714,28 @@ private class HandleRebalance implements ConsumerRebalanceListener {
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
log.debug("{} Partitions assigned {}", WorkerSinkTask.this, partitions);
-
+ Set<String> deletedTopics = new HashSet<>();
for (TopicPartition tp : partitions) {
- long pos = consumer.position(tp);
+ if (deletedTopics.contains(tp.topic())) {
+ log.debug("Not Committing offsets for topic-partition {} since the topic {} has been deleted", tp, tp.topic());
+ continue;
+ }
+ long pos;
+ try {
+ pos = consumer.position(tp);
+ } catch (TimeoutException e) {
Review Comment:
I had done some analysis and a plausible explanation this comment onwards: https://issues.apache.org/jira/browse/KAFKA-14750?focusedCommentId=17710077&page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel#comment-17710077
--
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] vamossagar12 commented on a diff in pull request #13504: KAFKA-14750: Check if topic exists in WorkerSinkTask when committing offsets.
Posted by "vamossagar12 (via GitHub)" <gi...@apache.org>.
vamossagar12 commented on code in PR #13504:
URL: https://github.com/apache/kafka/pull/13504#discussion_r1169768346
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java:
##########
@@ -700,9 +714,28 @@ private class HandleRebalance implements ConsumerRebalanceListener {
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
log.debug("{} Partitions assigned {}", WorkerSinkTask.this, partitions);
-
+ Set<String> deletedTopics = new HashSet<>();
for (TopicPartition tp : partitions) {
- long pos = consumer.position(tp);
+ if (deletedTopics.contains(tp.topic())) {
+ log.debug("Not Committing offsets for topic-partition {} since the topic {} has been deleted", tp, tp.topic());
+ continue;
+ }
+ long pos;
+ try {
+ pos = consumer.position(tp);
+ } catch (TimeoutException e) {
+ log.error("TimeoutException occurred when fetching position for topic partition {}. " +
+ "Checking if the topic {} exists", tp, tp.topic());
+ Map<String, TopicDescription> topic = topicAdmin.describeTopics(tp.topic());
+ if (topic.isEmpty()) {
+ log.debug("Not Committing offsets for topic-partition {} since the topic {} has been deleted", tp, tp.topic());
Review Comment:
Yeah would change it. Thanks
--
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] vamossagar12 commented on pull request #13504: KAFKA-14750: Check if topic exists in WorkerSinkTask when committing offsets.
Posted by "vamossagar12 (via GitHub)" <gi...@apache.org>.
vamossagar12 commented on PR #13504:
URL: https://github.com/apache/kafka/pull/13504#issuecomment-1560511613
@viktorsomogyi would you have some time to review this one?
--
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-14750: Check if topic exists in WorkerSinkTask when committing offsets. [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #13504: KAFKA-14750: Check if topic exists in WorkerSinkTask when committing offsets.
URL: https://github.com/apache/kafka/pull/13504
--
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] github-actions[bot] commented on pull request #13504: KAFKA-14750: Check if topic exists in WorkerSinkTask when committing offsets.
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13504:
URL: https://github.com/apache/kafka/pull/13504#issuecomment-1698435395
This PR is being marked as stale since it has not had any activity in 90 days. If you would like to keep this PR alive, please ask a committer for review. If the PR has merge conflicts, please update it with the latest from trunk (or appropriate release branch) <p> If this PR is no longer valid or desired, please feel free to close it. If no activity occurs in the next 30 days, it will be automatically closed.
--
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] vamossagar12 commented on a diff in pull request #13504: KAFKA-14750: Check if topic exists in WorkerSinkTask when committing offsets.
Posted by "vamossagar12 (via GitHub)" <gi...@apache.org>.
vamossagar12 commented on code in PR #13504:
URL: https://github.com/apache/kafka/pull/13504#discussion_r1169762441
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java:
##########
@@ -700,9 +714,28 @@ private class HandleRebalance implements ConsumerRebalanceListener {
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
log.debug("{} Partitions assigned {}", WorkerSinkTask.this, partitions);
-
+ Set<String> deletedTopics = new HashSet<>();
for (TopicPartition tp : partitions) {
- long pos = consumer.position(tp);
+ if (deletedTopics.contains(tp.topic())) {
Review Comment:
Yeah that's the behaviour that the consumer client exhibits. I thin it's due to the fact that a topic deletion forces a consumer rebalance and rebalance cycle typically goes through `onPartitionsRevoked` -> `onPArtitionasAssigned` cycle. I am not sure if this is documented anywhere explictily for topic deletion.
I wouldn't qualify it as a bug though. I think it has been there for a long time. I think Chris had also tried to rope in client team for another suggested change but we haven't heard back yet.
--
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] vamossagar12 commented on a diff in pull request #13504: KAFKA-14750: Check if topic exists in WorkerSinkTask when committing offsets.
Posted by "vamossagar12 (via GitHub)" <gi...@apache.org>.
vamossagar12 commented on code in PR #13504:
URL: https://github.com/apache/kafka/pull/13504#discussion_r1192983841
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java:
##########
@@ -700,9 +714,28 @@ private class HandleRebalance implements ConsumerRebalanceListener {
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
log.debug("{} Partitions assigned {}", WorkerSinkTask.this, partitions);
-
+ Set<String> deletedTopics = new HashSet<>();
for (TopicPartition tp : partitions) {
- long pos = consumer.position(tp);
+ if (deletedTopics.contains(tp.topic())) {
+ log.debug("Not Committing offsets for topic-partition {} since the topic {} has been deleted", tp, tp.topic());
+ continue;
+ }
+ long pos;
+ try {
+ pos = consumer.position(tp);
+ } catch (TimeoutException e) {
+ log.error("TimeoutException occurred when fetching position for topic partition {}. " +
+ "Checking if the topic {} exists", tp, tp.topic());
+ Map<String, TopicDescription> topic = topicAdmin.describeTopics(tp.topic());
+ if (topic.isEmpty()) {
+ log.debug("Not Committing offsets for topic-partition {} since the topic {} has been deleted", tp, tp.topic());
Review Comment:
changed.
--
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] vamossagar12 commented on a diff in pull request #13504: KAFKA-14750: Check if topic exists in WorkerSinkTask when committing offsets.
Posted by "vamossagar12 (via GitHub)" <gi...@apache.org>.
vamossagar12 commented on code in PR #13504:
URL: https://github.com/apache/kafka/pull/13504#discussion_r1207922909
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java:
##########
@@ -700,9 +714,28 @@ private class HandleRebalance implements ConsumerRebalanceListener {
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
log.debug("{} Partitions assigned {}", WorkerSinkTask.this, partitions);
-
+ Set<String> deletedTopics = new HashSet<>();
for (TopicPartition tp : partitions) {
- long pos = consumer.position(tp);
+ if (deletedTopics.contains(tp.topic())) {
Review Comment:
> It's unlikely that Kafka Connect is the only consumer application that is impacted by this scenario
I understand that Connect is not the only consumer application that would be impacted by this behaviour. But ATM, it does impact the users in certain scenarios as described in the ticket. Nonetheless, I would check with engineers from the Client team.
--
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-14750: Check if topic exists in WorkerSinkTask when committing offsets. [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13504:
URL: https://github.com/apache/kafka/pull/13504#issuecomment-2567257656
This PR is being marked as stale since it has not had any activity in 90 days. If you
would like to keep this PR alive, please leave a comment asking for a review. If the PR has
merge conflicts, update it with the latest from the base branch.
<p>
If you are having difficulty finding a reviewer, please reach out on the
[mailing list](https://kafka.apache.org/contact).
<p>
If this PR is no longer valid or desired, please feel free to close it. If no activity
occurs in the next 30 days, it will be automatically closed.
--
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] vamossagar12 commented on a diff in pull request #13504: KAFKA-14750: Check if topic exists in WorkerSinkTask when committing offsets.
Posted by "vamossagar12 (via GitHub)" <gi...@apache.org>.
vamossagar12 commented on code in PR #13504:
URL: https://github.com/apache/kafka/pull/13504#discussion_r1207962325
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java:
##########
@@ -1367,9 +1367,14 @@ public WorkerTask doBuild(Task task,
connectorClientConfigOverridePolicy, kafkaClusterId, ConnectorType.SINK);
KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(consumerProps);
+ Map<String, Object> adminOverrides = adminConfigs(id.connector(), "connector-worker-adminclient-" + id.connector(),
Review Comment:
Ok. Would do that.
--
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] vamossagar12 commented on pull request #13504: KAFKA-14750: Check if topic exists in WorkerSinkTask when committing offsets.
Posted by "vamossagar12 (via GitHub)" <gi...@apache.org>.
vamossagar12 commented on PR #13504:
URL: https://github.com/apache/kafka/pull/13504#issuecomment-1569680586
@C0urante , going by the update from Deng above, we will have to live with a fix like this for some time. That still has the question of the ACL requirements for the new Sink task. Since you called it out as a breaking change needing a KIP, we still need to get around that. I can see 2 options =>
1) We wait for the upcoming major release and start with a KIP.
2) We go ahead and try to fix this behaviour for cases when there's ACL configured or the Admin user has the access to describe topic. For such cases it would work and we should be good. In case the describeTopics call fails let's say due to a `ClusterAuthorizationException`, `TopicAuthorizationException`(wrapped in `ConnectException`), and if we assume that the topic exists, then we still won't have the position computed for that topic partition. Which means we can't update anything in `lastCommittedOffsets` and `currentOffsets` (eventually `sinkTaskMetricsGroup`) which is still an inconsistent state I feel. What if we throw the exception and add a log line to consider adding ACL to describe topic for the admin user ? Is something like that acceptable?
Note that, the consolidation of the admin client for SinkTasks is still pending on me. But before that I wanted to know what the next steps should be on this one.
--
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] vamossagar12 commented on a diff in pull request #13504: KAFKA-14750: Check if topic exists in WorkerSinkTask when committing offsets.
Posted by "vamossagar12 (via GitHub)" <gi...@apache.org>.
vamossagar12 commented on code in PR #13504:
URL: https://github.com/apache/kafka/pull/13504#discussion_r1169764236
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java:
##########
@@ -700,9 +714,28 @@ private class HandleRebalance implements ConsumerRebalanceListener {
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
log.debug("{} Partitions assigned {}", WorkerSinkTask.this, partitions);
-
+ Set<String> deletedTopics = new HashSet<>();
for (TopicPartition tp : partitions) {
- long pos = consumer.position(tp);
+ if (deletedTopics.contains(tp.topic())) {
+ log.debug("Not Committing offsets for topic-partition {} since the topic {} has been deleted", tp, tp.topic());
Review Comment:
Ok. I saw `lastCommittedOffsets` map being updated and assumed this is about committing offsets. Will change 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
[GitHub] [kafka] yashmayya commented on a diff in pull request #13504: KAFKA-14750: Check if topic exists in WorkerSinkTask when committing offsets.
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya commented on code in PR #13504:
URL: https://github.com/apache/kafka/pull/13504#discussion_r1169647018
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java:
##########
@@ -145,6 +154,10 @@ public WorkerSinkTask(ConnectorTaskId id,
this.isTopicTrackingEnabled = workerConfig.getBoolean(TOPIC_TRACKING_ENABLE_CONFIG);
this.taskStopped = false;
this.workerErrantRecordReporter = workerErrantRecordReporter;
+ Map<String, Object> adminProps = new HashMap<>();
+ adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, workerConfig.bootstrapServers());
Review Comment:
Connect supports overriding Kafka client configs per connector - see [KIP-458](https://cwiki.apache.org/confluence/display/KAFKA/KIP-458%3A+Connector+Client+Config+Override+Policy). If we're using an admin client for a specific connector (or task), it should respect these overrides - the connector may even be targeting a Kafka cluster different from the Connect cluster's backing Kafka cluster and we cannot use worker configurations to create Kafka clients for connectors. It would probably be cleaner to instantiate this admin client in the `Worker` and pass it to the `WorkerSinkTask` during creation. An example from `WorkerSourceTask` creation that might help - https://github.com/apache/kafka/blob/454b72161a76b1687a1263157d7cc30a1bdb2506/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java#L1424-L1446
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java:
##########
@@ -700,9 +714,28 @@ private class HandleRebalance implements ConsumerRebalanceListener {
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
log.debug("{} Partitions assigned {}", WorkerSinkTask.this, partitions);
-
+ Set<String> deletedTopics = new HashSet<>();
for (TopicPartition tp : partitions) {
- long pos = consumer.position(tp);
+ if (deletedTopics.contains(tp.topic())) {
+ log.debug("Not Committing offsets for topic-partition {} since the topic {} has been deleted", tp, tp.topic());
+ continue;
+ }
+ long pos;
+ try {
+ pos = consumer.position(tp);
+ } catch (TimeoutException e) {
Review Comment:
This seems kinda weird to me too - why would the consumer time out on trying to fetch the position for a topic partition that doesn't exist anymore? Is this expected / documented or just a poorly handled case in the consumer client?
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java:
##########
@@ -700,9 +714,28 @@ private class HandleRebalance implements ConsumerRebalanceListener {
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
log.debug("{} Partitions assigned {}", WorkerSinkTask.this, partitions);
-
+ Set<String> deletedTopics = new HashSet<>();
for (TopicPartition tp : partitions) {
- long pos = consumer.position(tp);
+ if (deletedTopics.contains(tp.topic())) {
+ log.debug("Not Committing offsets for topic-partition {} since the topic {} has been deleted", tp, tp.topic());
Review Comment:
We don't commit offsets in `onPartitionsAssigned` so I'm not sure what the purpose of this log is?
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java:
##########
@@ -700,9 +714,28 @@ private class HandleRebalance implements ConsumerRebalanceListener {
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
log.debug("{} Partitions assigned {}", WorkerSinkTask.this, partitions);
-
+ Set<String> deletedTopics = new HashSet<>();
for (TopicPartition tp : partitions) {
- long pos = consumer.position(tp);
+ if (deletedTopics.contains(tp.topic())) {
+ log.debug("Not Committing offsets for topic-partition {} since the topic {} has been deleted", tp, tp.topic());
+ continue;
+ }
+ long pos;
+ try {
+ pos = consumer.position(tp);
+ } catch (TimeoutException e) {
+ log.error("TimeoutException occurred when fetching position for topic partition {}. " +
+ "Checking if the topic {} exists", tp, tp.topic());
+ Map<String, TopicDescription> topic = topicAdmin.describeTopics(tp.topic());
+ if (topic.isEmpty()) {
+ log.debug("Not Committing offsets for topic-partition {} since the topic {} has been deleted", tp, tp.topic());
Review Comment:
Same thoughts as above regarding the purpose of this log line.
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java:
##########
@@ -700,9 +714,28 @@ private class HandleRebalance implements ConsumerRebalanceListener {
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
log.debug("{} Partitions assigned {}", WorkerSinkTask.this, partitions);
-
+ Set<String> deletedTopics = new HashSet<>();
for (TopicPartition tp : partitions) {
- long pos = consumer.position(tp);
+ if (deletedTopics.contains(tp.topic())) {
Review Comment:
I'm not sure I understand why the `onPartitionsAssigned` rebalance listener hook will be called for partitions of topics that are deleted? Is this expected behavior or a bug in the consumer client? If it's expected, is there any documentation around this? If it's a bug, I think it might be better to address the bug in the consumer client rather than introducing a workaround for it in Connect?
--
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] vamossagar12 commented on a diff in pull request #13504: KAFKA-14750: Check if topic exists in WorkerSinkTask when committing offsets.
Posted by "vamossagar12 (via GitHub)" <gi...@apache.org>.
vamossagar12 commented on code in PR #13504:
URL: https://github.com/apache/kafka/pull/13504#discussion_r1207577607
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java:
##########
@@ -700,9 +714,28 @@ private class HandleRebalance implements ConsumerRebalanceListener {
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
log.debug("{} Partitions assigned {}", WorkerSinkTask.this, partitions);
-
+ Set<String> deletedTopics = new HashSet<>();
for (TopicPartition tp : partitions) {
- long pos = consumer.position(tp);
+ if (deletedTopics.contains(tp.topic())) {
Review Comment:
This block of code uses already existing logic within `TopicAdmin` to figure out if a topic exists or not. https://github.com/apache/kafka/blob/6d72c26731fe69955127a90e3d43f6d9eb41e2d3/connect/runtime/src/main/java/org/apache/kafka/connect/util/TopicAdmin.java#L503-L511. I am just caching it because if we established that a topic is deleted for partition p1, then we don't need to check again for other partitions of the same topic within this flow.
--
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 #13504: KAFKA-14750: Check if topic exists in WorkerSinkTask when committing offsets.
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on code in PR #13504:
URL: https://github.com/apache/kafka/pull/13504#discussion_r1206885931
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java:
##########
@@ -695,9 +705,28 @@ private class HandleRebalance implements ConsumerRebalanceListener {
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
log.debug("{} Partitions assigned {}", WorkerSinkTask.this, partitions);
-
+ Set<String> deletedTopics = new HashSet<>();
for (TopicPartition tp : partitions) {
- long pos = consumer.position(tp);
+ if (deletedTopics.contains(tp.topic())) {
+ log.debug("Not assigning offsets for topic-partition {} since the topic {} has been deleted", tp, tp.topic());
+ continue;
+ }
+ long pos;
+ try {
+ pos = consumer.position(tp);
+ } catch (TimeoutException e) {
+ log.error("TimeoutException occurred when fetching position for topic partition {}. " +
+ "Checking if the topic {} exists", tp, tp.topic());
+ Map<String, TopicDescription> topic = topicAdmin.describeTopics(tp.topic());
Review Comment:
This adds new ACL requirements for sink connectors' admin clients, which is a breaking change and cannot be done until the next major release.
We also need to handle generic exceptions thrown during this call. Probably safest to assume that the topic exists if we fail to describe it.
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java:
##########
@@ -700,9 +714,28 @@ private class HandleRebalance implements ConsumerRebalanceListener {
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
log.debug("{} Partitions assigned {}", WorkerSinkTask.this, partitions);
-
+ Set<String> deletedTopics = new HashSet<>();
for (TopicPartition tp : partitions) {
- long pos = consumer.position(tp);
+ if (deletedTopics.contains(tp.topic())) {
Review Comment:
We should not be relying on undocumented behavior like this without confirmation from someone familiar with the clients library that it's intentional, or at the very least, that it won't change in the future without a KIP.
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java:
##########
@@ -1367,9 +1367,14 @@ public WorkerTask doBuild(Task task,
connectorClientConfigOverridePolicy, kafkaClusterId, ConnectorType.SINK);
KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(consumerProps);
+ Map<String, Object> adminOverrides = adminConfigs(id.connector(), "connector-worker-adminclient-" + id.connector(),
Review Comment:
We already construct an admin client for sink connectors [if they use a DLQ topic](https://github.com/apache/kafka/blob/6d72c26731fe69955127a90e3d43f6d9eb41e2d3/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java#L947-L951).
We should create at most one admin client per sink task, which means probably refactoring the `sinkTaskReporters` method to accept an admin client if we're going to be creating one unconditionally here.
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java:
##########
@@ -700,9 +714,28 @@ private class HandleRebalance implements ConsumerRebalanceListener {
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
log.debug("{} Partitions assigned {}", WorkerSinkTask.this, partitions);
-
+ Set<String> deletedTopics = new HashSet<>();
for (TopicPartition tp : partitions) {
- long pos = consumer.position(tp);
+ if (deletedTopics.contains(tp.topic())) {
Review Comment:
Also, this entire block relies on racy logic to detect deleted topics that may fail if a topic is deleted after this check takes place. Again, we should reach out to someone familiar with the clients library. It's unlikely that Kafka Connect is the only consumer application that is impacted by this scenario and it's better to fix this at the clients level instead of implementing Connect-specific workarounds.
--
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] vamossagar12 commented on pull request #13504: KAFKA-14750: Check if topic exists in WorkerSinkTask when committing offsets.
Posted by "vamossagar12 (via GitHub)" <gi...@apache.org>.
vamossagar12 commented on PR #13504:
URL: https://github.com/apache/kafka/pull/13504#issuecomment-1565419220
> Looks good to me overall. It indeed would be more convenient if the consumer didn't return dead topics because using an admin client is a bit resource intensive, however as I understand this scenario shouldn't be happening too much anyways, so it's fine.
>
> @C0urante would you please take quick look too?
Thanks @viktorsomogyi . Yeah I tried `listTopics` API from consumers which as per docs says
> This method will issue a remote call to the server.
but even that couldn't prevent this error. I agree Admin APIs are resource intensive and as Chris pointed out, bring in ACL requirements as well, I have looped in Deng to validate the findings (and fix on Clients/Admin side if needed).
--
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] vamossagar12 commented on pull request #13504: KAFKA-14750: Check if topic exists in WorkerSinkTask when committing offsets.
Posted by "vamossagar12 (via GitHub)" <gi...@apache.org>.
vamossagar12 commented on PR #13504:
URL: https://github.com/apache/kafka/pull/13504#issuecomment-1546785355
Test failures seem unrelated.
--
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] vamossagar12 commented on pull request #13504: KAFKA-14750: Check if topic exists in WorkerSinkTask when committing offsets.
Posted by "vamossagar12 (via GitHub)" <gi...@apache.org>.
vamossagar12 commented on PR #13504:
URL: https://github.com/apache/kafka/pull/13504#issuecomment-1569597113
Thanks @dengziming for your confirmation. Regarding
> one question is why only handle TimeoutException when getting offset
The observation was that when a topic is deleted, when we tried to fetch the `position` of a partition of it, we were getting a `TimeoutException`. The other exceptions didn't seem relevant in this case and I feel we should let those exceptions bubble up and let the connector fail.
--
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] vamossagar12 commented on a diff in pull request #13504: KAFKA-14750: Check if topic exists in WorkerSinkTask when committing offsets.
Posted by "vamossagar12 (via GitHub)" <gi...@apache.org>.
vamossagar12 commented on code in PR #13504:
URL: https://github.com/apache/kafka/pull/13504#discussion_r1169767646
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java:
##########
@@ -700,9 +714,28 @@ private class HandleRebalance implements ConsumerRebalanceListener {
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
log.debug("{} Partitions assigned {}", WorkerSinkTask.this, partitions);
-
+ Set<String> deletedTopics = new HashSet<>();
for (TopicPartition tp : partitions) {
- long pos = consumer.position(tp);
+ if (deletedTopics.contains(tp.topic())) {
+ log.debug("Not Committing offsets for topic-partition {} since the topic {} has been deleted", tp, tp.topic());
+ continue;
+ }
+ long pos;
+ try {
+ pos = consumer.position(tp);
+ } catch (TimeoutException e) {
Review Comment:
It works under normal scenarios. I tested by deleting 10 topics in one go and still didn't notice this error happening. In this case, when a large number of topics gets deleted is when we see this error. Note that I don't know what is the defintion of large but going by the update of the reporter of the ticket, they experience it with a few dozens of topics deleted.
--
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