adixitconfluent commented on code in PR #22780:
URL: https://github.com/apache/kafka/pull/22780#discussion_r3543294616
##########
core/src/main/java/kafka/server/share/DelayedShareFetch.java:
##########
@@ -332,43 +417,68 @@ private void
resetFetchOffsetMetadataForRemoteFetchPartitions(
});
}
- /**
- * Try to complete the fetch operation if we can acquire records for any
partition in the share fetch request.
- */
@Override
public boolean tryComplete() {
+ // Check if there's a pending fetch from a previous tryComplete
invocation.
+ // This happens when the async read didn't complete immediately and we
stored the future.
+ // If the async read is still in progress, stay in purgatory.
Otherwise, process the result.
+ if (pendingFetch != null) {
+ if (!pendingFetch.isDone()) {
+ // Async read still in flight - partition locks are still held.
+ return false;
+ }
+ // Async read completed - retrieve result and process
+ return maybeCompleteAsyncFetch(pendingFetch, partitionsAcquired);
+ }
+
// Check to see if the remote fetch is in flight. If there is an in
flight remote fetch we want to resolve it first.
+ // Remote fetch can only happen after async fetch succeeds.
if (pendingRemoteFetchesOpt.isPresent()) {
return maybeCompletePendingRemoteFetch();
}
+ // Try to acquire partition locks for all partitions in the request
LinkedHashMap<TopicIdPartition, Long> topicPartitionData =
acquirablePartitions(sharePartitions);
try {
if (!topicPartitionData.isEmpty()) {
+ // Successfully acquired locks for one or more partitions
// Update the metric to record the time taken to acquire the
locks for the share partitions.
updateAcquireElapsedTimeMetric();
- // In case, fetch offset metadata doesn't exist for one or
more topic partitions, we do a
- // readFromLog to populate the offset metadata and update the
fetch offset metadata for
- // those topic partitions.
- LinkedHashMap<TopicIdPartition, LogReadResult> readResponse =
maybeReadFromLog(topicPartitionData);
- // Store the remote fetch info for the topic partitions for
which we need to perform remote fetch.
- LinkedHashMap<TopicIdPartition, LogReadResult>
remoteStorageFetchInfoMap =
maybePrepareRemoteStorageFetchInfo(topicPartitionData, readResponse);
-
- if (!remoteStorageFetchInfoMap.isEmpty()) {
- return maybeProcessRemoteFetch(topicPartitionData,
remoteStorageFetchInfoMap);
- }
- maybeUpdateFetchOffsetMetadata(topicPartitionData,
readResponse);
- if (anyPartitionHasLogReadError(readResponse) ||
isMinBytesSatisfied(topicPartitionData,
partitionMaxBytesStrategy.maxBytes(shareFetch.fetchParams().maxBytes,
topicPartitionData.keySet(), topicPartitionData.size()))) {
- partitionsAcquired = topicPartitionData;
- localPartitionsAlreadyFetched = readResponse;
- return forceComplete();
+ // In case, fetch offset metadata doesn't exist for one or
more topic partitions, we do an
+ // async read to populate the offset metadata and update the
fetch offset metadata for
+ // those topic partitions. This uses readAsync with
readRemote=false.
+ CompletableFuture<LinkedHashMap<TopicIdPartition,
LogReadResult>> readFuture = maybeReadFromLog(topicPartitionData);
+ // Check if future completed synchronously or needs async
handling
+ if (readFuture.isDone()) {
+ // Fast path, async read completed synchronously.
+ // Process the result immediately without storing the
future or registering callbacks
+ return maybeCompleteAsyncFetch(readFuture,
topicPartitionData);
} else {
- log.debug("minBytes is not satisfied for the share fetch
request for group {}, member {}, " +
- "topic partitions {}", shareFetch.groupId(),
shareFetch.memberId(),
- sharePartitions.keySet());
- releasePartitionLocks(topicPartitionData.keySet());
+ // Store the future and register a callback. Partition
locks remain held during the async operation.
+ // When the async read completes, the callback will
trigger a purgatory retry which will
+ // invoke tryComplete() again, where we'll process the
result.
+ partitionsAcquired = topicPartitionData;
+ pendingFetch = readFuture;
+
+ // Register callback to be invoked when async read
completes
+ // This callback executes on the thread that completes the
future (IO thread pool)
+ readFuture.whenComplete((result, throwable) -> {
+ if (throwable != null) {
+ log.error("Async read failed for group {}, member
{}",
+ shareFetch.groupId(), shareFetch.memberId(),
throwable);
+ // Error is handled in maybeCompleteAsyncFetch()
via get() exception handling
+ }
+ // maybeCompleteAsyncFetch() handles: get(), remote
fetch check, minBytes check
+ // and calls forceComplete() internally when the
request should complete
+ maybeCompleteAsyncFetch(readFuture,
topicPartitionData);
+ // Note: return value is ignored since we're in
callback context, not tryComplete()
+ // forceComplete() is called internally by
processAsyncReadResponse() when needed.
+ });
+ // Return false to keep request in purgatory - locks
remain held
+ return false;
}
Review Comment:
The callback calls `maybeCompleteAsyncFetch` directly on the IO thread that
completes the future, bypassing the purgatory lock. Meanwhile, purgatory can
invoke safeTryComplete() → tryComplete() on a watcher thread under the lock.
Both threads can concurrently enter `maybeCompleteAsyncFetch` and mutate shared
state (pendingFetch, partitionsAcquired, localPartitionsAlreadyFetched,
releasePartitionLocks).`forceComplete` prevents double completion, but all the
state mutations before it are unprotected.
Fix: The callback should go through purgatory's checkAndComplete instead of
directly calling maybeCompleteAsyncFetch().
--
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]
