apoorvmittal10 commented on code in PR #22780:
URL: https://github.com/apache/kafka/pull/22780#discussion_r3546458375
##########
core/src/main/java/kafka/server/share/DelayedShareFetch.java:
##########
@@ -572,24 +688,108 @@ private void updateAcquireElapsedTimeMetric() {
acquireStartTimeMs = currentTimeMs;
}
+ private boolean maybeCompleteAsyncFetch(
+ CompletableFuture<LinkedHashMap<TopicIdPartition, LogReadResult>>
completedFuture,
+ LinkedHashMap<TopicIdPartition, Long> topicPartitionData
+ ) {
+ LinkedHashMap<TopicIdPartition, LogReadResult> readResponse;
+ try {
+ // Use getNow() to retrieve result without blocking (safe because
isDone() was true)
+ // getNow() returns the result immediately if future is complete
+ // or throws CompletionException if the async operation failed.
+ readResponse = completedFuture.get();
+ // Remove pending fetch as the request is completed now.
+ pendingFetch = null;
+ } catch (Exception e) {
+ // Force complete to send error response to client as the pending
future is not marked null
+ // yet hence force complete will try to retrieve the data again
which shall fail and complete
+ // the request.
+ return forceComplete();
+ }
+ // Process the successful read response - check for remote fetch,
minBytes, etc.
+ return processAsyncReadResponse(topicPartitionData, readResponse);
+ }
+
+ private boolean processAsyncReadResponse(
+ LinkedHashMap<TopicIdPartition, Long> topicPartitionData,
+ LinkedHashMap<TopicIdPartition, LogReadResult> readResponse
+ ) {
+ // Check if any partitions need remote storage fetch
+ // Store the remote fetch info for the topic partitions for which we
need to perform remote fetch.
+ // Remote fetch can only happen after local fetch completes
successfully.
+ LinkedHashMap<TopicIdPartition, LogReadResult>
remoteStorageFetchInfoMap =
+ maybePrepareRemoteStorageFetchInfo(topicPartitionData,
readResponse);
+
+ if (!remoteStorageFetchInfoMap.isEmpty()) {
+ // Some data is in remote storage - initiate remote fetch
+ // This will transition to remote fetch state
(pendingRemoteFetchesOpt)
+ return maybeProcessRemoteFetch(topicPartitionData,
remoteStorageFetchInfoMap);
+ }
+
+ // All data is fetched - update cached metadata for future requests
+ maybeUpdateFetchOffsetMetadata(topicPartitionData, readResponse);
+
+ // Check if we can complete the request. If any partition has a log
read error, we can complete
+ // the request with an error response.
+ if (anyPartitionHasLogReadError(readResponse) ||
isMinBytesSatisfied(topicPartitionData,
+
partitionMaxBytesStrategy.maxBytes(shareFetch.fetchParams().maxBytes,
+ topicPartitionData.keySet(), topicPartitionData.size()))) {
+ // Request can be completed - set state and force completion
+ partitionsAcquired = topicPartitionData;
+ localPartitionsAlreadyFetched = readResponse;
+ return forceComplete();
+ } else {
+ // minBytes not satisfied - release locks and return to purgatory
+ // Request will be retried when more data arrives or timeout occurs
+ 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());
+ return false;
+ }
+ }
+
+ private void
recordTopicPartitionsFetchRatioMetric(LinkedHashMap<TopicIdPartition, Long>
topicPartitionData) {
+ // Update metric to record acquired to requested partitions.
+ double requestTopicToAcquired = (double) topicPartitionData.size() /
shareFetch.topicIdPartitions().size();
+
shareGroupMetrics.recordTopicPartitionsFetchRatio(shareFetch.groupId(), (int)
(requestTopicToAcquired * 100));
+ }
+
+ private void
completeAsyncRequestWithEmptyResponse(LinkedHashMap<TopicIdPartition, Long>
topicPartitionData) {
+ try {
+ log.warn("Completing async fetch for group {}, member {},
partitions {} with empty reponse",
+ shareFetch.groupId(), shareFetch.memberId(),
topicPartitionData.keySet());
+ pendingFetch = null;
+ // Update fetch ration metric and complete the request with empty
response.
+ recordTopicPartitionsFetchRatioMetric(topicPartitionData);
+ shareFetch.maybeComplete(Map.of());
+ } finally {
+ // Release locks for all partitions acquired for this request
+
releasePartitionLocksAndAddToActionQueue(topicPartitionData.keySet(),
topicPartitionData.keySet());
Review Comment:
Make sense, done.
##########
core/src/main/java/kafka/server/share/DelayedShareFetch.java:
##########
@@ -234,9 +246,42 @@ public void onComplete() {
}
}
+ private void completeShareFetchAsyncRequest() {
+ // Check if async read is still not completed then rather waiting
further the request should
+ // be aborted. A warning log should be added to see if the fetches are
taking long.
+ if (!pendingFetch.isDone()) {
+ // Async read still in progress at timeout, cannot wait for it to
complete. So release
+ // locks and proceed without data. The new request will be created
on next poll from client.
+ completeAsyncRequestWithEmptyResponse(partitionsAcquired);
+ return;
+ }
+
+ // Async read completed just before/at timeout. We can include its
results in the response
+ // without blocking since future is done. Do not proceed for tiered
storage fetches
+ // as request has already timed out.
+ try {
+ localPartitionsAlreadyFetched = pendingFetch.get();
+ } catch (Exception e) {
+ log.error("Error getting async fetch result for group {}, member
{}",
+ shareFetch.groupId(), shareFetch.memberId(), e);
+ recordTopicPartitionsFetchRatioMetric(partitionsAcquired);
+ handleFetchException(shareFetch, partitionsAcquired.keySet(), e);
+
releasePartitionLocksAndAddToActionQueue(partitionsAcquired.keySet(),
partitionsAcquired.keySet());
Review Comment:
Done.
##########
core/src/main/java/kafka/server/share/DelayedShareFetch.java:
##########
@@ -234,9 +246,42 @@ public void onComplete() {
}
}
+ private void completeShareFetchAsyncRequest() {
+ // Check if async read is still not completed then rather waiting
further the request should
+ // be aborted. A warning log should be added to see if the fetches are
taking long.
+ if (!pendingFetch.isDone()) {
+ // Async read still in progress at timeout, cannot wait for it to
complete. So release
+ // locks and proceed without data. The new request will be created
on next poll from client.
+ completeAsyncRequestWithEmptyResponse(partitionsAcquired);
+ return;
+ }
+
+ // Async read completed just before/at timeout. We can include its
results in the response
+ // without blocking since future is done. Do not proceed for tiered
storage fetches
+ // as request has already timed out.
+ try {
+ localPartitionsAlreadyFetched = pendingFetch.get();
+ } catch (Exception e) {
+ log.error("Error getting async fetch result for group {}, member
{}",
+ shareFetch.groupId(), shareFetch.memberId(), e);
+ recordTopicPartitionsFetchRatioMetric(partitionsAcquired);
+ handleFetchException(shareFetch, partitionsAcquired.keySet(), e);
+
releasePartitionLocksAndAddToActionQueue(partitionsAcquired.keySet(),
partitionsAcquired.keySet());
+ return;
+ } finally {
+ // Clear the pending future regardless of completion status
+ pendingFetch = null;
+ }
+ // If the code reaches here then that does mean the async pending
fetch is completed successfully
+ // and the data is retrieved in localPartitionsAlreadyFetched. Hence,
we can proceed to complete
+ // the request with the data. Skip the local log fetch re-try as the
pending async fetch was
+ // not null hence a call fetch data has already procesed.
Review Comment:
Done.
--
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]
