adixitconfluent commented on code in PR #17437:
URL: https://github.com/apache/kafka/pull/17437#discussion_r1798155913
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1810,7 +1811,7 @@ &&
checkForStartOffsetWithinBatch(inFlightBatch.firstOffset(), inFlightBatch.las
// If we have an acquisition lock timeout for a
share-partition, then we should check if
// there is a pending share fetch request for the
share-partition and complete it.
DelayedShareFetchKey delayedShareFetchKey = new
DelayedShareFetchGroupKey(groupId, topicIdPartition.topicId(),
topicIdPartition.partition());
-
delayedShareFetchPurgatory.checkAndComplete(delayedShareFetchKey);
+
replicaManager.completeDelayedShareFetchRequest(delayedShareFetchKey);
Review Comment:
@apoorvmittal10 , we've added
`replicaManager.completeDelayedShareFetchRequest(delayedShareFetchKey)` on
acqusiition lock timeout functionality. This is independent of
`SharePartitionManager` functionalities like acknowledge, release acquired
records on session close etc. Hence, if we're not passing `ReplicaManager`
object, we'll have to pass `SharePartitionManager` object to `SharePartition`
class if we have to steer from `ReplicaManager`. Let me know if you have any
other alternatives.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
