[KAFKA-17742] Move DelayedShareFetchPurgatory declaration to ReplicaManager

• Type: Sub-task
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 4.0.0
• Assignee: Abhinav Dixit
• Reporter: Abhinav Dixit
• Created: 09/Oct/24 05:11
• Updated: 17/Oct/24 20:58
• Resolved: 17/Oct/24 20:58

Description

Hi, as per comments https://github.com/apache/kafka/pull/17380#issuecomment-2399040546 and https://github.com/apache/kafka/pull/17380#issuecomment-2400431605 , we need to do the following - 

1. Declare the delayed share fetch purgatory inside ReplicaManager along with the existing purgatories. We don't need a reference to this purgatory outside of ReplicaManager, we can just add accessor methods.
2. Currently, we check the produce/fetch purgatory when a replica becomes the follower or a replica is deleted from a broker through ReplicaManager. applyLocalFollowersDelta and ReplicaManager.applyDelta. We should do the same for share purgatory.
3. Perform a checkAndComplete when HWM is updated
4. Perform a shutdown of the delayed share fetch purgatory like the others in ReplicaManager

Issue Links

• links to: GitHub Pull Request #17437