KAFKA-17703: Moved DelayedActionsQueue outside DelayedShareFetch

[adixitconfluent]

About

In reference to comments #17177 (comment) and #17177 (comment) , this PR addresses the following -

1. Move ActionQueue outside DelayedShareFetch class to SharePartitionManager where SharePartitionManager has the ability to add a delayed action to the ActionQueue.

2. Add TopicPartition as a key for delayed share fetch along with SharePartition (that is already present right now). This will be utilized later when we add the code to know if new data has been added to the topic partition.

Testing

The code has been tested with the help of unit tests.

[apoorvmittal10]

Thanks for the PR. Left some comments.

[junrao]

@adixitconfluent : Thanks for the PR. A couple of comments.

[junrao]

@adixitconfluent : 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. Should we do the same for share purgatory?

[adixitconfluent]

@adixitconfluent : 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. Should we do the same for share purgatory?

@junrao , yes, I think we should do it. Moreover, with more use cases of delayed share fetch purgatory within ReplicaManager functionalities like -

1. Above use cases you mentioned
2. Perform a checkAndComplete when HWM is updated
3. Perform a shutdown of the delayed share fetch purgatory like the others in ReplicaManager
   I feel its better we 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. What do you think?

[junrao]

@adixitconfluent : Thanks for the updated PR. A few more comments.

Also, there seems to be another existing issue. If a sharePartition is being initialized, any pending shareFetchRequest is blocked in the purgatory. So, after the initialization completes, we need to trigger a check in the share fetch purgatory to wake up pending delayed operations.

[junrao]

TopicIdPartition contains both topicId and name. We just need the former in the key. Ditto for DelayedShareFetchPartitionKey.

[adixitconfluent]

I agree we don't need the name, but we will need the partition number, so I'll add Uuid topicId and int partition to both DelayedShareFetchGroupKey and DelayedShareFetchPartitionKey.

[junrao]

Perform a shutdown of the delayed share fetch purgatory like the others in ReplicaManager
I feel its better we 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. What do you think?

We can do that too if it makes code simpler.

[adixitconfluent]

Also, there seems to be another existing issue. If a sharePartition is being initialized, any pending shareFetchRequest is blocked in the purgatory. So, after the initialization completes, we need to trigger a check in the share fetch purgatory to wake up pending delayed operations.

Hi @junrao, we are aware about the share partition initialization issues and we have a JIRA for it https://issues.apache.org/jira/browse/KAFKA-17510 , I've added In the description what you mentioned and it should be completed as part of that JIRA

[adixitconfluent]

We can do that too if it makes code simpler.

Thanks @junrao , I've created a JIRA https://issues.apache.org/jira/browse/KAFKA-17742 to address this in my future PRs.

[apoorvmittal10]

Re-opening PR to trigger build.

[junrao]

@adixitconfluent : Thanks for the updated PR. LGTM. Waiting for the tests to complete.