adixitconfluent commented on code in PR #22780:
URL: https://github.com/apache/kafka/pull/22780#discussion_r3542167520
##########
core/src/main/java/kafka/server/share/DelayedShareFetch.java:
##########
@@ -572,24 +688,108 @@ private void updateAcquireElapsedTimeMetric() {
acquireStartTimeMs = currentTimeMs;
}
+ private boolean maybeCompleteAsyncFetch(
Review Comment:
since we are not doing completedFuture.isDone() check in the function
`maybeCompleteAsyncFetch`, we should add a comment here that the caller of this
function `maybeCompleteAsyncFetch` should perform this check.
##########
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:
nit: `processed` instead of `procesed`
##########
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.
Review Comment:
nit: ratio instead of ration
##########
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:
the second argument to `releasePartitionLocksAndAddToActionQueue` is for
topic partitions which return data, since in this case the `pendingFetch` did
not complete, I think we should pass an empty `Set`
##########
core/src/main/java/kafka/server/share/DelayedShareFetch.java:
##########
@@ -267,20 +311,63 @@ private void completeLocalLogShareFetchRequest() {
}
private void
processAcquiredTopicPartitionsForLocalLogFetch(LinkedHashMap<TopicIdPartition,
Long> topicPartitionData) {
- List<ShareFetchPartitionData> shareFetchPartitionDataList = new
ArrayList<>();
+ // If localPartitionsAlreadyFetched is empty hence the async fetch was
never attempted hence retry
+ // else try to complete the request.
+ CompletableFuture<LinkedHashMap<TopicIdPartition, LogReadResult>>
readFuture = null;
Review Comment:
nit: unneeded assignment to `readFuture`
##########
core/src/test/java/kafka/server/share/DelayedShareFetchTest.java:
##########
@@ -2437,13 +2506,28 @@ DelayedShareFetchBuilder withShareFetchData(ShareFetch
shareFetch) {
return this;
}
- DelayedShareFetchBuilder withReplicaManager(ReplicaManager
replicaManager) {
+ DelayedShareFetchBuilder withReplicaManagerLogReader(ReplicaManager
replicaManager) {
this.replicaManager = replicaManager;
this.logReader = new ReplicaManagerLogReader(replicaManager);
this.metadataProvider = new
ReplicaManagerPartitionMetadataProvider(replicaManager);
return this;
}
+ DelayedShareFetchBuilder withReplicaManager(ReplicaManager
replicaManager) {
+ this.replicaManager = replicaManager;
+ return this;
+ }
Review Comment:
nit: This function is not used in the code
##########
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:
similar comment - the second argument to
`releasePartitionLocksAndAddToActionQueue` is for topic partitions which
return data, since in this case the `pendingFetch` did not complete, I think we
should pass an empty `Set`
##########
core/src/main/java/kafka/server/share/DelayedShareFetch.java:
##########
@@ -912,19 +1135,56 @@ private void completeRemoteStorageShareFetchRequest() {
shareFetch, shareFetchPartitionDataList, sharePartitions,
metadataProvider, exceptionHandler);
shareFetch.maybeComplete(remoteFetchResponse);
log.trace("Remote share fetch request completed successfully,
response: {}", remoteFetchResponse);
- } catch (InterruptedException | ExecutionException e) {
- log.error("Exception occurred in completing remote fetch {} for
delayed share fetch request {}", pendingRemoteFetchesOpt.get(), e);
-
handleExceptionInCompletingRemoteStorageShareFetchRequest(acquiredNonRemoteFetchTopicPartitionData.keySet(),
e);
} catch (Exception e) {
log.error("Unexpected error in processing delayed share fetch
request", e);
handleExceptionInCompletingRemoteStorageShareFetchRequest(acquiredNonRemoteFetchTopicPartitionData.keySet(),
e);
} finally {
+ // TODO: Check maybe to skip
acquiredNonRemoteFetchTopicPartitionData from releasing locks
Review Comment:
from the look of the code (as mentioned in the TODO), acquired non remote
fetch topic partitions should be released via
`readAndProcessFetchResultCompletion` -> `processFetchResultAndComplete` ->
`releasePartitionLocksAndAddToActionQueue`. Why are we adding them here then?
##########
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)
Review Comment:
`get()` instead of now `getNow()`, I think
##########
core/src/main/java/kafka/server/share/DelayedShareFetch.java:
##########
@@ -912,19 +1135,56 @@ private void completeRemoteStorageShareFetchRequest() {
shareFetch, shareFetchPartitionDataList, sharePartitions,
metadataProvider, exceptionHandler);
shareFetch.maybeComplete(remoteFetchResponse);
log.trace("Remote share fetch request completed successfully,
response: {}", remoteFetchResponse);
- } catch (InterruptedException | ExecutionException e) {
- log.error("Exception occurred in completing remote fetch {} for
delayed share fetch request {}", pendingRemoteFetchesOpt.get(), e);
-
handleExceptionInCompletingRemoteStorageShareFetchRequest(acquiredNonRemoteFetchTopicPartitionData.keySet(),
e);
} catch (Exception e) {
log.error("Unexpected error in processing delayed share fetch
request", e);
handleExceptionInCompletingRemoteStorageShareFetchRequest(acquiredNonRemoteFetchTopicPartitionData.keySet(),
e);
} finally {
+ // TODO: Check maybe to skip
acquiredNonRemoteFetchTopicPartitionData from releasing locks
+ // as that might also be tried to release in
processFetchResultAndComplete. Avoid it in current
+ // implementation and validate later with tests. It won't have any
downside as SharePartition
+ // has the right protection against illegal locks release.
Set<TopicIdPartition> topicIdPartitions = new
LinkedHashSet<>(partitionsAcquired.keySet());
topicIdPartitions.addAll(acquiredNonRemoteFetchTopicPartitionData.keySet());
releasePartitionLocksAndAddToActionQueue(topicIdPartitions,
partitionsWithData(shareFetchPartitionDataList));
}
}
+ private void readAndProcessFetchResultCompletion(
+ LinkedHashMap<TopicIdPartition, Long> partitionsToFetch,
+ int readBytes,
+ Consumer<List<ShareFetchPartitionData>> completionConsumer
+ ) {
+ CompletableFuture<LinkedHashMap<TopicIdPartition, LogReadResult>>
readFuture = null;
Review Comment:
nit: not needed assignment
--
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]
