You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/08/16 13:31:07 UTC
[GitHub] [kafka] cadonna opened a new pull request, #12520: KAFKA-10199: Remove tasks from state updater on revocation
cadonna opened a new pull request, #12520:
URL: https://github.com/apache/kafka/pull/12520
Removes tasks from the state updater when the input partitions
of the tasks are revoked during a rebalance.
### 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] guozhangwang merged pull request #12520: KAFKA-10199: Remove tasks from state updater on revocation
Posted by GitBox <gi...@apache.org>.
guozhangwang merged PR #12520:
URL: https://github.com/apache/kafka/pull/12520
--
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] guozhangwang commented on a diff in pull request #12520: KAFKA-10199: Remove tasks from state updater on revocation
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on code in PR #12520:
URL: https://github.com/apache/kafka/pull/12520#discussion_r946996425
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -838,6 +838,21 @@ void handleRevocation(final Collection<TopicPartition> revokedPartitions) {
if (firstException.get() != null) {
throw firstException.get();
}
+
+ removeRevokedTasksFromStateUpdater(remainingRevokedPartitions);
Review Comment:
I think it's better to call this function before line 758
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -838,6 +838,21 @@ void handleRevocation(final Collection<TopicPartition> revokedPartitions) {
if (firstException.get() != null) {
throw firstException.get();
}
+
+ removeRevokedTasksFromStateUpdater(remainingRevokedPartitions);
+ }
+
+ private void removeRevokedTasksFromStateUpdater(final Set<TopicPartition> remainingRevokedPartitions) {
+ if (stateUpdater != null) {
+ for (final Task restoringTask : stateUpdater.getTasks()) {
+ if (restoringTask.isActive()) {
+ if (remainingRevokedPartitions.containsAll(restoringTask.inputPartitions())) {
Review Comment:
Should we also remove those partitions from the `remainingRevokedPartitions` as well?
--
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] cadonna commented on a diff in pull request #12520: KAFKA-10199: Remove tasks from state updater on revocation
Posted by GitBox <gi...@apache.org>.
cadonna commented on code in PR #12520:
URL: https://github.com/apache/kafka/pull/12520#discussion_r947117136
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -838,6 +838,21 @@ void handleRevocation(final Collection<TopicPartition> revokedPartitions) {
if (firstException.get() != null) {
throw firstException.get();
}
+
+ removeRevokedTasksFromStateUpdater(remainingRevokedPartitions);
Review Comment:
Ack
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -838,6 +838,21 @@ void handleRevocation(final Collection<TopicPartition> revokedPartitions) {
if (firstException.get() != null) {
throw firstException.get();
}
+
+ removeRevokedTasksFromStateUpdater(remainingRevokedPartitions);
+ }
+
+ private void removeRevokedTasksFromStateUpdater(final Set<TopicPartition> remainingRevokedPartitions) {
+ if (stateUpdater != null) {
+ for (final Task restoringTask : stateUpdater.getTasks()) {
+ if (restoringTask.isActive()) {
+ if (remainingRevokedPartitions.containsAll(restoringTask.inputPartitions())) {
Review Comment:
Ack
--
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