apoorvmittal10 commented on code in PR #22707:
URL: https://github.com/apache/kafka/pull/22707#discussion_r3509316008
##########
core/src/main/java/kafka/server/share/ReplicaManagerLogReader.java:
##########
@@ -218,6 +222,33 @@ CompletableFuture<FetchDataInfo>
readRemote(RemoteStorageFetchInfo remoteStorage
future.completeExceptionally(e);
}
+ // Bound the wait on the remote read so a stalled remote tier cannot
pin the read (and the
+ // resources held while it is pending) indefinitely. Use the broker's
timer wheel - as
+ // DelayedShareFetch does for its remote fetch - rather than
CompletableFuture#orTimeout. On
+ // expiry the future completes exceptionally with a TimeoutException,
which the caller treats
+ // as a (skippable) read error.
+ if (!future.isDone()) {
+ long timeoutMs = remoteFetchMaxWaitMs();
+ TimerTask timeoutTask = new TimerTask(timeoutMs) {
+ @Override
+ public void run() {
+ future.completeExceptionally(new TimeoutException(
+ "Remote read for " + remoteStorageFetchInfo + " did
not complete within " + timeoutMs + " ms."));
Review Comment:
Just completing it exceptionally might not be a good idea as the method can
still return the locally fetched data while skipping the remote storage
partitions.
--
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]
