[GitHub] [kafka] kumarpritam863 opened a new pull request, #12622: Update WorkerSinkTask.java

[GitBox <gi...@apache.org>]

kumarpritam863 opened a new pull request, #12622:
URL: https://github.com/apache/kafka/pull/12622

   In case of the revocation of partitions, the updation of "**partition count**" metrics is being done before updating the new set of assignments. "**invokePartitionsRevoked**" method of "**onJoinComplete**" function of "**ConsumerCoordinator**" class is being called before the "
   **subscriptions.assignFromSubscribed(assignedPartitions)**" of the same class. As a result of which the old assigned partition count is getting updated again and again even after future rebalances.


-- 
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 pull request #12622: Update WorkerSinkTask.java

[GitBox <gi...@apache.org>]

C0urante commented on PR #12622:
URL: https://github.com/apache/kafka/pull/12622#issuecomment-1245423635

   Thanks @kumarpritam863, this is an important fix. I did some research on the history behind incremental consumer rebalancing and how it relates to the rebalance listener API, and found that [this PR](https://github.com/apache/kafka/pull/6884#issue-452279459) contained some great details, including an explicit confirmation on the expected ordering of callback invocation:
   
   > The ordering of the callback would be the following:
   a. Callback onPartitionsRevoked / onPartitionsLost triggered.
   b. Update the assignment (both revoked and added).
   c. Callback onPartitionsAssigned triggered.
   
   One case I'm wondering about is when the task fails, which would cause its consumer to leave the group. I believe that this would cause `onPartitionsRevoked` to be invoked in the rebalance listener, but not `onPartitionsAssigned`, so the metrics for the task (which are still available for failed tasks until they are explicitly revoked from the worker) would become inaccurate.
   
   Could we forcibly reset the partition count metric to zero in `WorkerSinkTask::close` to address this case? (Probably with a brief comment explaining why we have this special logic).
   
   It'd also be nice to see some tests for this logic, both to verify correctness and help prevent regression.


-- 
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] kumarpritam863 closed pull request #12622: Update WorkerSinkTask.java

[GitBox <gi...@apache.org>]

kumarpritam863 closed pull request #12622: Update WorkerSinkTask.java
URL: https://github.com/apache/kafka/pull/12622


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