hachikuji commented on code in PR #12603:
URL: https://github.com/apache/kafka/pull/12603#discussion_r969124759
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java:
##########
@@ -799,64 +804,92 @@ protected boolean onJoinPrepare(Timer timer, int
generation, String memberId) {
}
}
+ Optional<Exception> exception = revokePartitions(partitionsToRevoke,
generation, memberId);
+
+ isLeader = false;
+ subscriptions.resetGroupSubscription();
+ joinPrepareTimer = null;
+ autoCommitOffsetRequestFuture = null;
+ timer.update();
+
+ if (exception.isPresent()) {
+ throw new KafkaException("User rebalance callback throws an
error", exception.get());
+ }
+ return true;
+ }
+
+ private SortedSet<TopicPartition> getPartitionsToRevoke(RebalanceProtocol
protocol, int generation, String memberId) {
+ SortedSet<TopicPartition> partitions = new TreeSet<>(COMPARATOR);
+ if (generation == Generation.NO_GENERATION.generationId ||
+ memberId.equals(Generation.NO_GENERATION.memberId)) {
+ partitions.addAll(subscriptions.assignedPartitions());
+ return partitions;
+ }
+
+ switch (protocol) {
+ case EAGER:
+ partitions.addAll(subscriptions.assignedPartitions());
+ break;
+
+ case COOPERATIVE:
+ // Delay the partition revocation because we don't revoke the
already owned partitions
Review Comment:
I filed this issue: https://issues.apache.org/jira/browse/KAFKA-14224.
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
