apoorvmittal10 commented on code in PR #17534:
URL: https://github.com/apache/kafka/pull/17534#discussion_r1807763887
##########
core/src/main/java/kafka/server/share/SharePartitionManager.java:
##########
@@ -547,31 +522,11 @@ private static String
partitionsToLogString(Collection<TopicIdPartition> partiti
return ShareSession.partitionsToLogString(partitions,
log.isTraceEnabled());
}
- /**
- * Recursive function to process all the fetch requests present inside the
fetch queue
- */
// Visible for testing.
- void maybeProcessFetchQueue() {
- if (!acquireProcessFetchQueueLock()) {
- // The queue is already being processed hence avoid re-triggering.
- return;
- }
-
- ShareFetchData shareFetchData = fetchQueue.poll();
- if (shareFetchData == null) {
- // No more requests to process, so release the lock. Though we
should not reach here as the lock
- // is acquired only when there are requests in the queue. But
still, it's safe to release the lock.
- releaseProcessFetchQueueLock();
- return;
- }
-
+ void processShareFetch(ShareFetchData shareFetchData) {
Review Comment:
Done. Though I am reafactoring the initilization handling in Share Partition
Manager as we need to address the concern when Share Partition is loaded late
and request is added to purgatory, etc. The PR will be out soon.
--
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]
