apoorvmittal10 commented on code in PR #22707:
URL: https://github.com/apache/kafka/pull/22707#discussion_r3504758362
##########
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."));
+ }
+ };
+ // Cancel the timer task once the read completes (either outcome)
so it does not linger in the wheel.
+ future.whenComplete((info, exception) -> timeoutTask.cancel());
+ replicaManager.addShareFetchTimerRequest(timeoutTask);
Review Comment:
Also why share fetch timer API being used here? Also should we make it
callers responsibility to handle the timeouts?
--
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]
