adixitconfluent commented on code in PR #22780:
URL: https://github.com/apache/kafka/pull/22780#discussion_r3543080704
##########
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",
Review Comment:
nit: `response` typo
##########
core/src/main/java/kafka/server/share/DelayedShareFetch.java:
##########
@@ -884,25 +1089,43 @@ private void completeRemoteStorageShareFetchRequest() {
if (!acquiredNonRemoteFetchTopicPartitionData.isEmpty()) {
log.trace("Fetchable local share partitions for a remote
share fetch request data: {} with groupId: {} fetch params: {}",
acquiredNonRemoteFetchTopicPartitionData,
shareFetch.groupId(), shareFetch.fetchParams());
-
- LinkedHashMap<TopicIdPartition, LogReadResult>
responseData = readFromLog(
+ LinkedHashMap<TopicIdPartition, Long> partititionsToFetch
= acquiredNonRemoteFetchTopicPartitionData;
Review Comment:
the variable name should be partitionsToFetch not partititionsToFetch
--
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]
