junrao commented on code in PR #17534:
URL: https://github.com/apache/kafka/pull/17534#discussion_r1806854706
##########
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:
While you are at this class, in the following code, `return` seems
unnecessary.
```
sharePartition.maybeInitialize().whenComplete((result,
throwable) -> {
if (throwable != null) {
maybeCompleteInitializationWithException(sharePartitionKey,
shareFetchData.future(), throwable);
return;
}
```
--
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]
