KAFKA-20736: Improve share session handling on leader change#22766
Conversation
| partitionsWithUpdatedLeaderInfo.put(tip.topicPartition(), new Metadata.LeaderIdAndEpoch( | ||
| Optional.of(partitionData.currentLeader().leaderId()), Optional.of(partitionData.currentLeader().leaderEpoch()))); | ||
|
|
||
| maybeUpdateLeaderCache(tip, partitionData.currentLeader().leaderId(), partitionData.currentLeader().leaderEpoch()); |
There was a problem hiding this comment.
could we have the case that the partition is not assigned anymore by the time we hit this? Metadata does consider the subscription and retains accordingly (the retainTopic logic), so wondering if we need something similar to be aware of the subscription while maintaining this cache?
There was a problem hiding this comment.
The awareness of the assigned partitions comes from SubscriptionState in the fetchablePartitions() call at the start of the poll. So, yes, I think it is possible that a partition might not be assigned any more by the time we hit this. However, because share groups have intentionally looser semantics than consumer groups, this is expected. If a share consumer receives some records from an assigned partition, and then the partition is revoked, it is still able to continue delivery of those records and complete their acknowledgements, potentially for a significant length of time (renew acks).
| int nodeId = leader.leaderId; | ||
| Node node = cluster.nodeById(nodeId); | ||
| if (node == null) { | ||
| continue; |
There was a problem hiding this comment.
here is the case that we had the leader in the sessionLeaderMap, but the node is not in the cluster metadata, don't we need to clear our local leader cache that has stale info? and also request metadata I guess, something changed from the view we had.
There was a problem hiding this comment.
Thanks for the review. I've been reworking this code in parallel to ensure I clean up as the cluster changes. For example, if a broker is decommissioned under a running share consumer, it needs to complete pending acks which it will never be able to send. I'll push a new commit shortly.
|
Flaky |
On further investigation, this PR probably DOES introduce the flakiness. I'll sort it out in this PR. |
ShivsundarR
left a comment
ShivsundarR
left a comment
There was a problem hiding this comment.
Thanks for the PR @AndrewJSchofield ! Looks like a very nice improvement. Just a minor nit comment. Otherwise looks good.
ShivsundarR
left a comment
ShivsundarR
left a comment
There was a problem hiding this comment.
Looks good to me! Thanks for the PR.
apoorvmittal10
left a comment
apoorvmittal10
left a comment
There was a problem hiding this comment.
This will be helpful, thanks for the PR. LGTM!
4 participants
In some situations, the share consumer does not tidy up share sessions
sufficiently on leadership change. In particular, if a broker loses
leadership of a partition, the share consumer should send a further
ShareFetch/ShareAcknowledge to remove the partition from its share
session. In most cases, this removal happens naturally, but it depends
upon whether the share consumer has other requests to send to that
broker. This PR tidies up the maintenance of share sessions when
leadership changes occur.
Reviewers: Lianet Magrans lmagrans@confluent.io, Shivsundar R
shr@confluent.io, Apoorv Mittal apoorvmittal10@gmail.com