bhattmanish98 commented on code in PR #8830:
URL: https://github.com/apache/hadoop/pull/8830#discussion_r4523398457
##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/ReadBufferManagerV2.java:
##########
@@ -632,7 +750,14 @@ public void doneReading(final ReadBuffer buffer,
buffer.setStatus(ReadBufferStatus.AVAILABLE);
buffer.setLength(bytesActuallyRead);
}
+ if(parentBuffer != null) {
+ parentBuffer.decrementActiveChildren();
+ }
} else {
+ // If a child buffer, then reset the active child count to 0 in
parent buffer
+ if (parentBuffer != null) {
+ parentBuffer.resetActiveChildCount();
Review Comment:
Confirmed: this can recycle an array while a sibling is still writing. The
local fix decrements the parent count once per completed child (including
failures), retains the shared allocation, and evicts the entire sibling group
only when no children or readers remain. Queue cancellation and stream-close
cleanup follow the same ownership rule. Regression coverage:
testFailedChildRetainsBufferUntilAllWritersFinish and
testCloseDoesNotReleaseSharedBufferWithRunningSibling.
These changes are prepared and verified locally (120 focused unit tests
passed), but are not yet committed or pushed to this PR. Leaving the thread
open until the updated patch is published.
##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/ReadBufferManagerV2.java:
##########
@@ -279,16 +276,135 @@ void init() {
}
/**
- * {@link AbfsInputStream} calls this method to queueing read-ahead.
- * @param stream which read-ahead is requested from.
- * @param requestedOffset The offset in the file which should be read.
- * @param requestedLength The length to read.
+ * {@inheritDoc}
*/
@Override
public void queueReadAhead(final AbfsInputStream stream,
final long requestedOffset,
final int requestedLength,
+ final List<BlobLayout.BlobRange> segments,
TracingContext tracingContext) {
+ synchronized (this) {
+ if (isAlreadyQueued(stream.getETag(), requestedOffset)) {
+ // Already queued for this offset, so skip queuing.
+ printTraceLog(
+ "Skipping queuing readAhead for file: {}, with eTag: {}, "
+ + "offset: {}, triggered by stream: {} as it is already
queued",
+ stream.getPath(), stream.getETag(), requestedOffset,
+ stream.hashCode());
+ return;
+ }
+
+ if (isFreeListEmpty() && !tryMemoryUpscale() && !tryEvict()) {
+ // No buffers are available and more buffers cannot be created. Skip
queuing.
+ printTraceLog(
+ "Skipping queuing readAhead for file: {}, with eTag: {}, offset:
{}, triggered by stream: {} as no buffers are available",
+ stream.getPath(), stream.getETag(), requestedOffset,
+ stream.hashCode());
+ return;
+ }
+
+ // Create new parent for this window
+ ReadBuffer parentBuffer = new ReadBuffer();
+ // To map buffer with stream that requested it
+ parentBuffer.setStream(stream);
+ // To map buffer with file it belongs to
+ parentBuffer.setETag(stream.getETag());
+ parentBuffer.setPath(stream.getPath());
+ parentBuffer.setOffset(requestedOffset);
+ parentBuffer.setBufferOffset(0);
+ parentBuffer.setLength(0);
+ parentBuffer.setStatus(ReadBufferStatus.NOT_AVAILABLE);
+ parentBuffer.setTracingContext(tracingContext);
+ parentBuffer.setRequestedLength(requestedLength);
+ parentBuffer.setParentToTrue();
+
+ if (isFreeListEmpty()) {
+ /*
+ * By now there should be at least one buffer available.
+ * This is to double sure that after upscaling or eviction,
+ * we still have free buffer available. If not, we skip queueing.
+ */
+ return;
+ }
+ Integer bufferIndex = popFromFreeList();
+ if (bufferIndex >= bufferPool.length) {
+ // This should never happen.
+ printTraceLog(
+ "Skipping queuing readAhead for file: {}, with eTag: {}, offset:
{}, triggered by stream: {} as invalid buffer index popped from free list",
+ stream.getPath(), stream.getETag(), requestedOffset,
+ stream.hashCode());
+ return;
+ }
+ parentBuffer.setBuffer(bufferPool[bufferIndex]);
+ parentBuffer.setBufferindex(bufferIndex);
+
+ printTraceLog(
+ "Created parent buffer for offset: {}, with index: {}, for file: {},
eTag: {}",
+ requestedOffset, bufferIndex, stream.getPath(), stream.getETag());
+
+ // Iterate over the segments and create children
+ for (BlobLayout.BlobRange segment : segments) {
+ long readStart = segment.start();
+ long readEnd = segment.end();
+ int rangeLength = Math.toIntExact(readEnd - readStart + 1);
+
+ printTraceLog(
+ "Start Queueing readAhead childBuffer for file: {}, with eTag: {},
"
+ + "offset: {}, length: {}, segments: {}, triggered by stream:
{}",
+ stream.getPath(), stream.getETag(), readStart, rangeLength,
+ segments.size(), stream.hashCode());
+
+ // Create child buffer for this segment
+ ReadBuffer childBuffer = new ReadBuffer();
+ childBuffer.setStream(stream);
+ childBuffer.setETag(
+ stream.getETag()); // To map buffer with file it belongs to
+ childBuffer.setPath(stream.getPath());
+ childBuffer.setOffset(readStart); // File position for this segment
+ childBuffer.setBufferOffset(
+ parentBuffer.getBufferOffset()); // Offset in shared array
+ childBuffer.setLength(0);
+ childBuffer.setRequestedLength(rangeLength);
+ childBuffer.setStatus(ReadBufferStatus.NOT_AVAILABLE);
+ childBuffer.setLatch(new CountDownLatch(1));
+ childBuffer.setTracingContext(tracingContext);
+
+ childBuffer.setBuffer(parentBuffer.getBuffer()); // Share parent's
array
+ childBuffer.setBufferindex(
+ parentBuffer.getBufferindex()); // Share parent's index
+ childBuffer.setParentBuffer(parentBuffer); // Link to parent
+ childBuffer.setReadEndpoint(segment.host());
+
+ parentBuffer.pushToChildBufferList(childBuffer);
+ parentBuffer.incrementActiveChildren();
+ parentBuffer.setBufferOffset(
Review Comment:
Confirmed: partial coverage and cross-host overlaps can misalign or overrun
the shared window. The local fix validates exact, ordered, contiguous coverage
before allocating children; gaps, overlaps, and out-of-window ranges fall back
to default-host single-buffer read-ahead. Each child uses readStart -
requestedOffset as its array offset. Regression coverage checks missing
prefixes/suffixes, middle gaps, overlaps, out-of-window ranges, and the
complete-window byte contents.
These changes are prepared and verified locally (120 focused unit tests
passed), but are not yet committed or pushed to this PR. Leaving the thread
open until the updated patch is published.
##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/ReadBufferManagerV2.java:
##########
@@ -890,43 +1119,128 @@ private ReadBuffer clearFromReadAheadQueue(final String
eTag,
*/
private int getBlockFromCompletedQueue(final String eTag, final long
position,
final int length, final byte[] buffer) throws IOException {
- ReadBuffer buf = getBufferFromCompletedQueue(eTag, position);
+ // Find any child buffer from the window that covers the requested position
+ ReadBuffer anyChild = getBufferFromCompletedQueue(eTag, position);
- if (buf == null) {
+ if (anyChild == null) {
+ // No buffer covers this position - cache miss
+ printTraceLog(
+ "No buffer in completed queue covers position: {} for eTag: {}",
+ position, eTag);
return 0;
}
- buf.startReading(); // atomic increment of refCount.
+ // Get the parent and its children
+ ReadBuffer parent = anyChild.getParentBuffer();
+
+ if (parent == null) {
+ // This is a standalone buffer (not part of a parent/child structure)
+ // Fall back to simple single-buffer copy
+ return copyFromSingleBuffer(anyChild, position, length, buffer);
+ }
+
+ List<ReadBuffer> children = parent.getChildBufferList();
+ // Increment refCount on ALL children first to prevent eviction
+ for (ReadBuffer child : children) {
+ child.startReading();
+ }
+
+ try {
+ for (ReadBuffer child : children) {
+ if (child.getStatus() == ReadBufferStatus.READ_FAILED) {
Review Comment:
Confirmed. The local fix requires every child to be AVAILABLE with its full
requested length before copying the shared window; queued, in-progress, or
short-read siblings produce a cache miss rather than exposing unfilled bytes.
Existing failed-read exception handling is preserved. Regression coverage:
testQueuedSiblingCannotExposeUnfilledBytes and
testV2NonblockingPeekCannotCopyUnfinishedParentChildren.
These changes are prepared and verified locally (120 focused unit tests
passed), but are not yet committed or pushed to this PR. Leaving the thread
open until the updated patch is published.
##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/BlobLayoutCache.java:
##########
@@ -0,0 +1,610 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.azurebfs.services;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.github.benmanes.caffeine.cache.Expiry;
+import com.github.benmanes.caffeine.cache.Ticker;
+import org.checkerframework.checker.nullness.qual.NonNull;
+
+import org.apache.hadoop.classification.VisibleForTesting;
+import org.apache.hadoop.fs.azurebfs.contracts.services.BlobLayoutResponse;
+import
org.apache.hadoop.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
+
+import static org.apache.hadoop.fs.azurebfs.services.AbfsInputStream.LOG;
+
+/**
+ * BlobLayoutCache is a singleton class that manages caching of BlobLayout
objects for Azure Blob File System (ABFS).
+ * It uses a Caffeine cache to store layout information for blobs, optimizing
range fetches and minimizing redundant network calls.
+ *
+ * <p>Key Features:</p>
+ * <ul>
+ * <li>Evicts cache entries based on idle timeout and cache weight (number
of cached ranges).</li>
+ * <li>Tracks active streams to prevent eviction while a stream is open.</li>
+ * <li>Manages in-flight promises for range fetches to avoid duplicate
requests.</li>
+ * <li>Provides methods to register/deregister streams, retrieve cached
layouts, compute gaps, and manage promises.</li>
+ * </ul>
+ *
+ * <p>Usage:</p>
+ * <ul>
+ * <li>Call {@link #registerStream(String, long)} when a stream is
opened.</li>
+ * <li>Call {@link #deregisterStream(String)} when a stream is closed.</li>
+ * <li>Use {@link #getBlobLayout(String, long, long)} and {@link
#getGaps(String, long, long)} to query cached layout info.</li>
+ * <li>Use {@link #putBlobLayout(String, BlobLayoutResponse, long)} to
update the cache with new layout info.</li>
+ * <li>Use {@link #getBridgeGap(String, long, long)} to determine the next
range to fetch, considering in-flight and cached ranges.</li>
+ * <li>Use {@link #removePromise(String, long, long)} to remove completed
in-flight promises.</li>
+ * </ul>
+ */
+public final class BlobLayoutCache {
+
+ /**
+ * LayoutEntry represents a cache entry for a blob's layout.
+ * It tracks the BlobLayout, active stream count, and last deregistered time
for eviction logic.
+ */
+ private static class LayoutEntry {
+
+ /** The BlobLayout instance for this entry. */
+ private final BlobLayout layout;
+
+ /** Indicates if the layout is present and valid. */
+ private AtomicBoolean isLayoutPresent = new AtomicBoolean(true);
+
+ /** Number of active streams using this layout. */
+ private final AtomicInteger activeStreams = new AtomicInteger(0);
+
+ /** Last time (in nanoseconds) when all streams were deregistered. Used
for idle eviction. */
+ private final AtomicLong lastDeregisteredTimeInNanos = new AtomicLong(
+ Long.MAX_VALUE);
+
+ /**
+ * Constructs a LayoutEntry with the given content length.
+ * @param contentLength the length of the blob content
+ */
+ LayoutEntry(long contentLength) {
+ this.layout = new BlobLayout(contentLength);
+ }
+
+ /**
+ * Checks if the layout is unavailable (not present).
+ * @return true if layout is unavailable, false otherwise
+ */
+ boolean isLayoutUnavailable() {
+ return !isLayoutPresent.get();
+ }
+ }
+
+ /**
+ * Caffeine Cache instance replacing ConcurrentHashMap.
+ * Uses weighted eviction based on the number of ranges stored in each
layout.
+ */
+ private final Cache<String, LayoutEntry> cache;
+
+ /** Idle timeout in milliseconds for cache eviction. */
+ private final long idleTimeoutMs;
+
+ /** Maximum cache weight (total number of cached ranges) before eviction. */
+ private final long maxCacheWeight;
+
+ /** Time source used for idle-eviction bookkeeping (injectable for tests). */
+ private final Ticker ticker;
+
+ /** Singleton instance of BlobLayoutCache. */
+ private static volatile BlobLayoutCache instance = null;
+
+ /** Guards against repeatedly logging the singleton config-mismatch warning.
*/
+ private static volatile boolean configMismatchWarned = false;
+
+ /**
+ * Registry of in-flight promises for range fetches, keyed by file ETag.
+ * Each value is a thread-safe list of InFlightPromise objects.
+ */
+ private final ConcurrentHashMap<String,
CopyOnWriteArrayList<InFlightPromise>>
+ promiseRegistry = new ConcurrentHashMap<>();
+
+ /**
+ * InFlightPromise represents a pending fetch for a specific range.
+ * @param start the start offset of the range
+ * @param end the end offset of the range
+ * @param future the CompletableFuture representing the fetch operation
+ */
+ public record InFlightPromise(long start, long end,
+ CompletableFuture<Void> future) {}
+
+ /**
+ * Executor to asynchronously execute get blob layout calls
+ */
+ private static final ExecutorService FETCH_EXECUTOR = new ThreadPoolExecutor(
+ 8, 32, 60L, TimeUnit.SECONDS,
+ new LinkedBlockingQueue<>(1024), // Bounded queue to prevent OOM
+ new ThreadFactoryBuilder()
+ .setNameFormat("abfs-blob-layout-fetch-%d")
+ .setDaemon(true)
+ .build(),
+ // AbortPolicy keeps the executor non-blocking: when the pool and queue
are
+ // saturated, submission throws RejectedExecutionException instead of
running
+ // the fetch on the caller (read) thread. The submitter is responsible
for
+ // cleaning up the in-flight promise so a later read can retry the gap.
+ new ThreadPoolExecutor.AbortPolicy());
+
+ /**
+ * Optional override for the fetch executor, used only by tests to inject a
+ * deterministic executor (e.g. one that always rejects). {@code null} means
the
+ * shared {@link #FETCH_EXECUTOR} is used.
+ */
+ private static volatile ExecutorService fetchExecutorOverride = null;
+
+ static {
+ // Add fetch executor to JVM shutdown hook to ensure orderly shutdown of
thread during JVM shutdown.
+ Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+ FETCH_EXECUTOR.shutdown();
+ try {
+ if (!FETCH_EXECUTOR.awaitTermination(5, TimeUnit.SECONDS)) {
+ FETCH_EXECUTOR.shutdownNow();
+ }
+ } catch (InterruptedException e) {
+ FETCH_EXECUTOR.shutdownNow();
+ }
+ }));
+ }
+
+ /**
+ * Private constructor for singleton pattern.
+ * @param evictionTime the idle timeout in minutes for cache eviction
+ * @param maxCacheWeight the maximum total number of cached ranges before
eviction
+ * @param ticker the time source used for idle-eviction bookkeeping
+ */
+ private BlobLayoutCache(long evictionTime, long maxCacheWeight, Ticker
ticker) {
+ this.idleTimeoutMs = TimeUnit.MINUTES.toMillis(evictionTime);
+ this.maxCacheWeight = maxCacheWeight;
+ this.ticker = ticker;
+ this.cache = Caffeine.newBuilder()
+ .maximumWeight(maxCacheWeight)
+ .ticker(ticker)
+ // Weight is determined by the number of cached ranges in the layout
+ .weigher(
+ (String key, LayoutEntry entry) -> entry.layout.getRangeMapSize())
+ .expireAfter(new Expiry<String, LayoutEntry>() {
+ /**
+ * Calculates the expiry time for a cache entry based on active
streams and idle time.
+ * @param value the LayoutEntry
+ * @param currentTime the current time in nanoseconds
+ * @return the time in nanoseconds until expiry
+ */
+ private long calculateExpiry(LayoutEntry value, long currentTime) {
+ long lastDeregistered = value.lastDeregisteredTimeInNanos.get();
+ // Do not expire while a stream is active, or before the first
+ // deregistration has happened (sentinel Long.MAX_VALUE). The
latter
+ // check also avoids long overflow in the subtraction below.
+ if (value.activeStreams.get() > 0
+ || lastDeregistered == Long.MAX_VALUE) {
+ return Long.MAX_VALUE;
+ }
+
+ // How long has it been since the last stream closed?
+ long elapsedNanos = currentTime - lastDeregistered;
+ long thresholdNanos = TimeUnit.MILLISECONDS.toNanos(
+ idleTimeoutMs);
+
+ // If we are already past the threshold, expire immediately
(return 0)
+ // Otherwise, return the remaining time until the threshold is hit
+ return Math.max(0, thresholdNanos - elapsedNanos);
+ }
+
+ @Override
+ public long expireAfterCreate(@NonNull String key,
+ @NonNull LayoutEntry value, long currentTime) {
+ return calculateExpiry(value, currentTime);
+ }
+
+ @Override
+ public long expireAfterUpdate(@NonNull String key,
+ @NonNull LayoutEntry value, long currentTime,
+ long currentDuration) {
+ return calculateExpiry(value, currentTime);
+ }
+
+ @Override
+ public long expireAfterRead(@NonNull String key,
+ @NonNull LayoutEntry value, long currentTime,
+ long currentDuration) {
+ // LEASE LOGIC: If streams are active, the entry never expires.
+ // Once activeStreams == 0, the 5-minute idle clock starts.
+ return calculateExpiry(value, currentTime);
+ }
+ })
+ .removalListener((key, value, cause) -> {
+ // Cleanup corresponding promises when an entry is evicted from cache
+ if (key != null) {
+ promiseRegistry.remove(key);
+ }
+ })
+ .build();
+ }
+
+ /**
+ * Returns the process-wide singleton instance of BlobLayoutCache, creating
it
+ * if necessary.
+ * <p>
+ * The cache is intentionally shared across all filesystem instances in the
JVM
+ * so that its total memory footprint stays bounded by a single weight
budget.
+ * As a consequence, the {@code evictionTime} and {@code maxCacheWeight}
supplied
+ * by the <em>first</em> caller win; values from later callers are ignored.
When a
+ * later caller requests a different configuration, a one-time warning is
logged
+ * rather than silently discarding it.
+ * </p>
+ * @param evictionTime the idle timeout in minutes for cache eviction
+ * @param maxCacheWeight the maximum total number of cached ranges before
eviction
+ * @return the singleton BlobLayoutCache instance
+ */
+ public static BlobLayoutCache getInstance(long evictionTime,
+ long maxCacheWeight) {
+ BlobLayoutCache local = instance;
+ if (local == null) {
+ synchronized (BlobLayoutCache.class) {
+ local = instance;
+ if (local == null) {
+ local = new BlobLayoutCache(evictionTime, maxCacheWeight,
+ Ticker.systemTicker());
+ instance = local;
+ }
+ }
+ }
+ warnIfConfigDiffers(local, evictionTime, maxCacheWeight);
+ return local;
+ }
+
+ /**
+ * Logs a one-time warning if a caller requests a configuration that differs
+ * from the one the singleton was initialized with.
+ */
+ private static void warnIfConfigDiffers(BlobLayoutCache existing,
+ long evictionTime, long maxCacheWeight) {
+ if (configMismatchWarned) {
+ return;
+ }
+ long requestedIdleMs = TimeUnit.MINUTES.toMillis(evictionTime);
+ if (existing.idleTimeoutMs != requestedIdleMs
+ || existing.maxCacheWeight != maxCacheWeight) {
+ configMismatchWarned = true;
+ LOG.warn("BlobLayoutCache is process-wide and already initialized with "
+ + "evictionMs={}, maxWeight={}; ignoring requested
evictionMs={}, "
+ + "maxWeight={}.", existing.idleTimeoutMs,
existing.maxCacheWeight,
+ requestedIdleMs, maxCacheWeight);
+ }
+ }
+
+ /**
+ * Creates an isolated (non-singleton) cache instance for deterministic
testing,
+ * allowing a custom {@link Ticker} so idle-based eviction can be exercised
+ * without real time passing.
+ * @param evictionTime the idle timeout in minutes for cache eviction
+ * @param maxCacheWeight the maximum total number of cached ranges before
eviction
+ * @param ticker the time source to use
+ * @return a fresh, isolated BlobLayoutCache instance
+ */
+ @VisibleForTesting
+ static BlobLayoutCache newInstanceForTesting(long evictionTime,
+ long maxCacheWeight, Ticker ticker) {
+ return new BlobLayoutCache(evictionTime, maxCacheWeight, ticker);
+ }
+
+ /**
+ * Forces any pending cache maintenance (eviction/expiration) to run. For
tests.
+ */
+ @VisibleForTesting
+ void cleanUp() {
+ cache.cleanUp();
+ }
+
+ /**
+ * Returns true if an entry for the given key is currently present (after
running
+ * pending maintenance). For tests.
+ * @param key the cache key (file ETag)
+ * @return true if the entry is still cached
+ */
+ @VisibleForTesting
+ boolean isCached(String key) {
+ cache.cleanUp();
+ return cache.getIfPresent(key) != null;
+ }
+
+ /**
+ * Returns the approximate number of entries in the cache (after running
pending
+ * maintenance). For tests.
+ * @return the estimated cache size
+ */
+ @VisibleForTesting
+ long estimatedSize() {
+ cache.cleanUp();
+ return cache.estimatedSize();
+ }
+
+ /**
+ * Returns the shared executor used to run asynchronous blob layout fetch
tasks.
+ * Package-private: intended for use by {@code AbfsInputStream} only.
+ * @return the shared fetch executor
+ */
+ static ExecutorService getFetchExecutor() {
+ ExecutorService override = fetchExecutorOverride;
+ return override != null ? override : FETCH_EXECUTOR;
+ }
+
+ /**
+ * Overrides the fetch executor for tests so rejection/non-blocking
behaviour can
+ * be exercised deterministically. Callers must invoke
+ * {@link #resetFetchExecutorForTesting()} afterwards.
+ * @param executor the executor to use, or {@code null} to reset
+ */
+ @VisibleForTesting
+ static void setFetchExecutorForTesting(ExecutorService executor) {
+ fetchExecutorOverride = executor;
+ }
+
+ /**
+ * Restores the default shared fetch executor after a test override. For
tests.
+ */
+ @VisibleForTesting
+ static void resetFetchExecutorForTesting() {
+ fetchExecutorOverride = null;
+ }
+
+ /**
+ * Returns the number of in-flight fetch promises currently registered for
the
+ * given file ETag. Used by tests to assert that promises are not leaked.
For tests.
+ * @param fileETag the file ETag key
+ * @return the number of in-flight promises, or 0 if none are registered
+ */
+ @VisibleForTesting
+ int getInFlightPromiseCount(String fileETag) {
+ CopyOnWriteArrayList<InFlightPromise> promises =
promiseRegistry.get(fileETag);
+ return promises == null ? 0 : promises.size();
+ }
+
+ /**
+ * Registers a stream for the given file ETag and content length.
+ * Increments the active stream count and blocks eviction.
+ * @param fileETag the file ETag
+ * @param contentLength the content length of the blob
+ */
+ public void registerStream(String fileETag, long contentLength) {
+ cache.asMap().compute(fileETag, (key, entry) -> {
+ if (entry == null) {
+ entry = new LayoutEntry(contentLength);
+ }
+ entry.activeStreams.incrementAndGet();
+ entry.lastDeregisteredTimeInNanos.set(
+ Long.MAX_VALUE); // Block eviction logic
+ return entry;
+ });
+ }
+
+ /**
+ * Deregisters a stream for the given file ETag.
+ * Decrements the active stream count and updates the last deregistered time
for eviction.
+ * @param fileETag the file ETag
+ */
+ public void deregisterStream(String fileETag) {
+ cache.asMap().computeIfPresent(fileETag, (key, entry) -> {
+ // Clamp at zero so unbalanced deregister calls cannot drive the counter
+ // negative and keep resetting the idle clock.
+ int remaining = entry.activeStreams.updateAndGet(v -> v > 0 ? v - 1 : 0);
+ if (remaining == 0) {
+ entry.lastDeregisteredTimeInNanos.set(ticker.read());
+ }
+ return entry;
+ });
+ }
+
+ /**
+ * Retrieves the cached blob layout ranges for the specified key and range.
+ * @param key the cache key (file ETag)
+ * @param start the start offset
+ * @param end the end offset
+ * @return a list of BlobRange objects, or null if not available
+ */
+ public List<BlobLayout.BlobRange> getBlobLayout(final String key,
+ final long start, final long end) {
+ LayoutEntry layoutEntry = cache.getIfPresent(key);
+ return (layoutEntry == null || layoutEntry.isLayoutUnavailable())
+ ? null : layoutEntry.layout.getRanges(start, end);
+ }
+
+ /**
+ * Retrieves the gaps (unfetched ranges) in the cached layout for the
specified key and range.
+ * @param key the cache key (file ETag)
+ * @param start the start offset
+ * @param end the end offset
+ * @return a list of BlobRange objects representing gaps, or null if not
available
+ */
+ public List<BlobLayout.BlobRange> getGaps(final String key,
+ final long start, final long end) {
+ LayoutEntry layoutEntry = cache.getIfPresent(key);
+ return layoutEntry == null || layoutEntry.isLayoutUnavailable()
+ ? null : layoutEntry.layout.getGaps(start, end);
+ }
+
+ /**
+ * Determines the next range to fetch (bridge gap) for the given file ETag,
position, and max fetch size.
+ * Considers both cached and in-flight (promised) ranges to avoid redundant
fetches.
+ * @param fileETag the file ETag
+ * @param pos the current position
+ * @param maxFetch the maximum fetch size
+ * @return a BlobRange representing the next gap to fetch, or null if none
+ */
+ public BlobLayout.BlobRange getBridgeGap(String fileETag,
+ long pos,
+ long maxFetch) {
+ LayoutEntry layoutEntry = cache.getIfPresent(fileETag);
+ if (layoutEntry == null || layoutEntry.isLayoutUnavailable()) {
+ return null;
+ }
+
+ long fetchStart = layoutEntry.layout.getFetchStart(pos, maxFetch);
+ if (fetchStart == -1) {
+ return null;
+ }
+
+ long nextPromisedStart = Long.MAX_VALUE;
+ CopyOnWriteArrayList<InFlightPromise> filePromises = promiseRegistry.get(
+ fileETag);
+ if (filePromises != null) {
+ for (InFlightPromise p : filePromises) {
+ if (p.start() > pos && p.start() < nextPromisedStart) {
+ nextPromisedStart = p.start();
+ }
+ }
+ }
+
+ long nextCachedStart = layoutEntry.layout.getNextCachedStart(pos);
+ long nextWall = Math.min(nextCachedStart, nextPromisedStart);
+
+ long fetchEnd;
+ if (nextWall != Long.MAX_VALUE && (nextWall - fetchStart <= maxFetch)) {
+ fetchEnd = nextWall - 1;
+ } else {
+ fetchEnd = Math.min(layoutEntry.layout.getContentLength() - 1,
+ fetchStart + maxFetch - 1);
+ }
+
+ return new BlobLayout.BlobRange(fetchStart, fetchEnd, null);
+ }
+
+ /**
+ * Updates the cache with a new BlobLayoutResponse for the specified key.
+ * <p>
+ * Semantics:
+ * <ul>
+ * <li>A {@code null} response represents a <em>transient</em> fetch
failure. The
+ * entry is left untouched (still available) so the next read can retry
the gap;
+ * it is <strong>not</strong> permanently poisoned.</li>
+ * <li>A non-null response with no endpoints and no ranges means the
server has
+ * no distributed layout for this blob. The entry is marked unavailable so
callers
+ * fall back to host-URL based fetches.</li>
+ * <li>A non-null response with layout data is merged in and the entry is
(re)marked
+ * available, allowing recovery from a previous unavailable state.</li>
+ * </ul>
+ * The update runs inside {@link
java.util.concurrent.ConcurrentHashMap#compute}
+ * (via the cache's map view) so that Caffeine re-evaluates the entry weight
after
+ * ranges are added; otherwise weight-based eviction would never see the
growth.
+ * </p>
+ * @param key the cache key (file ETag)
+ * @param layoutResponse the BlobLayoutResponse containing new layout info
+ * @param contentLength the content length of the blob
+ */
+ public void putBlobLayout(final String key,
+ final BlobLayoutResponse layoutResponse,
+ final long contentLength) {
+ cache.asMap().compute(key, (k, existing) -> {
+ LayoutEntry entry = (existing != null)
Review Comment:
Confirmed. The local fix initializes an entry's idle timestamp from the
injected ticker when it is created, so entries created without registered
streams no longer have an unlimited idle lifetime. Registration still protects
active entries from idle expiry. testPutRecreatesEntryAfterEviction now checks
that the recreated layout survives just before the deadline and expires
immediately after it.
These changes are prepared and verified locally (120 focused unit tests
passed), but are not yet committed or pushed to this PR. Leaving the thread
open until the updated patch is published.
##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/BlobLayoutCache.java:
##########
@@ -0,0 +1,610 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.azurebfs.services;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.github.benmanes.caffeine.cache.Expiry;
+import com.github.benmanes.caffeine.cache.Ticker;
+import org.checkerframework.checker.nullness.qual.NonNull;
+
+import org.apache.hadoop.classification.VisibleForTesting;
+import org.apache.hadoop.fs.azurebfs.contracts.services.BlobLayoutResponse;
+import
org.apache.hadoop.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
+
+import static org.apache.hadoop.fs.azurebfs.services.AbfsInputStream.LOG;
+
+/**
+ * BlobLayoutCache is a singleton class that manages caching of BlobLayout
objects for Azure Blob File System (ABFS).
+ * It uses a Caffeine cache to store layout information for blobs, optimizing
range fetches and minimizing redundant network calls.
+ *
+ * <p>Key Features:</p>
+ * <ul>
+ * <li>Evicts cache entries based on idle timeout and cache weight (number
of cached ranges).</li>
+ * <li>Tracks active streams to prevent eviction while a stream is open.</li>
+ * <li>Manages in-flight promises for range fetches to avoid duplicate
requests.</li>
+ * <li>Provides methods to register/deregister streams, retrieve cached
layouts, compute gaps, and manage promises.</li>
+ * </ul>
+ *
+ * <p>Usage:</p>
+ * <ul>
+ * <li>Call {@link #registerStream(String, long)} when a stream is
opened.</li>
+ * <li>Call {@link #deregisterStream(String)} when a stream is closed.</li>
+ * <li>Use {@link #getBlobLayout(String, long, long)} and {@link
#getGaps(String, long, long)} to query cached layout info.</li>
+ * <li>Use {@link #putBlobLayout(String, BlobLayoutResponse, long)} to
update the cache with new layout info.</li>
+ * <li>Use {@link #getBridgeGap(String, long, long)} to determine the next
range to fetch, considering in-flight and cached ranges.</li>
+ * <li>Use {@link #removePromise(String, long, long)} to remove completed
in-flight promises.</li>
+ * </ul>
+ */
+public final class BlobLayoutCache {
+
+ /**
+ * LayoutEntry represents a cache entry for a blob's layout.
+ * It tracks the BlobLayout, active stream count, and last deregistered time
for eviction logic.
+ */
+ private static class LayoutEntry {
+
+ /** The BlobLayout instance for this entry. */
+ private final BlobLayout layout;
+
+ /** Indicates if the layout is present and valid. */
+ private AtomicBoolean isLayoutPresent = new AtomicBoolean(true);
+
+ /** Number of active streams using this layout. */
+ private final AtomicInteger activeStreams = new AtomicInteger(0);
+
+ /** Last time (in nanoseconds) when all streams were deregistered. Used
for idle eviction. */
+ private final AtomicLong lastDeregisteredTimeInNanos = new AtomicLong(
+ Long.MAX_VALUE);
+
+ /**
+ * Constructs a LayoutEntry with the given content length.
+ * @param contentLength the length of the blob content
+ */
+ LayoutEntry(long contentLength) {
+ this.layout = new BlobLayout(contentLength);
+ }
+
+ /**
+ * Checks if the layout is unavailable (not present).
+ * @return true if layout is unavailable, false otherwise
+ */
+ boolean isLayoutUnavailable() {
+ return !isLayoutPresent.get();
+ }
+ }
+
+ /**
+ * Caffeine Cache instance replacing ConcurrentHashMap.
+ * Uses weighted eviction based on the number of ranges stored in each
layout.
+ */
+ private final Cache<String, LayoutEntry> cache;
+
+ /** Idle timeout in milliseconds for cache eviction. */
+ private final long idleTimeoutMs;
+
+ /** Maximum cache weight (total number of cached ranges) before eviction. */
+ private final long maxCacheWeight;
+
+ /** Time source used for idle-eviction bookkeeping (injectable for tests). */
+ private final Ticker ticker;
+
+ /** Singleton instance of BlobLayoutCache. */
+ private static volatile BlobLayoutCache instance = null;
+
+ /** Guards against repeatedly logging the singleton config-mismatch warning.
*/
+ private static volatile boolean configMismatchWarned = false;
+
+ /**
+ * Registry of in-flight promises for range fetches, keyed by file ETag.
+ * Each value is a thread-safe list of InFlightPromise objects.
+ */
+ private final ConcurrentHashMap<String,
CopyOnWriteArrayList<InFlightPromise>>
+ promiseRegistry = new ConcurrentHashMap<>();
+
+ /**
+ * InFlightPromise represents a pending fetch for a specific range.
+ * @param start the start offset of the range
+ * @param end the end offset of the range
+ * @param future the CompletableFuture representing the fetch operation
+ */
+ public record InFlightPromise(long start, long end,
+ CompletableFuture<Void> future) {}
+
+ /**
+ * Executor to asynchronously execute get blob layout calls
+ */
+ private static final ExecutorService FETCH_EXECUTOR = new ThreadPoolExecutor(
+ 8, 32, 60L, TimeUnit.SECONDS,
+ new LinkedBlockingQueue<>(1024), // Bounded queue to prevent OOM
+ new ThreadFactoryBuilder()
+ .setNameFormat("abfs-blob-layout-fetch-%d")
+ .setDaemon(true)
+ .build(),
+ // AbortPolicy keeps the executor non-blocking: when the pool and queue
are
+ // saturated, submission throws RejectedExecutionException instead of
running
+ // the fetch on the caller (read) thread. The submitter is responsible
for
+ // cleaning up the in-flight promise so a later read can retry the gap.
+ new ThreadPoolExecutor.AbortPolicy());
+
+ /**
+ * Optional override for the fetch executor, used only by tests to inject a
+ * deterministic executor (e.g. one that always rejects). {@code null} means
the
+ * shared {@link #FETCH_EXECUTOR} is used.
+ */
+ private static volatile ExecutorService fetchExecutorOverride = null;
+
+ static {
+ // Add fetch executor to JVM shutdown hook to ensure orderly shutdown of
thread during JVM shutdown.
+ Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+ FETCH_EXECUTOR.shutdown();
+ try {
+ if (!FETCH_EXECUTOR.awaitTermination(5, TimeUnit.SECONDS)) {
+ FETCH_EXECUTOR.shutdownNow();
+ }
+ } catch (InterruptedException e) {
+ FETCH_EXECUTOR.shutdownNow();
+ }
+ }));
+ }
+
+ /**
+ * Private constructor for singleton pattern.
+ * @param evictionTime the idle timeout in minutes for cache eviction
+ * @param maxCacheWeight the maximum total number of cached ranges before
eviction
+ * @param ticker the time source used for idle-eviction bookkeeping
+ */
+ private BlobLayoutCache(long evictionTime, long maxCacheWeight, Ticker
ticker) {
+ this.idleTimeoutMs = TimeUnit.MINUTES.toMillis(evictionTime);
+ this.maxCacheWeight = maxCacheWeight;
+ this.ticker = ticker;
+ this.cache = Caffeine.newBuilder()
+ .maximumWeight(maxCacheWeight)
+ .ticker(ticker)
+ // Weight is determined by the number of cached ranges in the layout
+ .weigher(
+ (String key, LayoutEntry entry) -> entry.layout.getRangeMapSize())
+ .expireAfter(new Expiry<String, LayoutEntry>() {
+ /**
+ * Calculates the expiry time for a cache entry based on active
streams and idle time.
+ * @param value the LayoutEntry
+ * @param currentTime the current time in nanoseconds
+ * @return the time in nanoseconds until expiry
+ */
+ private long calculateExpiry(LayoutEntry value, long currentTime) {
+ long lastDeregistered = value.lastDeregisteredTimeInNanos.get();
+ // Do not expire while a stream is active, or before the first
+ // deregistration has happened (sentinel Long.MAX_VALUE). The
latter
+ // check also avoids long overflow in the subtraction below.
+ if (value.activeStreams.get() > 0
+ || lastDeregistered == Long.MAX_VALUE) {
+ return Long.MAX_VALUE;
+ }
+
+ // How long has it been since the last stream closed?
+ long elapsedNanos = currentTime - lastDeregistered;
+ long thresholdNanos = TimeUnit.MILLISECONDS.toNanos(
+ idleTimeoutMs);
+
+ // If we are already past the threshold, expire immediately
(return 0)
+ // Otherwise, return the remaining time until the threshold is hit
+ return Math.max(0, thresholdNanos - elapsedNanos);
+ }
+
+ @Override
+ public long expireAfterCreate(@NonNull String key,
+ @NonNull LayoutEntry value, long currentTime) {
+ return calculateExpiry(value, currentTime);
+ }
+
+ @Override
+ public long expireAfterUpdate(@NonNull String key,
+ @NonNull LayoutEntry value, long currentTime,
+ long currentDuration) {
+ return calculateExpiry(value, currentTime);
+ }
+
+ @Override
+ public long expireAfterRead(@NonNull String key,
+ @NonNull LayoutEntry value, long currentTime,
+ long currentDuration) {
+ // LEASE LOGIC: If streams are active, the entry never expires.
+ // Once activeStreams == 0, the 5-minute idle clock starts.
+ return calculateExpiry(value, currentTime);
+ }
+ })
+ .removalListener((key, value, cause) -> {
+ // Cleanup corresponding promises when an entry is evicted from cache
+ if (key != null) {
+ promiseRegistry.remove(key);
+ }
+ })
+ .build();
+ }
+
+ /**
+ * Returns the process-wide singleton instance of BlobLayoutCache, creating
it
+ * if necessary.
+ * <p>
+ * The cache is intentionally shared across all filesystem instances in the
JVM
+ * so that its total memory footprint stays bounded by a single weight
budget.
+ * As a consequence, the {@code evictionTime} and {@code maxCacheWeight}
supplied
+ * by the <em>first</em> caller win; values from later callers are ignored.
When a
+ * later caller requests a different configuration, a one-time warning is
logged
+ * rather than silently discarding it.
+ * </p>
+ * @param evictionTime the idle timeout in minutes for cache eviction
+ * @param maxCacheWeight the maximum total number of cached ranges before
eviction
+ * @return the singleton BlobLayoutCache instance
+ */
+ public static BlobLayoutCache getInstance(long evictionTime,
+ long maxCacheWeight) {
+ BlobLayoutCache local = instance;
+ if (local == null) {
+ synchronized (BlobLayoutCache.class) {
+ local = instance;
+ if (local == null) {
+ local = new BlobLayoutCache(evictionTime, maxCacheWeight,
+ Ticker.systemTicker());
+ instance = local;
+ }
+ }
+ }
+ warnIfConfigDiffers(local, evictionTime, maxCacheWeight);
+ return local;
+ }
+
+ /**
+ * Logs a one-time warning if a caller requests a configuration that differs
+ * from the one the singleton was initialized with.
+ */
+ private static void warnIfConfigDiffers(BlobLayoutCache existing,
+ long evictionTime, long maxCacheWeight) {
+ if (configMismatchWarned) {
+ return;
+ }
+ long requestedIdleMs = TimeUnit.MINUTES.toMillis(evictionTime);
+ if (existing.idleTimeoutMs != requestedIdleMs
+ || existing.maxCacheWeight != maxCacheWeight) {
+ configMismatchWarned = true;
+ LOG.warn("BlobLayoutCache is process-wide and already initialized with "
+ + "evictionMs={}, maxWeight={}; ignoring requested
evictionMs={}, "
+ + "maxWeight={}.", existing.idleTimeoutMs,
existing.maxCacheWeight,
+ requestedIdleMs, maxCacheWeight);
+ }
+ }
+
+ /**
+ * Creates an isolated (non-singleton) cache instance for deterministic
testing,
+ * allowing a custom {@link Ticker} so idle-based eviction can be exercised
+ * without real time passing.
+ * @param evictionTime the idle timeout in minutes for cache eviction
+ * @param maxCacheWeight the maximum total number of cached ranges before
eviction
+ * @param ticker the time source to use
+ * @return a fresh, isolated BlobLayoutCache instance
+ */
+ @VisibleForTesting
+ static BlobLayoutCache newInstanceForTesting(long evictionTime,
+ long maxCacheWeight, Ticker ticker) {
+ return new BlobLayoutCache(evictionTime, maxCacheWeight, ticker);
+ }
+
+ /**
+ * Forces any pending cache maintenance (eviction/expiration) to run. For
tests.
+ */
+ @VisibleForTesting
+ void cleanUp() {
+ cache.cleanUp();
+ }
+
+ /**
+ * Returns true if an entry for the given key is currently present (after
running
+ * pending maintenance). For tests.
+ * @param key the cache key (file ETag)
+ * @return true if the entry is still cached
+ */
+ @VisibleForTesting
+ boolean isCached(String key) {
+ cache.cleanUp();
+ return cache.getIfPresent(key) != null;
+ }
+
+ /**
+ * Returns the approximate number of entries in the cache (after running
pending
+ * maintenance). For tests.
+ * @return the estimated cache size
+ */
+ @VisibleForTesting
+ long estimatedSize() {
+ cache.cleanUp();
+ return cache.estimatedSize();
+ }
+
+ /**
+ * Returns the shared executor used to run asynchronous blob layout fetch
tasks.
+ * Package-private: intended for use by {@code AbfsInputStream} only.
+ * @return the shared fetch executor
+ */
+ static ExecutorService getFetchExecutor() {
+ ExecutorService override = fetchExecutorOverride;
+ return override != null ? override : FETCH_EXECUTOR;
+ }
+
+ /**
+ * Overrides the fetch executor for tests so rejection/non-blocking
behaviour can
+ * be exercised deterministically. Callers must invoke
+ * {@link #resetFetchExecutorForTesting()} afterwards.
+ * @param executor the executor to use, or {@code null} to reset
+ */
+ @VisibleForTesting
+ static void setFetchExecutorForTesting(ExecutorService executor) {
+ fetchExecutorOverride = executor;
+ }
+
+ /**
+ * Restores the default shared fetch executor after a test override. For
tests.
+ */
+ @VisibleForTesting
+ static void resetFetchExecutorForTesting() {
+ fetchExecutorOverride = null;
+ }
+
+ /**
+ * Returns the number of in-flight fetch promises currently registered for
the
+ * given file ETag. Used by tests to assert that promises are not leaked.
For tests.
+ * @param fileETag the file ETag key
+ * @return the number of in-flight promises, or 0 if none are registered
+ */
+ @VisibleForTesting
+ int getInFlightPromiseCount(String fileETag) {
+ CopyOnWriteArrayList<InFlightPromise> promises =
promiseRegistry.get(fileETag);
+ return promises == null ? 0 : promises.size();
+ }
+
+ /**
+ * Registers a stream for the given file ETag and content length.
+ * Increments the active stream count and blocks eviction.
+ * @param fileETag the file ETag
+ * @param contentLength the content length of the blob
+ */
+ public void registerStream(String fileETag, long contentLength) {
+ cache.asMap().compute(fileETag, (key, entry) -> {
+ if (entry == null) {
+ entry = new LayoutEntry(contentLength);
+ }
+ entry.activeStreams.incrementAndGet();
+ entry.lastDeregisteredTimeInNanos.set(
+ Long.MAX_VALUE); // Block eviction logic
+ return entry;
+ });
+ }
+
+ /**
+ * Deregisters a stream for the given file ETag.
+ * Decrements the active stream count and updates the last deregistered time
for eviction.
+ * @param fileETag the file ETag
+ */
+ public void deregisterStream(String fileETag) {
+ cache.asMap().computeIfPresent(fileETag, (key, entry) -> {
+ // Clamp at zero so unbalanced deregister calls cannot drive the counter
+ // negative and keep resetting the idle clock.
+ int remaining = entry.activeStreams.updateAndGet(v -> v > 0 ? v - 1 : 0);
+ if (remaining == 0) {
+ entry.lastDeregisteredTimeInNanos.set(ticker.read());
+ }
+ return entry;
+ });
+ }
+
+ /**
+ * Retrieves the cached blob layout ranges for the specified key and range.
+ * @param key the cache key (file ETag)
+ * @param start the start offset
+ * @param end the end offset
+ * @return a list of BlobRange objects, or null if not available
+ */
+ public List<BlobLayout.BlobRange> getBlobLayout(final String key,
+ final long start, final long end) {
+ LayoutEntry layoutEntry = cache.getIfPresent(key);
+ return (layoutEntry == null || layoutEntry.isLayoutUnavailable())
+ ? null : layoutEntry.layout.getRanges(start, end);
+ }
+
+ /**
+ * Retrieves the gaps (unfetched ranges) in the cached layout for the
specified key and range.
+ * @param key the cache key (file ETag)
+ * @param start the start offset
+ * @param end the end offset
+ * @return a list of BlobRange objects representing gaps, or null if not
available
+ */
+ public List<BlobLayout.BlobRange> getGaps(final String key,
+ final long start, final long end) {
+ LayoutEntry layoutEntry = cache.getIfPresent(key);
+ return layoutEntry == null || layoutEntry.isLayoutUnavailable()
Review Comment:
Confirmed: a missing entry must not be interpreted as a server-declared
unavailable layout. The local fix recreates missing entries before discovery
and logs the recreation at DEBUG; an existing unavailable entry is left
untouched and does not trigger a fetch. Weight-based eviction remains enabled
to retain the memory bound. Regression coverage:
testWeightEvictedLayoutIsFetchedAgain and
testUnavailableLayoutDoesNotRestartDiscovery.
These changes are prepared and verified locally (120 focused unit tests
passed), but are not yet committed or pushed to this PR. Leaving the thread
open until the updated patch is published.
##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/KeepAliveCache.java:
##########
@@ -61,216 +65,405 @@ class KeepAliveCache extends
LinkedBlockingDeque<HttpClientConnection>
KeepAliveCache.class);
/**
- * Flag to indicate if the cache is closed.
+ * Indicates whether the cache has been closed.
+ * Used to prevent further operations after closure.
+ */
+ private final AtomicBoolean closed = new AtomicBoolean(false);
+
+ /**
+ * Maximum number of connections allowed in the default host cache.
*/
- private final AtomicBoolean isClosed = new AtomicBoolean(false);
+ private final int maxDefaultConnections;
/**
- * Maximum number of connections that can be cached.
+ * Maximum number of connections allowed in the cluster host cache.
*/
- private final int maxCacheConnections;
+ private final int maxClusterConnections;
/**
- * Account name for which the cache is created. To be used only in exception
- * messages.
+ * The account name path associated with this cache instance.
*/
private final String accountNamePath;
/**
- * Executor server to trigger connection refresh from cache manager.
+ * Executor service for single-threaded cache refresh operations.
*/
- private ExecutorService singleThreadPool = null;
+ private ExecutorService singleThreadPool;
/**
- * Executor service to trigger async cache warmup.
+ * Executor service for fixed-threaded cache warmup and refresh operations.
+ */
+ private ExecutorService fixedThreadPool;
+
+ /**
+ * Represents a pooled HTTP connection with a usage flag and cache key.
+ * <p>
+ * Each instance wraps a {@link HttpClientConnection} and tracks whether it
is currently in use.
+ * The {@code cacheKey} identifies the host this connection is associated
with.
+ * The {@code inUse} flag is used to prevent race conditions during eviction
or retrieval.
*/
- private ExecutorService fixedThreadPool = null;
+ @VisibleForTesting
+ public static final class PooledConnection {
+
+ final HttpClientConnection conn;
+
+ final String cacheKey;
+
+ final AtomicBoolean inUse = new AtomicBoolean(false); // New Flag
+ PooledConnection(HttpClientConnection conn, String cacheKey) {
+ this.conn = conn;
+ this.cacheKey = cacheKey;
+ }
+ }
/**
- * Creates an {@link KeepAliveCache} instance using filesystem's
configuration.
+ * HostQueue is a thread-safe queue for storing pooled HTTP connections per
host.
* <p>
- * The size of the cache is determined by the configuration
- * {@value
org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys#FS_AZURE_APACHE_HTTP_CLIENT_MAX_CACHE_SIZE}.
- * If the configuration is not set, the default value is
- * {@value
org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations#DEFAULT_APACHE_HTTP_CLIENT_MAX_CACHE_SIZE}.
- * <p>.
+ * Uses a ConcurrentLinkedDeque to allow concurrent access and lock-free
operations.
+ * Each HostQueue instance manages the connections for a specific host in
the cache.
+ */
+ @VisibleForTesting
+ public static final class HostQueue {
+
+ // ConcurrentLinkedDeque is thread-safe; no manual lock needed
+ final ConcurrentLinkedDeque<PooledConnection> queue
+ = new ConcurrentLinkedDeque<>();
+ }
+
+ private final ConcurrentHashMap<String, HostQueue> hostCaches
+ = new ConcurrentHashMap<>();
+
+ private final ConcurrentLinkedQueue<PooledConnection> clusterQueue
+ = new ConcurrentLinkedQueue<>();
+
+ private final AtomicInteger defaultSize = new AtomicInteger(0);
+
+ private final AtomicInteger clusterSize = new AtomicInteger(0);
+
+ /**
+ * Checks if the cache is closed.
+ *
+ * @return true if the cache is closed, false otherwise
+ */
+ boolean isClosed() {
+ return closed.get();
+ }
+
+ /**
+ * Gets the current size of the default host cache.
+ *
+ * @return the number of connections cached for the default host
+ */
+ int getCachedDefaultSize() {
+ return defaultSize.get();
+ }
+
+ /**
+ * Constructs a KeepAliveCache instance with the specified configuration.
+ *
+ * @param abfsConfiguration the ABFS configuration
*/
KeepAliveCache(AbfsConfiguration abfsConfiguration) {
- this.accountNamePath =
- abfsConfiguration.getAccountName();
- this.maxCacheConnections =
- abfsConfiguration.getApacheMaxCacheSize();
- // Initialise singleThreadPool if cache refresh is enabled.
+ this.accountNamePath = abfsConfiguration.getAccountName();
+ this.maxDefaultConnections
+ = abfsConfiguration.getApacheMaxDefaultCacheSize();
+ this.maxClusterConnections
+ = abfsConfiguration.getApacheMaxNonDefaultCacheSize();
+
+ // Fix: Always use Daemon threads to prevent JVM hang on exit
if (abfsConfiguration.getApacheCacheRefreshCount() > 0) {
this.singleThreadPool = Executors.newSingleThreadExecutor(r -> {
- Thread thread = new Thread(r);
- thread.setName("CacheRefreshThread");
- thread.setDaemon(true);
- return thread;
+ Thread t = new Thread(r, "CacheRefreshThread");
+ t.setDaemon(true);
+ return t;
});
}
- // Initialise fixedThreadPool if cache warmup or cache refresh is enabled.
- if (abfsConfiguration.getApacheCacheWarmupCount() > 0
- || abfsConfiguration.getApacheCacheRefreshCount() > 0) {
- this.fixedThreadPool = Executors.newFixedThreadPool(Math.min(5,
- Math.max(abfsConfiguration.getApacheCacheWarmupCount(),
- abfsConfiguration.getApacheCacheRefreshCount())), r -> {
- Thread thread = new Thread(r);
- thread.setName("AsyncCacheConnectionThread");
- thread.setDaemon(true);
- return thread;
- });
+ int warmup = abfsConfiguration.getApacheCacheWarmupCount();
+ int refresh = abfsConfiguration.getApacheCacheRefreshCount();
+ if (warmup > 0 || refresh > 0) {
+ this.fixedThreadPool = Executors.newFixedThreadPool(
+ Math.min(5, Math.max(warmup, refresh)), r -> {
+ Thread t = new Thread(r, "AsyncCacheConnectionThread");
+ t.setDaemon(true);
+ return t;
+ });
}
}
/**
- * Safe close of the HttpClientConnection.
+ * Retrieves a pooled HTTP connection for the specified host.
+ * <p>
+ * If the cache is closed, throws ClosedIOException. If no connection is
available, returns null.
+ * Decrements the appropriate counter and closes stale connections.
*
- * @param hc HttpClientConnection to be closed
+ * @param host the host to retrieve a connection for
+ * @param isDefaultHost true if the host is the default host
+ * @return a valid HttpClientConnection or null if unavailable
+ * @throws IOException if the cache is closed
*/
- private void closeHttpClientConnection(final HttpClientConnection hc) {
- try {
- hc.close();
- } catch (IOException ex) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Close failed for connection: {}", hc, ex);
+ public HttpClientConnection get(String host, boolean isDefaultHost)
+ throws IOException {
+ if (closed.get()) {
+ throw new ClosedIOException(accountNamePath, KEEP_ALIVE_CACHE_CLOSED);
+ }
+ HostQueue hq = hostCaches.get(host);
+ if (hq == null) {
+ return null;
+ }
+
+ PooledConnection pooled;
+ // poll() is now an atomic, non-blocking operation
+ while ((pooled = hq.queue.poll()) != null) {
+ if (!pooled.inUse.compareAndSet(false, true)) {
+ continue; // Already claimed by eviction
+ }
+
+ if (!isDefaultHost) {
+ clusterSize.decrementAndGet();
+ } else {
+ defaultSize.decrementAndGet();
+ }
+
+ if (pooled.conn.isOpen() && !pooled.conn.isStale()) {
+ return pooled.conn;
}
+ closeQuietly(pooled.conn);
}
+ return null;
}
/**
- * Close all connections in cache.
+ * Adds a connection to the cache for the specified host type.
+ * <p>
+ * Only valid, open, non-stale connections are cached. Defensive casting
ensures only
+ * AbfsManagedApacheHttpConnection instances are accepted. Returns true if
the connection
+ * was successfully cached, false otherwise.
+ *
+ * @param conn the connection to cache
+ * @param isDefaultHost true if the host is the default host
+ * @return true if the connection was cached, false otherwise
*/
- @Override
- public void close() {
- boolean closed = isClosed.getAndSet(true);
- if (closed) {
- return;
+ public boolean put(HttpClientConnection conn, boolean isDefaultHost) {
+ if (conn == null || closed.get() || !conn.isOpen() || conn.isStale()) {
+ closeQuietly(conn);
+ return false;
}
- closeInternal();
- if (singleThreadPool != null && !singleThreadPool.isShutdown()) {
- singleThreadPool.shutdownNow();
+
+ // Defensive casting check
+ if (!(conn instanceof AbfsManagedApacheHttpConnection)) {
+ closeQuietly(conn);
+ return false;
}
- if (fixedThreadPool != null && !fixedThreadPool.isShutdown()) {
- fixedThreadPool.shutdownNow();
+ String host = ((AbfsManagedApacheHttpConnection) conn).getTargetHost()
+ .toHostString();
+ PooledConnection pooled = new PooledConnection(conn, host);
+ HostQueue hq = hostCaches.computeIfAbsent(host, k -> new HostQueue());
+
+ return isDefaultHost ? putDefault(hq, pooled) : putCluster(hq, pooled);
+ }
+
+ /**
+ * Adds a connection to the default host cache, evicting the oldest if the
cache exceeds its limit.
+ *
+ * @param hq the HostQueue for the default host
+ * @param pooled the pooled connection to add
+ * @return true if the connection was cached, false otherwise
+ */
+ private boolean putDefault(HostQueue hq, PooledConnection pooled) {
+ if (maxDefaultConnections <= 0) {
+ closeQuietly(pooled.conn);
+ return false;
}
+
+ hq.queue.offer(pooled);
+ if (defaultSize.incrementAndGet() > maxDefaultConnections) {
+ PooledConnection evicted = hq.queue.poll(); // Evict oldest
+ if (evicted != null) {
+ // Double-check 'inUse' to ensure we don't close a connection just
handed out
+ if (evicted.inUse.compareAndSet(false, true)) {
+ defaultSize.decrementAndGet();
+ closeQuietly(evicted.conn);
+ } else {
+ // If we couldn't claim it for eviction, it was just grabbed by get()
+ // defaultSize was already decremented by get(), so just move on.
+ }
+ }
+ }
+ return true;
}
/**
- * @return true if the cache is closed, false otherwise.
+ * Adds a connection to the cluster host cache, evicting connections if the
cache exceeds its limit.
+ *
+ * @param hq the HostQueue for the cluster host
+ * @param pooled the pooled connection to add
+ * @return true if the connection was cached, false otherwise
*/
- public boolean getIsClosed() {
- return isClosed.get();
+ private boolean putCluster(HostQueue hq, PooledConnection pooled) {
+ if (maxClusterConnections <= 0) {
+ closeQuietly(pooled.conn);
+ return false;
+ }
+
+ while (clusterSize.get() >= maxClusterConnections) {
+ PooledConnection candidate = clusterQueue.poll();
+ if (candidate == null) {break;}
+
+ if (candidate.inUse.compareAndSet(false, true)) {
+ clusterSize.decrementAndGet();
+ HostQueue evictHq = hostCaches.get(candidate.cacheKey);
+ if (evictHq != null) {
+ // remove() on ConcurrentLinkedDeque is thread-safe and lock-free
+ evictHq.queue.remove(candidate);
+ }
+ closeQuietly(candidate.conn);
+ }
+ }
+
+ hq.queue.offer(pooled);
+ clusterQueue.offer(pooled);
Review Comment:
Confirmed: normal reuse can leave unbounded stale wrappers in clusterQueue.
The local fix removes a claimed cluster wrapper from that queue during get(),
for both healthy and stale connections, and also cleans up if get() claims it
before putCluster() finishes publishing it.
testClusterReuseDoesNotRetainQueueEntries exercises 1,000 reuse cycles below
the eviction threshold and stale-connection cleanup, asserting both queue size
and connection count return to zero.
These changes are prepared and verified locally (120 focused unit tests
passed), but are not yet committed or pushed to this PR. Leaving the thread
open until the updated patch is published.
##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsInputStream.java:
##########
@@ -588,11 +673,294 @@ protected int readInternal(final long position, final
byte[] b, final int offset
// got nothing from read-ahead, do our own read now
TracingContext tc = new TracingContext(tracingContext);
tc.setReadType(ReadType.MISSEDCACHE_READ);
- receivedBytes = readRemote(position, b, offset, length, tc);
+
+ String endpoint = findEndpointForPosition(position, length);
+ receivedBytes = readRemote(position, b, offset, length, tc, endpoint);
return receivedBytes;
} else {
LOG.debug("read ahead disabled, reading remote");
- return readRemote(position, b, offset, length, new
TracingContext(tracingContext));
+ return readRemote(position, b, offset, length,
+ new TracingContext(tracingContext),
+ findEndpointForPosition(position, length));
+ }
+ }
+
+ /**
+ * Get blob layout for a file based on start and end position.
+ * In case the layout is present in multiple pages, this method will
internally
+ * fetch all the pages and construct the layout.
+ *
+ * @param start start position of the read
+ * @param end end position of the read
+ * @return blob layout for the file
+ * @throws IllegalStateException if the client is not an AbfsBlobClient
+ */
+ private BlobLayoutResponse getBlobLayout(final long start, final long end,
+ final TracingContext tracingContext) throws AzureBlobFileSystemException
{
+ BlobLayoutResponse fullLayout = new BlobLayoutResponse();
+ TracingContext context = new TracingContext(tracingContext);
+ context.setOperation(FSOperationType.GET_BLOB_LAYOUT);
+ String nextMarker = null;
+ do {
+ AbfsRestOperation op = ((AbfsBlobClient) client).getBlobLayout(path,
+ start, end, eTag, nextMarker, context);
+ try {
+ InputStream stream = op.getResult().getListResultStream();
+ stream.reset();
+
+ SAXParserFactory factory = SAXParserFactory.newInstance();
+ SAXParser parser = factory.newSAXParser();
+ BlobLayoutXmlParser handler = new BlobLayoutXmlParser();
+ parser.parse(stream, handler);
+
+ BlobLayoutResponse currPage = handler.getResponse();
+ fullLayout.addBlobLayoutResponse(currPage);
+ nextMarker = currPage.getNextMarker();
+ } catch (Exception ex) {
+ throw new AbfsRestOperationException(-1, "",
+ "Failed to parse blob layout response", ex);
+ }
+ }
+ while (!StringUtils.isEmpty(nextMarker));
+ return fullLayout;
+ }
+
+ /**
+ * Get Blob layout for the range. This method will first try to get the
layout
+ * from cache and if not present, it will call get blob layout API and update
+ * the cache and return the request data to the caller method. This method
fetch
+ * additional layout to save future calls to get blob layout API.
+ * @param start start position
+ * @param end end position
+ * @param tracingContext tracing context
+ * @return List of blob ranges for the requested range
+ */
+ private List<BlobLayout.BlobRange> getBlobRanges(long start, long end,
+ TracingContext tracingContext) {
+ if (!isDataLocalityCheckEnabled) {
+ return null;
+ }
+
+ TracingContext tracingContext1 = new TracingContext(tracingContext);
+ tracingContext1.setOperation(FSOperationType.GET_BLOB_LAYOUT);
+ List<BlobLayout.BlobRange> gaps = layoutCache.getGaps(eTag, start, end);
+ AbfsCounters abfsCounters = client.getAbfsCounters();
+
+ Set<CompletableFuture<Void>> dependencies = new HashSet<>();
+ if (gaps == null) {
+ // This case will come when data is not distributed across layouts.
+ // In this case, we need to use host URL to fetch the data instead of
+ // iterating through layouts.
+ if (abfsCounters != null) {
+ abfsCounters.incrementCounter(AbfsStatistic.LAYOUT_NOT_PRESENT, 1);
+ }
+ return null;
+ } else if (!gaps.isEmpty()) {
+ if (abfsCounters != null) {
+ abfsCounters.incrementCounter(AbfsStatistic.LAYOUT_CACHE_MISS, 1);
+ }
+ for (BlobLayout.BlobRange gap : gaps) {
+ // Determine the optimal range to fetch using bridge logic
+ BlobLayout.BlobRange bridge = layoutCache.getBridgeGap(
+ eTag, gap.start(), MAX_FETCH_LIMIT);
+
+ long fStart = (bridge != null) ? bridge.start() : gap.start();
+ long fEnd = (bridge != null) ? bridge.end() :
+ Math.min(contentLength, gap.start() + MAX_FETCH_LIMIT) - 1;
+
+ // Atomic operation: registers if absent, returns existing if present
+ dependencies.add(
+ registerAndFetch(start, end, fStart, fEnd, tracingContext1));
+ }
+ } else {
+ if (abfsCounters != null) {
+ abfsCounters.incrementCounter(AbfsStatistic.LAYOUT_CACHE_HIT, 1);
+ }
+ }
+
+ if (!dependencies.isEmpty()) {
+ try {
+ CompletableFuture.allOf(dependencies.toArray(new CompletableFuture[0]))
+ .get(client.getAbfsConfiguration()
+ .getBlobLayoutFetchTimeoutInMillis(), TimeUnit.MILLISECONDS);
+ } catch (Exception e) {
+ // Transient failure (timeout/exception) waiting on in-flight fetches.
Review Comment:
Confirmed. The local code now catches InterruptedException separately,
restores Thread.currentThread().interrupt(), and falls back without cancelling
another waiter's shared fetch. Timeout and execution-failure paths are
separate, and the old promise-removal claim is gone. Regression coverage:
testInterruptedWaiterDoesNotCancelSharedFetch,
testTimeoutPreservesLateResponseAndContinuationToken, and
testShortWaiterDoesNotCancelAnotherWaitersFetch.
These changes are prepared and verified locally (120 focused unit tests
passed), but are not yet committed or pushed to this PR. Leaving the thread
open until the updated patch is published.
##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsInputStream.java:
##########
@@ -588,11 +673,294 @@ protected int readInternal(final long position, final
byte[] b, final int offset
// got nothing from read-ahead, do our own read now
TracingContext tc = new TracingContext(tracingContext);
tc.setReadType(ReadType.MISSEDCACHE_READ);
- receivedBytes = readRemote(position, b, offset, length, tc);
+
+ String endpoint = findEndpointForPosition(position, length);
+ receivedBytes = readRemote(position, b, offset, length, tc, endpoint);
return receivedBytes;
} else {
LOG.debug("read ahead disabled, reading remote");
- return readRemote(position, b, offset, length, new
TracingContext(tracingContext));
+ return readRemote(position, b, offset, length,
+ new TracingContext(tracingContext),
+ findEndpointForPosition(position, length));
+ }
+ }
+
+ /**
+ * Get blob layout for a file based on start and end position.
+ * In case the layout is present in multiple pages, this method will
internally
+ * fetch all the pages and construct the layout.
+ *
+ * @param start start position of the read
+ * @param end end position of the read
+ * @return blob layout for the file
+ * @throws IllegalStateException if the client is not an AbfsBlobClient
+ */
+ private BlobLayoutResponse getBlobLayout(final long start, final long end,
+ final TracingContext tracingContext) throws AzureBlobFileSystemException
{
+ BlobLayoutResponse fullLayout = new BlobLayoutResponse();
+ TracingContext context = new TracingContext(tracingContext);
+ context.setOperation(FSOperationType.GET_BLOB_LAYOUT);
+ String nextMarker = null;
+ do {
+ AbfsRestOperation op = ((AbfsBlobClient) client).getBlobLayout(path,
+ start, end, eTag, nextMarker, context);
+ try {
+ InputStream stream = op.getResult().getListResultStream();
+ stream.reset();
+
+ SAXParserFactory factory = SAXParserFactory.newInstance();
+ SAXParser parser = factory.newSAXParser();
+ BlobLayoutXmlParser handler = new BlobLayoutXmlParser();
+ parser.parse(stream, handler);
+
+ BlobLayoutResponse currPage = handler.getResponse();
+ fullLayout.addBlobLayoutResponse(currPage);
+ nextMarker = currPage.getNextMarker();
+ } catch (Exception ex) {
+ throw new AbfsRestOperationException(-1, "",
+ "Failed to parse blob layout response", ex);
+ }
+ }
+ while (!StringUtils.isEmpty(nextMarker));
+ return fullLayout;
+ }
+
+ /**
+ * Get Blob layout for the range. This method will first try to get the
layout
+ * from cache and if not present, it will call get blob layout API and update
+ * the cache and return the request data to the caller method. This method
fetch
+ * additional layout to save future calls to get blob layout API.
+ * @param start start position
+ * @param end end position
+ * @param tracingContext tracing context
+ * @return List of blob ranges for the requested range
+ */
+ private List<BlobLayout.BlobRange> getBlobRanges(long start, long end,
+ TracingContext tracingContext) {
+ if (!isDataLocalityCheckEnabled) {
+ return null;
+ }
+
+ TracingContext tracingContext1 = new TracingContext(tracingContext);
+ tracingContext1.setOperation(FSOperationType.GET_BLOB_LAYOUT);
+ List<BlobLayout.BlobRange> gaps = layoutCache.getGaps(eTag, start, end);
+ AbfsCounters abfsCounters = client.getAbfsCounters();
+
+ Set<CompletableFuture<Void>> dependencies = new HashSet<>();
+ if (gaps == null) {
+ // This case will come when data is not distributed across layouts.
+ // In this case, we need to use host URL to fetch the data instead of
+ // iterating through layouts.
+ if (abfsCounters != null) {
+ abfsCounters.incrementCounter(AbfsStatistic.LAYOUT_NOT_PRESENT, 1);
+ }
+ return null;
+ } else if (!gaps.isEmpty()) {
+ if (abfsCounters != null) {
+ abfsCounters.incrementCounter(AbfsStatistic.LAYOUT_CACHE_MISS, 1);
+ }
+ for (BlobLayout.BlobRange gap : gaps) {
+ // Determine the optimal range to fetch using bridge logic
+ BlobLayout.BlobRange bridge = layoutCache.getBridgeGap(
+ eTag, gap.start(), MAX_FETCH_LIMIT);
+
+ long fStart = (bridge != null) ? bridge.start() : gap.start();
+ long fEnd = (bridge != null) ? bridge.end() :
+ Math.min(contentLength, gap.start() + MAX_FETCH_LIMIT) - 1;
+
+ // Atomic operation: registers if absent, returns existing if present
+ dependencies.add(
+ registerAndFetch(start, end, fStart, fEnd, tracingContext1));
+ }
+ } else {
+ if (abfsCounters != null) {
+ abfsCounters.incrementCounter(AbfsStatistic.LAYOUT_CACHE_HIT, 1);
+ }
+ }
+
+ if (!dependencies.isEmpty()) {
+ try {
+ CompletableFuture.allOf(dependencies.toArray(new CompletableFuture[0]))
Review Comment:
Confirmed. The local fix reduces the default lookup budget from 10 seconds
to 100 milliseconds and adds a one-second per-ETag discovery cooldown after a
failure, executor rejection, or waiter timeout. Cached coverage remains usable,
and late responses are preserved. The local read path also resolves only the
foreground starting byte; speculative read-ahead windows use cached locality
without extra layout waits. Regression coverage verifies cooldown across
enumerations, no repeat sharing/wait during cooldown, retry after cooldown, and
late-response recovery. The timeout and cooldown behavior are documented.
These changes are prepared and verified locally (120 focused unit tests
passed), but are not yet committed or pushed to this PR. Leaving the thread
open until the updated patch is published.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
