machi1990 commented on code in PR #13665:
URL: https://github.com/apache/kafka/pull/13665#discussion_r1198712020
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java:
##########
@@ -346,6 +355,10 @@ private Exception invokePartitionsRevoked(final
SortedSet<TopicPartition> revoke
final long startMs = time.milliseconds();
listener.onPartitionsRevoked(revokedPartitions);
sensors.revokeCallbackSensor.record(time.milliseconds() - startMs);
+ // remove the offset metadata cache for revoked partitions
+ for (TopicPartition revokedPartition: revokedPartitions) {
+ this.partitionOffsetsCache.remove(revokedPartition);
+ }
Review Comment:
> invokePartitionsRevoked won't happen in eagar consumer protocol after
partition assignment in each rebalance (check
ConsumerCoordinator#onJoinComplete).
> This is correct. This is only called only when cooperative protocol is
used.
An edit on what I said earlier, my statement above is only true the logic
done in ConsumerCoordinator#onJoinComplete.
Eager consumer protocol requires a consumer to always revoke its owned
partitions during a rebalance event: according to this javadoc
https://github.com/apache/kafka/blob/401fb417bf60864e6d380f979d268d895c5ad727/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerPartitionAssignor.java#L250-L251
and that happens here
https://github.com/apache/kafka/blob/401fb417bf60864e6d380f979d268d895c5ad727/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java#L849-L871
Do you think we are good in that regard? @showuon
--
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
