[GitHub] [kafka] vamossagar12 commented on a diff in pull request #13504: KAFKA-14750: Check if topic exists in WorkerSinkTask when committing offsets.

["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