This is an automated email from the ASF dual-hosted git repository.
apoorvmittal10 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new f2b857133fa KAFKA-20783: Integrate async log reader fetch in
DelayedShareFetch (1/N) (#22780)
f2b857133fa is described below
commit f2b857133fa526174eaaadbc90d73a0687beb104
Author: Apoorv Mittal <[email protected]>
AuthorDate: Thu Jul 9 11:37:01 2026 +0100
KAFKA-20783: Integrate async log reader fetch in DelayedShareFetch (1/N)
(#22780)
The PR integrates async processing in DelayedShareFetch which ll help
move all the remote fetches out of DelayedShareFetch in future. The
DelayedShareFetch is too complex today and shall be simplified by moving
the code to read data out of DelayedShareFetch.
In subsequent PR more tests will be added for this change.
Tested with perf-test:
```
bin/kafka-share-consumer-perf-test.sh --bootstrap-server localhost:9092
--topic test-topic-1 --messages 100000000 --timeout 300000 --group
share-group-1 --threads 10
2026-07-07 20:22:00:959, 2026-07-07 20:22:01:960, 83.4237, 2.6258,
305928.0719, 9719560, 1001 for share consumer 1
2026-07-07 20:22:00:960, 2026-07-07 20:22:01:960, 83.2087, 2.6367,
307197.0000, 9694514, 1000 for share consumer 2
2026-07-07 20:22:00:961, 2026-07-07 20:22:01:961, 83.2831, 2.6202,
305271.0000, 9703189, 1000 for share consumer 4
2026-07-07 20:22:00:959, 2026-07-07 20:22:01:962, 83.4402, 2.6370,
307238.2851, 9721486, 1003 for share consumer 8
2026-07-07 20:22:00:961, 2026-07-07 20:22:01:962, 83.4071, 2.6423,
307852.1479, 9717634, 1001 for share consumer 6
2026-07-07 20:22:00:962, 2026-07-07 20:22:01:963, 83.5146, 2.6341,
306890.1099, 9730153, 1001 for share consumer 7
2026-07-07 20:22:00:963, 2026-07-07 20:22:01:963, 83.4980, 2.6367,
307197.0000, 9728227, 1000 for share consumer 10
2026-07-07 20:22:00:962, 2026-07-07 20:22:01:963, 83.3409, 2.6423,
307852.1479, 9709920, 1001 for share consumer 5
2026-07-07 20:22:00:964, 2026-07-07 20:22:01:964, 83.3906, 2.6119,
304308.0000, 9715708, 1000 for share consumer 9
2026-07-07 20:21:12:236, 2026-07-07 20:22:02:329, 858.3300, 17.1347,
1996340.6264, 100002691, 50093
```
Reviewers: Andrew Schofield <[email protected]>, Abhinav Dixit
<[email protected]>
---
.../java/kafka/server/share/DelayedShareFetch.java | 443 +++++++++++++++++----
.../kafka/server/share/DelayedShareFetchTest.java | 185 +++++++--
.../server/share/SharePartitionManagerTest.java | 8 +-
3 files changed, 507 insertions(+), 129 deletions(-)
diff --git a/core/src/main/java/kafka/server/share/DelayedShareFetch.java
b/core/src/main/java/kafka/server/share/DelayedShareFetch.java
index 0ac16bea2e5..e60d3979b20 100644
--- a/core/src/main/java/kafka/server/share/DelayedShareFetch.java
+++ b/core/src/main/java/kafka/server/share/DelayedShareFetch.java
@@ -70,6 +70,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Lock;
import java.util.function.BiConsumer;
+import java.util.function.Consumer;
import static kafka.server.share.PendingRemoteFetches.RemoteFetch;
@@ -106,6 +107,12 @@ public class DelayedShareFetch extends DelayedOperation {
private LinkedHashMap<TopicIdPartition, Long> partitionsAcquired;
private LinkedHashMap<TopicIdPartition, LogReadResult>
localPartitionsAlreadyFetched;
private Optional<PendingRemoteFetches> pendingRemoteFetchesOpt;
+ /** Holds the CompletableFuture for a pending async fetch operation.
+ * Unlike remote fetches (which have one future per partition), async
fetch has a single future
+ * for all partitions. This is null when no async operation is pending.
Partition locks are held
+ * while this is non-null.
+ */
+ private CompletableFuture<LinkedHashMap<TopicIdPartition, LogReadResult>>
pendingFetch;
private Optional<Exception> remoteStorageFetchException;
private final AtomicBoolean outsidePurgatoryCallbackLock;
private final long remoteFetchMaxWaitMs;
@@ -219,7 +226,12 @@ public class DelayedShareFetch extends DelayedOperation {
+ "topic partitions {}", shareFetch.groupId(),
shareFetch.memberId(),
partitionsAcquired.keySet());
- if (remoteStorageFetchException.isPresent()) {
+ // When the request reaches timeout, there could be an in-flight async
read operation which
+ // may or may not have completed. We need to check if the async read
has completed to decide
+ // whether to use its results or release the locks and complete with
whatever data we have.
+ if (pendingFetch != null) {
+ completeShareFetchAsyncRequest();
+ } else if (remoteStorageFetchException.isPresent()) {
completeErroneousRemoteShareFetchRequest();
} else if (pendingRemoteFetchesOpt.isPresent()) {
if (maybeRegisterCallbackPendingRemoteFetch()) {
@@ -234,9 +246,44 @@ public class DelayedShareFetch extends DelayedOperation {
}
}
+ private void completeShareFetchAsyncRequest() {
+ // If the async read is still not completed at timeout, rather than
waiting further the request
+ // should be completed without data.
+ 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);
+ // No data could be fetched as the async read failed, hence pass
an empty set of partitions
+ // with data to the action queue.
+
releasePartitionLocksAndAddToActionQueue(partitionsAcquired.keySet(), Set.of());
+ 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 to fetch data has already been processed.
+ completeLocalLogShareFetchRequest();
+ }
+
private void completeLocalLogShareFetchRequest() {
LinkedHashMap<TopicIdPartition, Long> topicPartitionData;
- // tryComplete did not invoke forceComplete, so we need to check if we
have any partitions to fetch.
+ // If tryComplete did not invoke forceComplete, so we need to check if
we have any partitions to fetch.
if (partitionsAcquired.isEmpty()) {
topicPartitionData = acquirablePartitions(sharePartitions);
// The TopicPartitionsAcquireTimeMs metric signifies the tension
when acquiring the locks
@@ -246,7 +293,8 @@ public class DelayedShareFetch extends DelayedOperation {
// for the metric.
updateAcquireElapsedTimeMetric();
} else {
- // tryComplete invoked forceComplete, so we can use the data from
tryComplete.
+ // tryComplete invoked forceComplete, so we can use the data from
tryComplete. Or the async
+ // fetch future was not completed, in either case some partitions
should still be acquired.
topicPartitionData = partitionsAcquired;
}
@@ -256,9 +304,7 @@ public class DelayedShareFetch extends DelayedOperation {
shareFetch.maybeComplete(Map.of());
return;
} else {
- // Update metric to record acquired to requested partitions.
- double requestTopicToAcquired = (double) topicPartitionData.size()
/ shareFetch.topicIdPartitions().size();
-
shareGroupMetrics.recordTopicPartitionsFetchRatio(shareFetch.groupId(), (int)
(requestTopicToAcquired * 100));
+ recordTopicPartitionsFetchRatioMetric(topicPartitionData);
}
log.trace("Fetchable share partitions data: {} with groupId: {} fetch
params: {}",
topicPartitionData, shareFetch.groupId(),
shareFetch.fetchParams());
@@ -267,20 +313,65 @@ public class DelayedShareFetch extends DelayedOperation {
}
private void
processAcquiredTopicPartitionsForLocalLogFetch(LinkedHashMap<TopicIdPartition,
Long> topicPartitionData) {
- List<ShareFetchPartitionData> shareFetchPartitionDataList = new
ArrayList<>();
+ // If localPartitionsAlreadyFetched is empty then the async fetch was
never attempted, so read
+ // from the log now. Otherwise, use the already fetched data and only
read the remaining partitions.
+ CompletableFuture<LinkedHashMap<TopicIdPartition, LogReadResult>>
readFuture;
try {
- LinkedHashMap<TopicIdPartition, LogReadResult> responseData;
- if (localPartitionsAlreadyFetched.isEmpty())
- responseData = readFromLog(
+ if (localPartitionsAlreadyFetched.isEmpty()) {
+ readFuture = readFromLog(
topicPartitionData,
-
partitionMaxBytesStrategy.maxBytes(shareFetch.fetchParams().maxBytes,
topicPartitionData.keySet(), topicPartitionData.size()));
- else
- // There shouldn't be a case when we have a
partitionsAlreadyFetched value here and this variable is getting
- // updated in a different tryComplete thread.
- responseData = combineLogReadResponse(topicPartitionData,
localPartitionsAlreadyFetched);
+
partitionMaxBytesStrategy.maxBytes(shareFetch.fetchParams().maxBytes,
+ topicPartitionData.keySet(), topicPartitionData.size())
+ );
+ } else {
+ // Combine already-fetched data with new partitions that need
fetching. Maybe only
+ // use already fetched data from async fetch and never attempt
for another retry as
+ // data should have already been fetched for the partitions
which were acquired in
+ // tryComplete.
+ readFuture = combineLogReadResponse(topicPartitionData,
localPartitionsAlreadyFetched);
+ }
+ } catch (Exception e) {
+ // Handle synchronous exceptions thrown by async read before
future is returned
+ log.error("Error initiating async fetch during request completion
for group {}, member {}",
+ shareFetch.groupId(), shareFetch.memberId(), e);
+ handleFetchException(shareFetch, topicPartitionData.keySet(), e);
+
releasePartitionLocksAndAddToActionQueue(topicPartitionData.keySet(), new
HashSet<>());
+ return;
+ }
+ // If the future is successfully created then complete the request
asynchronously. The completion
+ // handler is registered from on-complete hence no 2 threads can run
concurrently.
+ readFuture.whenComplete((result, throwable) -> {
+ if (throwable != null) {
+ log.error("Async fetch failed during request completion for
group {}, member {}",
+ shareFetch.groupId(), shareFetch.memberId(), throwable);
+ handleFetchException(shareFetch, topicPartitionData.keySet(),
throwable);
+
releasePartitionLocksAndAddToActionQueue(topicPartitionData.keySet(), new
HashSet<>());
+ return;
+ }
+ // Complete the share fetch request and release the locks as
completion handler will complete
+ // the request.
+ processFetchResultAndComplete(result, topicPartitionData,
shareFetchPartitionDataList ->
+ shareFetch.maybeComplete(ShareFetchUtils.processFetchResponse(
+ shareFetch,
+ shareFetchPartitionDataList,
+ sharePartitions,
+ metadataProvider,
+ exceptionHandler
+ )));
+ });
+ }
+ private void processFetchResultAndComplete(
+ LinkedHashMap<TopicIdPartition, LogReadResult> responseData,
+ LinkedHashMap<TopicIdPartition, Long> topicPartitionData,
+ Consumer<List<ShareFetchPartitionData>> completionConsumer
+ ) {
+ List<ShareFetchPartitionData> shareFetchPartitionDataList = new
ArrayList<>();
+ try {
+ // Reset fetch offset metadata for partitions requiring remote
fetch
resetFetchOffsetMetadataForRemoteFetchPartitions(topicPartitionData,
responseData);
+ // Build response data excluding partitions needing remote fetch
responseData.forEach((topicIdPartition, logReadResult) -> {
if (logReadResult.info().delayedRemoteStorageFetch.isEmpty()) {
shareFetchPartitionDataList.add(new
ShareFetchPartitionData(
@@ -290,19 +381,17 @@ public class DelayedShareFetch extends DelayedOperation {
));
}
});
-
- shareFetch.maybeComplete(ShareFetchUtils.processFetchResponse(
- shareFetch,
- shareFetchPartitionDataList,
- sharePartitions,
- metadataProvider,
- exceptionHandler
- ));
+ completionConsumer.accept(shareFetchPartitionDataList);
} catch (Exception e) {
- log.error("Error processing delayed share fetch request", e);
+ log.error("Error processing delayed share fetch request for group
{}, member {}",
+ shareFetch.groupId(), shareFetch.memberId(), e);
handleFetchException(shareFetch, topicPartitionData.keySet(), e);
} finally {
-
releasePartitionLocksAndAddToActionQueue(topicPartitionData.keySet(),
partitionsWithData(shareFetchPartitionDataList));
+ // Always release locks and update action queue.
+ releasePartitionLocksAndAddToActionQueue(
+ topicPartitionData.keySet(),
+ partitionsWithData(shareFetchPartitionDataList)
+ );
}
}
@@ -332,43 +421,74 @@ public class DelayedShareFetch extends DelayedOperation {
});
}
- /**
- * 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).
Hence, rather than processing the result
+ // directly here, trigger a purgatory check so that
tryComplete() is invoked again for
+ // this operation under the operation lock (via
safeTryComplete). Processing the async
+ // result inside tryComplete() ensures the shared state
mutations (pendingFetch,
+ // partitionsAcquired, localPartitionsAlreadyFetched)
happen under the lock and avoids a
+ // race with a concurrent tryComplete() invoked by a
purgatory watcher thread.
+ // Also, the operation is watched on the group key for
each topic partition in the request,
+ // hence triggering a check on any one of them will
re-invoke tryComplete()
+ // for this operation. Just take the first in the list.
+ TopicIdPartition topicIdPartition =
topicPartitionData.keySet().iterator().next();
+ 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
+ // when tryComplete() re-processes the completed
future.
+ }
+ replicaManager.completeDelayedShareFetchRequest(
+ new
DelayedShareFetchGroupKey(shareFetch.groupId(), topicIdPartition));
+ });
+ // Return false to keep request in purgatory - locks
remain held
+ return false;
}
} else {
+ // Could not acquire locks for any partitions (all locked by
other requests or at capacity)
log.trace("Can't acquire any partitions in the share fetch
request for group {}, member {}, " +
"topic partitions {}", shareFetch.groupId(),
shareFetch.memberId(),
sharePartitions.keySet());
@@ -388,7 +508,8 @@ public class DelayedShareFetch extends DelayedOperation {
// local log read. We do not release locks for partitions which
have a remote storage read because we need to
// complete the share fetch request in onComplete and if we
release the locks early here, some other DelayedShareFetch
// request might get the locks for those partitions without this
one getting complete.
- if (remoteStorageFetchException.isEmpty()) {
+ // Similarly, if we have a pending async fetch, don't release
locks yet.
+ if (remoteStorageFetchException.isEmpty() && pendingFetch == null)
{
releasePartitionLocks(topicPartitionData.keySet());
partitionsAcquired.clear();
localPartitionsAlreadyFetched.clear();
@@ -432,7 +553,9 @@ public class DelayedShareFetch extends DelayedOperation {
return topicPartitionData;
}
- private LinkedHashMap<TopicIdPartition, LogReadResult>
maybeReadFromLog(LinkedHashMap<TopicIdPartition, Long> topicPartitionData) {
+ private CompletableFuture<LinkedHashMap<TopicIdPartition, LogReadResult>>
maybeReadFromLog(
+ LinkedHashMap<TopicIdPartition, Long> topicPartitionData
+ ) {
LinkedHashMap<TopicIdPartition, Long>
partitionsNotMatchingFetchOffsetMetadata = new LinkedHashMap<>();
topicPartitionData.forEach((topicIdPartition, fetchOffset) -> {
SharePartition sharePartition =
sharePartitions.get(topicIdPartition);
@@ -441,12 +564,15 @@ public class DelayedShareFetch extends DelayedOperation {
}
});
if (partitionsNotMatchingFetchOffsetMetadata.isEmpty()) {
- return new LinkedHashMap<>();
+ return CompletableFuture.completedFuture(new LinkedHashMap<>());
}
- // We fetch data from replica manager corresponding to the topic
partitions that have missing fetch offset metadata.
+
+ // We fetch data from log reader corresponding to the topic partitions
that have missing fetch offset metadata.
// Although we are fetching partition max bytes for
partitionsNotMatchingFetchOffsetMetadata,
// we will take acquired partitions size = topicPartitionData.size()
because we do not want to let the
// leftover partitions to starve which will be fetched later.
+ // readRemote=false ensures to skip remote reads as they are dealt
separately outside the log
+ // reader, currently.
return readFromLog(
partitionsNotMatchingFetchOffsetMetadata,
partitionMaxBytesStrategy.maxBytes(shareFetch.fetchParams().maxBytes,
partitionsNotMatchingFetchOffsetMetadata.keySet(), topicPartitionData.size()));
@@ -523,11 +649,11 @@ public class DelayedShareFetch extends DelayedOperation {
return metadataProvider.endOffsetMetadata(topicIdPartition,
isolationType);
}
- private LinkedHashMap<TopicIdPartition, LogReadResult>
readFromLog(LinkedHashMap<TopicIdPartition, Long> topicPartitionFetchOffsets,
-
LinkedHashMap<TopicIdPartition, Integer> partitionMaxBytes) {
+ private CompletableFuture<LinkedHashMap<TopicIdPartition, LogReadResult>>
readFromLog(LinkedHashMap<TopicIdPartition, Long> topicPartitionFetchOffsets,
+
LinkedHashMap<TopicIdPartition, Integer> partitionMaxBytes) {
// Filter if there already exists any erroneous topic partition.
Set<TopicIdPartition> partitionsToFetch =
shareFetch.filterErroneousTopicPartitions(topicPartitionFetchOffsets.keySet());
- return logReader.read(shareFetch.fetchParams(), partitionsToFetch,
topicPartitionFetchOffsets, partitionMaxBytes);
+ return logReader.readAsync(shareFetch.fetchParams(),
partitionsToFetch, topicPartitionFetchOffsets, partitionMaxBytes, false);
}
private boolean
anyPartitionHasLogReadError(LinkedHashMap<TopicIdPartition, LogReadResult>
readResponse) {
@@ -572,9 +698,96 @@ public class DelayedShareFetch extends DelayedOperation {
acquireStartTimeMs = currentTimeMs;
}
+ /**
+ * Process the result of a completed async read future. The caller must
ensure the future is
+ * already completed (i.e. {@code completedFuture.isDone()} is true)
before invoking this method,
+ * as {@link CompletableFuture#get()} is called without any additional
wait.
+ */
+ private boolean maybeCompleteAsyncFetch(
+ CompletableFuture<LinkedHashMap<TopicIdPartition, LogReadResult>>
completedFuture,
+ LinkedHashMap<TopicIdPartition, Long> topicPartitionData
+ ) {
+ LinkedHashMap<TopicIdPartition, LogReadResult> readResponse;
+ try {
+ // Retrieve the result without blocking (safe because the caller
ensured isDone() was true).
+ // get() returns the result immediately since the future is
complete, or throws
+ // ExecutionException 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 response",
+ shareFetch.groupId(), shareFetch.memberId(),
topicPartitionData.keySet());
+ pendingFetch = null;
+ // Update fetch ratio metric and complete the request with empty
response.
+ recordTopicPartitionsFetchRatioMetric(topicPartitionData);
+ shareFetch.maybeComplete(Map.of());
+ } finally {
+ // Release locks for all partitions acquired for this request. The
async fetch did not
+ // complete, so no partition has data, hence pass an empty set as
the partitions with data.
+
releasePartitionLocksAndAddToActionQueue(topicPartitionData.keySet(), Set.of());
+ }
+ }
+
// Visible for testing.
- LinkedHashMap<TopicIdPartition, LogReadResult>
combineLogReadResponse(LinkedHashMap<TopicIdPartition, Long> topicPartitionData,
-
LinkedHashMap<TopicIdPartition, LogReadResult> existingFetchedData) {
+ CompletableFuture<LinkedHashMap<TopicIdPartition, LogReadResult>>
combineLogReadResponse(LinkedHashMap<TopicIdPartition, Long> topicPartitionData,
+
LinkedHashMap<TopicIdPartition, LogReadResult>
existingFetchedData) {
LinkedHashMap<TopicIdPartition, Long> missingLogReadTopicPartitions =
new LinkedHashMap<>();
topicPartitionData.forEach((topicIdPartition, fetchOffset) -> {
if (!existingFetchedData.containsKey(topicIdPartition)) {
@@ -582,14 +795,17 @@ public class DelayedShareFetch extends DelayedOperation {
}
});
if (missingLogReadTopicPartitions.isEmpty()) {
- return existingFetchedData;
+ return CompletableFuture.completedFuture(existingFetchedData);
}
- LinkedHashMap<TopicIdPartition, LogReadResult>
missingTopicPartitionsLogReadResponse = readFromLog(
+ return readFromLog(
missingLogReadTopicPartitions,
-
partitionMaxBytesStrategy.maxBytes(shareFetch.fetchParams().maxBytes,
missingLogReadTopicPartitions.keySet(), topicPartitionData.size()));
- missingTopicPartitionsLogReadResponse.putAll(existingFetchedData);
- return missingTopicPartitionsLogReadResponse;
+
partitionMaxBytesStrategy.maxBytes(shareFetch.fetchParams().maxBytes,
+ missingLogReadTopicPartitions.keySet(),
topicPartitionData.size())
+ ).thenApply(missingTopicPartitionsLogReadResponse -> {
+ missingTopicPartitionsLogReadResponse.putAll(existingFetchedData);
+ return missingTopicPartitionsLogReadResponse;
+ });
}
// Visible for testing.
@@ -626,6 +842,11 @@ public class DelayedShareFetch extends DelayedOperation {
return expiredRequestMeter;
}
+ // Visible for testing.
+ CompletableFuture<LinkedHashMap<TopicIdPartition, LogReadResult>>
pendingFetch() {
+ return pendingFetch;
+ }
+
private LinkedHashMap<TopicIdPartition, LogReadResult>
maybePrepareRemoteStorageFetchInfo(
LinkedHashMap<TopicIdPartition, Long> topicPartitionData,
LinkedHashMap<TopicIdPartition, LogReadResult> readResponse
@@ -871,7 +1092,7 @@ public class DelayedShareFetch extends DelayedOperation {
}
}
- // If remote fetch bytes < shareFetch.fetchParams().maxBytes,
then we will try for a local read.
+ // If remote fetch bytes < shareFetch.fetchParams().maxBytes, then
we will try for a local read.
if (readableBytes < shareFetch.fetchParams().maxBytes) {
// Get the local log read based topic partitions.
LinkedHashMap<TopicIdPartition, SharePartition>
nonRemoteFetchSharePartitions = new LinkedHashMap<>();
@@ -884,25 +1105,43 @@ public class DelayedShareFetch extends DelayedOperation {
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> partitionsToFetch =
acquiredNonRemoteFetchTopicPartitionData;
+ // Register for further read and request completion.
+ readAndProcessFetchResultCompletion(
acquiredNonRemoteFetchTopicPartitionData,
-
partitionMaxBytesStrategy.maxBytes(shareFetch.fetchParams().maxBytes -
readableBytes, acquiredNonRemoteFetchTopicPartitionData.keySet(),
acquiredNonRemoteFetchTopicPartitionData.size()));
-
resetFetchOffsetMetadataForRemoteFetchPartitions(acquiredNonRemoteFetchTopicPartitionData,
responseData);
- for (Map.Entry<TopicIdPartition, LogReadResult> entry :
responseData.entrySet()) {
- if
(entry.getValue().info().delayedRemoteStorageFetch.isEmpty()) {
- shareFetchPartitionDataList.add(
- new ShareFetchPartitionData(
- entry.getKey(),
-
acquiredNonRemoteFetchTopicPartitionData.get(entry.getKey()),
-
entry.getValue().toFetchPartitionData(false)
- )
- );
- }
- }
+ readableBytes,
+ addedShareFetchPartitionDataList -> {
+
shareFetchPartitionDataList.addAll(addedShareFetchPartitionDataList);
+
completeRemoteFetchRequest(shareFetchPartitionDataList, partitionsToFetch);
+ });
+ return;
}
}
+ // Complete with remote data (and any local data if no additional
async read was needed)
+ completeRemoteFetchRequest(shareFetchPartitionDataList,
acquiredNonRemoteFetchTopicPartitionData);
+ } catch (InterruptedException | ExecutionException e) {
+ log.error("Exception occurred in completing remote fetch {} for
delayed share fetch request {}", pendingRemoteFetchesOpt.get(), e);
+
handleExceptionInCompletingRemoteStorageShareFetchRequest(acquiredNonRemoteFetchTopicPartitionData.keySet(),
e);
+ // Release locks on error
+ Set<TopicIdPartition> topicIdPartitions = new
LinkedHashSet<>(partitionsAcquired.keySet());
+
topicIdPartitions.addAll(acquiredNonRemoteFetchTopicPartitionData.keySet());
+ releasePartitionLocksAndAddToActionQueue(topicIdPartitions, new
HashSet<>());
+ } catch (Exception e) {
+ log.error("Unexpected error in processing delayed share fetch
request", e);
+
handleExceptionInCompletingRemoteStorageShareFetchRequest(acquiredNonRemoteFetchTopicPartitionData.keySet(),
e);
+ // Release locks on error
+ Set<TopicIdPartition> topicIdPartitions = new
LinkedHashSet<>(partitionsAcquired.keySet());
+
topicIdPartitions.addAll(acquiredNonRemoteFetchTopicPartitionData.keySet());
+ releasePartitionLocksAndAddToActionQueue(topicIdPartitions, new
HashSet<>());
+ }
+ }
+
+ private void completeRemoteFetchRequest(
+ List<ShareFetchPartitionData> shareFetchPartitionDataList,
+ LinkedHashMap<TopicIdPartition, Long>
acquiredNonRemoteFetchTopicPartitionData
+ ) {
+ try {
// Update metric to record acquired to requested partitions.
double acquiredRatio = (double) (partitionsAcquired.size() +
acquiredNonRemoteFetchTopicPartitionData.size()) /
shareFetch.topicIdPartitions().size();
if (acquiredRatio > 0)
@@ -912,19 +1151,57 @@ public class DelayedShareFetch extends DelayedOperation {
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 {
- Set<TopicIdPartition> topicIdPartitions = new
LinkedHashSet<>(partitionsAcquired.keySet());
-
topicIdPartitions.addAll(acquiredNonRemoteFetchTopicPartitionData.keySet());
- releasePartitionLocksAndAddToActionQueue(topicIdPartitions,
partitionsWithData(shareFetchPartitionDataList));
+ // The locks for acquiredNonRemoteFetchTopicPartitionData are
released by
+ // processFetchResultAndComplete (invoked via
readAndProcessFetchResultCompletion). As the
+ // locks are release in finally block in
readAndProcessFetchResultCompletion hence share
+ // fetch will be completed in the current try block prior locks
getting released. Hence,
+ // need to release the locks for the remote fetch partitions here.
+
releasePartitionLocksAndAddToActionQueue(partitionsAcquired.keySet(),
partitionsWithData(shareFetchPartitionDataList));
}
}
+ private void readAndProcessFetchResultCompletion(
+ LinkedHashMap<TopicIdPartition, Long> partitionsToFetch,
+ int readBytes,
+ Consumer<List<ShareFetchPartitionData>> completionConsumer
+ ) {
+ CompletableFuture<LinkedHashMap<TopicIdPartition, LogReadResult>>
readFuture;
+ // Start async read for additional local partitions
+ try {
+ readFuture = readFromLog(
+ partitionsToFetch,
+
partitionMaxBytesStrategy.maxBytes(shareFetch.fetchParams().maxBytes -
readBytes,
+ partitionsToFetch.keySet(),
+ partitionsToFetch.size())
+ );
+ } catch (Exception e) {
+ // Handle synchronous exception from readFromLog()
+ log.error("Error initiating async fetch in remote completion for
group {}, member {}",
+ shareFetch.groupId(), shareFetch.memberId(), e);
+ // Continue to complete with just remote data.
+ completionConsumer.accept(List.of());
+ return;
+ }
+
+ // Handle async read completion. This async completion is registered
only from on-complete
+ // hence multiple threads cannot execute at same time. It's safe to
register further processing
+ // in completion handler.
+ readFuture.whenComplete((responseData, throwable) -> {
+ if (throwable != null) {
+ log.error("Async fetch failed in remote completion for group
{}, member {}",
+ shareFetch.groupId(), shareFetch.memberId(), throwable);
+ // Continue to complete with just remote data.
+ completionConsumer.accept(List.of());
+ return;
+ }
+ processFetchResultAndComplete(responseData, partitionsToFetch,
completionConsumer);
+ });
+ }
+
private void handleExceptionInCompletingRemoteStorageShareFetchRequest(
Set<TopicIdPartition> acquiredNonRemoteFetchTopicPartitions,
Exception e
diff --git a/core/src/test/java/kafka/server/share/DelayedShareFetchTest.java
b/core/src/test/java/kafka/server/share/DelayedShareFetchTest.java
index ba50773b4ac..e04274f0751 100644
--- a/core/src/test/java/kafka/server/share/DelayedShareFetchTest.java
+++ b/core/src/test/java/kafka/server/share/DelayedShareFetchTest.java
@@ -174,7 +174,7 @@ public class DelayedShareFetchTest {
.withSharePartitions(sharePartitions)
.withShareGroupMetrics(shareGroupMetrics)
.withFetchId(fetchId)
- .withReplicaManager(replicaManager)
+ .withReplicaManagerLogReader(replicaManager)
.build());
when(sp0.maybeAcquireFetchLock(fetchId)).thenReturn(true);
@@ -247,7 +247,7 @@ public class DelayedShareFetchTest {
DelayedShareFetch delayedShareFetch =
spy(DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
.withSharePartitions(sharePartitions)
- .withReplicaManager(replicaManager)
+ .withReplicaManagerLogReader(replicaManager)
.withExceptionHandler(exceptionHandler)
.withPartitionMaxBytesStrategy(partitionMaxBytesStrategy)
.withShareGroupMetrics(shareGroupMetrics)
@@ -320,7 +320,7 @@ public class DelayedShareFetchTest {
DelayedShareFetch delayedShareFetch =
spy(DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
.withSharePartitions(sharePartitions)
- .withReplicaManager(replicaManager)
+ .withReplicaManagerLogReader(replicaManager)
.withExceptionHandler(exceptionHandler)
.withFetchId(fetchId)
.build());
@@ -375,7 +375,7 @@ public class DelayedShareFetchTest {
DelayedShareFetch delayedShareFetch =
spy(DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
.withSharePartitions(sharePartitions)
- .withReplicaManager(replicaManager)
+ .withReplicaManagerLogReader(replicaManager)
.withPartitionMaxBytesStrategy(partitionMaxBytesStrategy)
.withShareGroupMetrics(shareGroupMetrics)
.withTime(time)
@@ -435,7 +435,7 @@ public class DelayedShareFetchTest {
Uuid fetchId = Uuid.randomUuid();
DelayedShareFetch delayedShareFetch =
spy(DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
- .withReplicaManager(replicaManager)
+ .withReplicaManagerLogReader(replicaManager)
.withSharePartitions(sharePartitions)
.withShareGroupMetrics(shareGroupMetrics)
.withTime(time)
@@ -496,7 +496,7 @@ public class DelayedShareFetchTest {
BiConsumer<SharePartitionKey, Throwable> exceptionHandler =
mockExceptionHandler();
DelayedShareFetch delayedShareFetch =
spy(DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
- .withReplicaManager(replicaManager)
+ .withReplicaManagerLogReader(replicaManager)
.withSharePartitions(sharePartitions)
.withPartitionMaxBytesStrategy(partitionMaxBytesStrategy)
.withShareGroupMetrics(shareGroupMetrics)
@@ -556,7 +556,7 @@ public class DelayedShareFetchTest {
Uuid fetchId = Uuid.randomUuid();
DelayedShareFetch delayedShareFetch =
spy(DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
- .withReplicaManager(replicaManager)
+ .withReplicaManagerLogReader(replicaManager)
.withSharePartitions(sharePartitions)
.withShareGroupMetrics(shareGroupMetrics)
.withFetchId(fetchId)
@@ -635,9 +635,9 @@ public class DelayedShareFetchTest {
when(replicaManager.getPartitionOrException(tp2.topicPartition())).thenReturn(p2);
Uuid fetchId1 = Uuid.randomUuid();
- DelayedShareFetch delayedShareFetch1 =
DelayedShareFetchTest.DelayedShareFetchBuilder.builder()
+ DelayedShareFetch delayedShareFetch1 =
DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch1)
- .withReplicaManager(replicaManager)
+ .withReplicaManagerLogReader(replicaManager)
.withSharePartitions(sharePartitions1)
.withFetchId(fetchId1)
.build();
@@ -673,7 +673,7 @@ public class DelayedShareFetchTest {
BiConsumer<SharePartitionKey, Throwable> exceptionHandler =
mockExceptionHandler();
DelayedShareFetch delayedShareFetch2 =
spy(DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch2)
- .withReplicaManager(replicaManager)
+ .withReplicaManagerLogReader(replicaManager)
.withSharePartitions(sharePartitions2)
.withPartitionMaxBytesStrategy(partitionMaxBytesStrategy)
.withFetchId(fetchId2)
@@ -736,9 +736,9 @@ public class DelayedShareFetchTest {
topicIdPartitions1.forEach(topicIdPartition ->
delayedShareFetchWatchKeys.add(new DelayedShareFetchGroupKey(groupId,
topicIdPartition.topicId(), topicIdPartition.partition())));
Uuid fetchId1 = Uuid.randomUuid();
- DelayedShareFetch delayedShareFetch1 =
DelayedShareFetchTest.DelayedShareFetchBuilder.builder()
+ DelayedShareFetch delayedShareFetch1 =
DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch1)
- .withReplicaManager(replicaManager)
+ .withReplicaManagerLogReader(replicaManager)
.withSharePartitions(sharePartitions1)
.withFetchId(fetchId1)
.build();
@@ -775,7 +775,7 @@ public class DelayedShareFetchTest {
BiConsumer<SharePartitionKey, Throwable> exceptionHandler =
mockExceptionHandler();
DelayedShareFetch delayedShareFetch2 =
spy(DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch2)
- .withReplicaManager(replicaManager)
+ .withReplicaManagerLogReader(replicaManager)
.withSharePartitions(sharePartitions2)
.withPartitionMaxBytesStrategy(partitionMaxBytesStrategy)
.withFetchId(fetchId2)
@@ -834,7 +834,7 @@ public class DelayedShareFetchTest {
DelayedShareFetch delayedShareFetch =
DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
- .withReplicaManager(replicaManager)
+ .withReplicaManagerLogReader(replicaManager)
.withSharePartitions(sharePartitions)
.withPartitionMaxBytesStrategy(partitionMaxBytesStrategy)
.build();
@@ -853,7 +853,7 @@ public class DelayedShareFetchTest {
logReadResponse.put(tp0, logReadResult);
doAnswer(invocation ->
buildLogReadResult(List.of(tp1))).when(replicaManager).readFromLog(any(),
any(), any(ReplicaQuota.class), anyBoolean());
- LinkedHashMap<TopicIdPartition, LogReadResult> combinedLogReadResponse
= delayedShareFetch.combineLogReadResponse(topicPartitionData, logReadResponse);
+ LinkedHashMap<TopicIdPartition, LogReadResult> combinedLogReadResponse
= delayedShareFetch.combineLogReadResponse(topicPartitionData,
logReadResponse).join();
assertEquals(topicPartitionData.keySet(),
combinedLogReadResponse.keySet());
assertEquals(combinedLogReadResponse.get(tp0),
logReadResponse.get(tp0));
@@ -861,7 +861,7 @@ public class DelayedShareFetchTest {
logReadResponse = new LinkedHashMap<>();
logReadResponse.put(tp0, mock(LogReadResult.class));
logReadResponse.put(tp1, mock(LogReadResult.class));
- combinedLogReadResponse =
delayedShareFetch.combineLogReadResponse(topicPartitionData, logReadResponse);
+ combinedLogReadResponse =
delayedShareFetch.combineLogReadResponse(topicPartitionData,
logReadResponse).join();
assertEquals(topicPartitionData.keySet(),
combinedLogReadResponse.keySet());
assertEquals(combinedLogReadResponse.get(tp0),
logReadResponse.get(tp0));
assertEquals(combinedLogReadResponse.get(tp1),
logReadResponse.get(tp1));
@@ -909,7 +909,7 @@ public class DelayedShareFetchTest {
DelayedShareFetch delayedShareFetch =
spy(DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
.withSharePartitions(sharePartitions)
- .withReplicaManager(replicaManager)
+ .withReplicaManagerLogReader(replicaManager)
.withExceptionHandler(exceptionHandler)
.withPartitionMaxBytesStrategy(partitionMaxBytesStrategy)
.withShareGroupMetrics(shareGroupMetrics)
@@ -977,7 +977,7 @@ public class DelayedShareFetchTest {
DelayedShareFetch delayedShareFetch =
spy(DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
.withSharePartitions(sharePartitions)
- .withReplicaManager(replicaManager)
+ .withReplicaManagerLogReader(replicaManager)
.withPartitionMaxBytesStrategy(partitionMaxBytesStrategy)
.withFetchId(fetchId)
.build());
@@ -1016,10 +1016,10 @@ public class DelayedShareFetchTest {
PartitionMaxBytesStrategy partitionMaxBytesStrategy =
mockPartitionMaxBytes(Set.of(tp0));
Uuid fetchId = Uuid.randomUuid();
- DelayedShareFetch delayedShareFetch =
DelayedShareFetchTest.DelayedShareFetchBuilder.builder()
+ DelayedShareFetch delayedShareFetch =
DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
.withSharePartitions(sharePartitions1)
- .withReplicaManager(replicaManager)
+ .withReplicaManagerLogReader(replicaManager)
.withPartitionMaxBytesStrategy(partitionMaxBytesStrategy)
.withFetchId(fetchId)
.build();
@@ -1057,11 +1057,11 @@ public class DelayedShareFetchTest {
ReplicaManager replicaManager = mock(ReplicaManager.class);
when(replicaManager.getPartitionOrException(tp0.topicPartition())).thenReturn(p0);
- DelayedShareFetch delayedShareFetch =
DelayedShareFetchTest.DelayedShareFetchBuilder.builder()
+ DelayedShareFetch delayedShareFetch =
DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
.withSharePartitions(sharePartitions)
.withFetchId(fetchId)
- .withReplicaManager(replicaManager)
+ .withReplicaManagerLogReader(replicaManager)
.build();
when(sp0.maybeAcquireFetchLock(fetchId)).thenReturn(true);
@@ -1174,7 +1174,7 @@ public class DelayedShareFetchTest {
DelayedShareFetch delayedShareFetch =
spy(DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
.withSharePartitions(sharePartitions)
- .withReplicaManager(replicaManager)
+ .withReplicaManagerLogReader(replicaManager)
.withPartitionMaxBytesStrategy(PartitionMaxBytesStrategy.type(PartitionMaxBytesStrategy.StrategyType.UNIFORM))
.withFetchId(fetchId)
.withExceptionHandler(exceptionHandler)
@@ -1272,7 +1272,7 @@ public class DelayedShareFetchTest {
DelayedShareFetch delayedShareFetch =
spy(DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
.withSharePartitions(sharePartitions)
- .withReplicaManager(replicaManager)
+ .withReplicaManagerLogReader(replicaManager)
.withPartitionMaxBytesStrategy(PartitionMaxBytesStrategy.type(PartitionMaxBytesStrategy.StrategyType.UNIFORM))
.withFetchId(fetchId)
.withExceptionHandler(exceptionHandler)
@@ -1342,7 +1342,7 @@ public class DelayedShareFetchTest {
DelayedShareFetch delayedShareFetch =
DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
- .withReplicaManager(replicaManager)
+ .withReplicaManagerLogReader(replicaManager)
.withSharePartitions(sharePartitions)
.withPartitionMaxBytesStrategy(PartitionMaxBytesStrategy.type(PartitionMaxBytesStrategy.StrategyType.UNIFORM))
.build();
@@ -1366,7 +1366,7 @@ public class DelayedShareFetchTest {
fetchableTopicPartitions.add(tp2);
// We will be doing replica manager fetch only for tp1 and tp2.
doAnswer(invocation ->
buildLogReadResult(fetchableTopicPartitions.stream().toList())).when(replicaManager).readFromLog(any(),
any(), any(ReplicaQuota.class), anyBoolean());
- LinkedHashMap<TopicIdPartition, LogReadResult> combinedLogReadResponse
= delayedShareFetch.combineLogReadResponse(topicPartitionData, logReadResponse);
+ LinkedHashMap<TopicIdPartition, LogReadResult> combinedLogReadResponse
= delayedShareFetch.combineLogReadResponse(topicPartitionData,
logReadResponse).join();
assertEquals(topicPartitionData.keySet(),
combinedLogReadResponse.keySet());
// Since only 2 partitions are fetchable but the third one has already
been fetched, maxbytes per partition = requestMaxBytes(i.e. 1024*1020) /
acquiredTopicPartitions(i.e. 3)
@@ -1467,7 +1467,7 @@ public class DelayedShareFetchTest {
DelayedShareFetch delayedShareFetch =
spy(DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
.withSharePartitions(sharePartitions)
- .withReplicaManager(replicaManager)
+ .withReplicaManagerLogReader(replicaManager)
.withPartitionMaxBytesStrategy(mockPartitionMaxBytes(Set.of(tp0,
tp1, tp2)))
.withFetchId(fetchId)
.build());
@@ -1527,7 +1527,7 @@ public class DelayedShareFetchTest {
DelayedShareFetch delayedShareFetch =
spy(DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
.withSharePartitions(sharePartitions)
- .withReplicaManager(replicaManager)
+ .withReplicaManagerLogReader(replicaManager)
.withPartitionMaxBytesStrategy(mockPartitionMaxBytes(Set.of(tp0)))
.withFetchId(fetchId)
.build());
@@ -1609,7 +1609,7 @@ public class DelayedShareFetchTest {
DelayedShareFetch delayedShareFetch =
spy(DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
.withSharePartitions(sharePartitions)
- .withReplicaManager(replicaManager)
+ .withReplicaManagerLogReader(replicaManager)
.withExceptionHandler(exceptionHandler)
.withPartitionMaxBytesStrategy(mockPartitionMaxBytes(Set.of(tp0,
tp1, tp2)))
.withFetchId(fetchId)
@@ -1711,7 +1711,7 @@ public class DelayedShareFetchTest {
DelayedShareFetch delayedShareFetch =
spy(DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
.withSharePartitions(sharePartitions)
- .withReplicaManager(replicaManager)
+ .withReplicaManagerLogReader(replicaManager)
.withPartitionMaxBytesStrategy(mockPartitionMaxBytes(Set.of(tp0, tp1, tp2)))
.withFetchId(fetchId)
.build());
@@ -1730,8 +1730,12 @@ public class DelayedShareFetchTest {
List<RemoteFetch> remoteFetches =
delayedShareFetch.pendingRemoteFetches().remoteFetches();
assertEquals(1, remoteFetches.size());
assertTrue(remoteFetches.get(0).remoteFetchTask().isCancelled());
- // Partition locks should be released for all 3 topic partitions
- Mockito.verify(delayedShareFetch,
times(1)).releasePartitionLocks(Set.of(tp0, tp1, tp2));
+ // Partition locks should be released for all 3 topic partitions.
tp0 and tp1 (local log read)
+ // locks are released twice: once when the remote fetch is
prepared (as non-remote partitions)
+ // and again after they are re-acquired for the additional local
read in onComplete. tp2 (remote
+ // storage read) lock is released once when the remote fetch
completes.
+ Mockito.verify(delayedShareFetch,
times(2)).releasePartitionLocks(Set.of(tp0, tp1));
+ Mockito.verify(delayedShareFetch,
times(1)).releasePartitionLocks(Set.of(tp2));
assertTrue(shareFetch.isCompleted());
// Share fetch response contained tp0 and tp1 (local fetch) but
not tp2, since it errored out.
assertEquals(Set.of(tp0, tp1), future.join().keySet());
@@ -1797,7 +1801,7 @@ public class DelayedShareFetchTest {
DelayedShareFetch delayedShareFetch =
spy(DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
.withSharePartitions(sharePartitions)
- .withReplicaManager(replicaManager)
+ .withReplicaManagerLogReader(replicaManager)
.withPartitionMaxBytesStrategy(mockPartitionMaxBytes(Set.of(tp0,
tp1)))
.withFetchId(fetchId)
.withExceptionHandler(exceptionHandler)
@@ -1876,7 +1880,7 @@ public class DelayedShareFetchTest {
DelayedShareFetch delayedShareFetch =
spy(DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
.withSharePartitions(sharePartitions)
- .withReplicaManager(replicaManager)
+ .withReplicaManagerLogReader(replicaManager)
.withPartitionMaxBytesStrategy(mockPartitionMaxBytes(Set.of(tp0)))
.withFetchId(fetchId)
.build());
@@ -1982,7 +1986,7 @@ public class DelayedShareFetchTest {
Uuid fetchId = Uuid.randomUuid();
DelayedShareFetch delayedShareFetch =
spy(DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
- .withReplicaManager(replicaManager)
+ .withReplicaManagerLogReader(replicaManager)
.withSharePartitions(sharePartitions)
.withPartitionMaxBytesStrategy(mockPartitionMaxBytes(Set.of(tp0, tp1, tp2)))
.withFetchId(fetchId)
@@ -2002,8 +2006,12 @@ public class DelayedShareFetchTest {
// the future of shareFetch completes.
assertTrue(shareFetch.isCompleted());
assertEquals(Set.of(tp0, tp1, tp2), future.join().keySet());
- // Verify the locks are released for both local log and remote
storage read topic partitions tp0, tp1 and tp2.
- Mockito.verify(delayedShareFetch,
times(1)).releasePartitionLocks(Set.of(tp0, tp1, tp2));
+ // Verify the locks are released for both local log and remote
storage read topic partitions.
+ // tp0 and tp1 (local log read) locks are released twice: once
when the remote fetch is prepared
+ // (as non-remote partitions) and again after they are re-acquired
for the additional local read
+ // in onComplete. tp2 (remote storage read) lock is released once
when the remote fetch completes.
+ Mockito.verify(delayedShareFetch,
times(2)).releasePartitionLocks(Set.of(tp0, tp1));
+ Mockito.verify(delayedShareFetch,
times(1)).releasePartitionLocks(Set.of(tp2));
assertEquals(Errors.NONE.code(),
future.join().get(tp0).errorCode());
assertEquals(Errors.NONE.code(),
future.join().get(tp1).errorCode());
assertEquals(Errors.NONE.code(),
future.join().get(tp2).errorCode());
@@ -2074,7 +2082,7 @@ public class DelayedShareFetchTest {
DelayedShareFetch delayedShareFetch =
spy(DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
.withSharePartitions(sharePartitions)
- .withReplicaManager(replicaManager)
+ .withReplicaManagerLogReader(replicaManager)
.withPartitionMaxBytesStrategy(mockPartitionMaxBytes(Set.of(tp0,
tp1)))
.withFetchId(fetchId)
.build());
@@ -2128,7 +2136,7 @@ public class DelayedShareFetchTest {
Uuid fetchId = Uuid.randomUuid();
DelayedShareFetch delayedShareFetch =
spy(DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
- .withReplicaManager(replicaManager)
+ .withReplicaManagerLogReader(replicaManager)
.withSharePartitions(sharePartitions)
.withPartitionMaxBytesStrategy(PartitionMaxBytesStrategy.type(PartitionMaxBytesStrategy.StrategyType.UNIFORM))
.withPendingRemoteFetches(pendingRemoteFetches)
@@ -2209,7 +2217,7 @@ public class DelayedShareFetchTest {
Uuid fetchId = Uuid.randomUuid();
DelayedShareFetch delayedShareFetch =
spy(DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
- .withReplicaManager(replicaManager)
+ .withReplicaManagerLogReader(replicaManager)
.withSharePartitions(sharePartitions)
.withPartitionMaxBytesStrategy(PartitionMaxBytesStrategy.type(PartitionMaxBytesStrategy.StrategyType.UNIFORM))
.withPendingRemoteFetches(pendingRemoteFetches)
@@ -2293,7 +2301,7 @@ public class DelayedShareFetchTest {
Uuid fetchId = Uuid.randomUuid();
DelayedShareFetch delayedShareFetch =
spy(DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
- .withReplicaManager(replicaManager)
+ .withReplicaManagerLogReader(replicaManager)
.withSharePartitions(sharePartitions)
.withPartitionMaxBytesStrategy(PartitionMaxBytesStrategy.type(PartitionMaxBytesStrategy.StrategyType.UNIFORM))
.withPendingRemoteFetches(pendingRemoteFetches)
@@ -2357,6 +2365,74 @@ public class DelayedShareFetchTest {
}
}
+ @Test
+ public void testTryCompleteAsyncCompletion() {
+ String groupId = "grp";
+ Uuid topicId = Uuid.randomUuid();
+ LogReader logReader = mock(LogReader.class);
+ PartitionMetadataProvider metadataProvider =
mock(PartitionMetadataProvider.class);
+ ReplicaManager replicaManager = mock(ReplicaManager.class);
+ TopicIdPartition tp0 = new TopicIdPartition(topicId, new
TopicPartition("foo", 0));
+
+ SharePartition sp0 = mock(SharePartition.class);
+ LinkedHashMap<TopicIdPartition, SharePartition> sharePartitions = new
LinkedHashMap<>();
+ sharePartitions.put(tp0, sp0);
+
+ ShareFetch shareFetch = new ShareFetch(FETCH_PARAMS, groupId,
Uuid.randomUuid().toString(),
+ new CompletableFuture<>(), List.of(tp0), BATCH_OPTIMIZED,
BATCH_SIZE, MAX_FETCH_RECORDS, BROKER_TOPIC_STATS);
+
+ CompletableFuture<LinkedHashMap<TopicIdPartition, LogReadResult>>
asyncReadFuture = new CompletableFuture<>();
+ when(logReader.readAsync(any(), any(), any(), any(),
anyBoolean())).thenReturn(asyncReadFuture);
+
+ when(sp0.maybeAcquireFetchLock(any())).thenReturn(true);
+ when(sp0.canAcquireRecords()).thenReturn(true);
+ when(sp0.nextFetchOffset()).thenReturn(0L);
+ when(sp0.fetchOffsetMetadata(anyLong()))
+ .thenReturn(Optional.empty())
+ .thenReturn(Optional.of(new LogOffsetMetadata(0, 1, 0)));
+ when(metadataProvider.endOffsetMetadata(any(), any())).thenReturn(new
LogOffsetMetadata(1, 1, 1));
+
+ PartitionMaxBytesStrategy partitionMaxBytesStrategy =
mockPartitionMaxBytes(Set.of(tp0));
+
+ Uuid fetchId = Uuid.randomUuid();
+ DelayedShareFetch delayedShareFetch =
DelayedShareFetchBuilder.builder()
+ .withShareFetchData(shareFetch)
+ .withSharePartitions(sharePartitions)
+ .withReplicaManager(replicaManager)
+ .withLogReader(logReader)
+ .withMetadataProvider(metadataProvider)
+ .withPartitionMaxBytesStrategy(partitionMaxBytesStrategy)
+ .withFetchId(fetchId)
+ .build();
+
+ // Pending fetch shall be null as of now.
+ assertNull(delayedShareFetch.pendingFetch());
+ // Async read should create pending fetch as the future is not
completed yet
+ assertFalse(delayedShareFetch.tryComplete());
+ // Validate pending fetch is created.
+ assertNotNull(delayedShareFetch.pendingFetch());
+ // Validate locks are held.
+ Mockito.verify(sp0, times(0)).releaseFetchLock(any());
+ // Invoke tryComplete again to simulate async read completion but
should return false as future
+ // is not yet completed.
+ assertFalse(delayedShareFetch.tryComplete());
+ // Complete future to simulate async read completion. The registered
callback should trigger a
+ // purgatory check rather than completing the request directly on the
completing thread.
+ asyncReadFuture.complete(buildLogReadResultMap(List.of(tp0)));
+ // Verify the callback handler executed the request completion.
+ Mockito.verify(replicaManager).completeDelayedShareFetchRequest(any());
+ // The request should not be completed yet as the purgatory check
(mocked) is a no-op here.
+ assertFalse(delayedShareFetch.isCompleted());
+ // Simulate the purgatory re-invoking tryComplete now that the async
read is done.
+ assertTrue(delayedShareFetch.tryComplete());
+ // Force complete should have been executed and the request must be
completed.
+ assertTrue(delayedShareFetch.isCompleted());
+
+ // Verify readFromLog was called (which is used by async readAsync
internally)
+ Mockito.verify(logReader).readAsync(any(), any(), any(), any(),
anyBoolean());
+ Mockito.verify(sp0).releaseFetchLock(fetchId);
+ }
+
static void mockTopicIdPartitionToReturnDataEqualToMinBytes(ReplicaManager
replicaManager, TopicIdPartition topicIdPartition, int minBytes) {
LogOffsetMetadata hwmOffsetMetadata = new LogOffsetMetadata(1, 1,
minBytes);
LogOffsetSnapshot endOffsetSnapshot = new LogOffsetSnapshot(1,
mock(LogOffsetMetadata.class),
@@ -2418,6 +2494,16 @@ public class DelayedShareFetchTest {
return mock(BiConsumer.class);
}
+ private static LinkedHashMap<TopicIdPartition, LogReadResult>
buildLogReadResultMap(List<TopicIdPartition> topicIdPartitions) {
+ return
CollectionConverters.asJava(buildLogReadResult(topicIdPartitions)).stream()
+ .collect(Collectors.toMap(
+ Tuple2::_1,
+ Tuple2::_2,
+ (oldValue, newValue) -> oldValue,
+ LinkedHashMap::new
+ ));
+ }
+
@SuppressWarnings("unchecked")
static class DelayedShareFetchBuilder {
private ShareFetch shareFetch = mock(ShareFetch.class);
@@ -2437,13 +2523,28 @@ public class DelayedShareFetchTest {
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;
+ }
+
+ DelayedShareFetchBuilder withLogReader(LogReader logReader) {
+ this.logReader = logReader;
+ return this;
+ }
+
+ DelayedShareFetchBuilder
withMetadataProvider(PartitionMetadataProvider metadataProvider) {
+ this.metadataProvider = metadataProvider;
+ return this;
+ }
+
DelayedShareFetchBuilder
withExceptionHandler(BiConsumer<SharePartitionKey, Throwable> exceptionHandler)
{
this.exceptionHandler = exceptionHandler;
return this;
diff --git
a/core/src/test/java/kafka/server/share/SharePartitionManagerTest.java
b/core/src/test/java/kafka/server/share/SharePartitionManagerTest.java
index 116a92f1bc9..010c83276f4 100644
--- a/core/src/test/java/kafka/server/share/SharePartitionManagerTest.java
+++ b/core/src/test/java/kafka/server/share/SharePartitionManagerTest.java
@@ -1771,7 +1771,7 @@ public class SharePartitionManagerTest {
DelayedShareFetch delayedShareFetch =
DelayedShareFetchTest.DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
- .withReplicaManager(mockReplicaManager)
+ .withReplicaManagerLogReader(mockReplicaManager)
.withSharePartitions(sharePartitions)
.withPartitionMaxBytesStrategy(PartitionMaxBytesStrategy.type(PartitionMaxBytesStrategy.StrategyType.UNIFORM))
.build();
@@ -1882,7 +1882,7 @@ public class SharePartitionManagerTest {
DelayedShareFetch delayedShareFetch =
DelayedShareFetchTest.DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
- .withReplicaManager(mockReplicaManager)
+ .withReplicaManagerLogReader(mockReplicaManager)
.withSharePartitions(sharePartitions)
.build();
@@ -1989,7 +1989,7 @@ public class SharePartitionManagerTest {
DelayedShareFetch delayedShareFetch =
DelayedShareFetchTest.DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
- .withReplicaManager(mockReplicaManager)
+ .withReplicaManagerLogReader(mockReplicaManager)
.withSharePartitions(sharePartitions)
.withPartitionMaxBytesStrategy(PartitionMaxBytesStrategy.type(PartitionMaxBytesStrategy.StrategyType.UNIFORM))
.build();
@@ -2097,7 +2097,7 @@ public class SharePartitionManagerTest {
DelayedShareFetch delayedShareFetch =
DelayedShareFetchTest.DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
- .withReplicaManager(mockReplicaManager)
+ .withReplicaManagerLogReader(mockReplicaManager)
.withSharePartitions(sharePartitions)
.build();
