POC Code changes for Data Locality and Data View

[anujmodi2021]

Description of PR

POC implementation of Data Locality and Data View features for Azure Blob File System (ABFS). This includes:

• BlobLayoutCache: Singleton Caffeine-backed cache storing blob layout (range-to-endpoint mappings) keyed by file ETag. Supports weighted eviction, active-stream lease logic, and in-flight promise deduplication to avoid redundant network calls. Houses a shared static fetchExecutor (with JVM shutdown hook) used by all streams. Promise registry access encapsulated via processInFlightPromises(eTag, action) which atomically applies caller logic inside a ConcurrentHashMap.compute call.
• BlobLayout: Lock-free range management using ConcurrentSkipListMap. Tracks which byte ranges are cached per blob, merges overlapping ranges, and computes gaps for efficient fetching.
• BlobLayoutXmlParser / BlobLayoutResponse: SAX-based XML parser and response model for the GetBlobLayout API (Azure Blob endpoint ?comp=layout). addBlobLayoutResponse correctly accumulates ranges and endpoints across paginated responses.
• AbfsInputStream: Extended to support data-locality-aware reads. When fs.azure.enable.data.locality=true and the client is AbfsBlobClient (checked at construction time via isDataLocalityCheckEnabled), reads are routed to the endpoint hosting the relevant byte range. Async layout fetching with gap bridging via registerAndFetch / buildCompositeFuture with short-circuit exception propagation.
• AbfsBlobClient: Added getBlobLayout() method to call the new layout API.
• KeepAliveCache: Redesigned with separate default/cluster host pools, ConcurrentLinkedDeque-backed HostQueue, atomic eviction via inUse flag, and daemon thread pools to prevent JVM hang on exit.
• ReadBufferManagerV2: Extended queueReadAhead to accept List<BlobLayout.BlobRange> segments for per-range child-buffer queuing, enabling data-locality-aware prefetch. Removed duplicate LOCK field that shadowed the parent class.
• AbfsConfiguration: Added configuration keys for data locality (fs.azure.enable.data.locality), blob layout cache eviction timeout, cache max weight, and configurable layout fetch timeout (fs.azure.blob.layout.fetch.timeout.millis, default 10s).

Bug fixes and improvements applied during code review:

• AbfsInputStream.close(): deregisterStream was called with path instead of eTag, preventing cache eviction.
• AbfsInputStream.findEndpointForPosition(): Added isEmpty() guard before blobRanges.get(0) to prevent IndexOutOfBoundsException.
• AbfsInputStream.getBlobLayout(): Fixed wrong TracingContext being mutated (caller's copy instead of the local copy).
• AbfsInputStream: isDataLocalityCheckEnabled now includes client instanceof AbfsBlobClient, ensuring the data locality path is only taken on Blob endpoint streams.
• BlobLayoutCache.putBlobLayout(): Replaced isLayoutPresent = new AtomicBoolean(false) with isLayoutPresent.set(false) to avoid a non-atomic field replacement data race. Removed dead MAX_CACHE_WEIGHT constant.
• BlobLayoutCache.fetchExecutor: Moved from a per-instance field in AbfsInputStream to a shared static in BlobLayoutCache with a JVM shutdown hook for orderly cleanup, preventing per-stream thread pool proliferation.
• BlobLayoutCache.processInFlightPromises(): Replaced direct promiseRegistry exposure with an atomic compute-based API; getFetchExecutor() made package-private.
• ReadBufferManagerV2.queueReadAhead(): Fixed off-by-one (> bufferPool.length → >= bufferPool.length) that could cause ArrayIndexOutOfBoundsException.
• BlobLayout.getMergedRangesInternal(): Replaced next.host().equals() with Objects.equals(next.host(), ...) to prevent NPE when host is null (valid for gap ranges).

How was this patch tested?

• Unit tests added/updated: BlobLayoutCacheTest (including tests for processInFlightPromises initialization, persistence, error propagation, concurrency, and async behavior), BlobLayoutTest, TestAbfsInputStream, TestApacheClientConnectionPool
• Integration tests updated: ITestAbfsClient, ITestAbfsInputStream (including tests for sub-interval coalescing, bridge gap splitting, partial failure short-circuit, and single-byte boundary), ITestAbfsInputStreamReadFooter, ITestAbfsInputStreamSmallFileReads, ITestReadBufferManagerV2

For code changes:

• Does the title or this PR starts with the corresponding JIRA issue id (e.g. 'HADOOP-17799. Your PR title ...')?
• Object storage: have the integration tests been executed and the endpoint declared according to the connector-specific documentation?
• If adding new dependencies to the code, are these dependencies licensed in a way that is compatible for inclusion under ASF 2.0?
• If applicable, have you updated the LICENSE, LICENSE-binary, NOTICE-binary files?

AI Tooling

If an AI tool was used:

• The PR includes the phrase "Contains content generated by GitHub Copilot"
• My use of AI contributions follows the ASF legal policy
  https://www.apache.org/legal/generative-tooling.html

Contains content generated by GitHub Copilot

[bhattmanish98]

@copilot Review the PR

[bhattmanish98]

@copilot review the PR

[Copilot]

@copilot review the PR

Here is a comprehensive review of the Data Locality and Data View POC changes (commits 9caa508 → d8577c1). I fixed several bugs in commit d796215.

---

✅ What's Good

• The overall architecture is sound: BlobLayoutCache (Caffeine-backed singleton with weighted eviction), BlobLayout (ConcurrentSkipListMap for lock-free range management), and the KeepAliveCache redesign are well-structured.
• Java 21 record types for BlobRange, Range, Endpoint, and InFlightPromise are idiomatic.
• Good use of CompletableFuture for async layout fetches with gap deduplication via promiseRegistry.
• The daemon-thread fix in KeepAliveCache (prevents JVM hang on exit) is correct.
• AtomicBoolean.compareAndSet used consistently in KeepAliveCache to prevent race conditions during eviction.

---

🐛 Bugs Fixed (commit d796215)

1. Wrong key in deregisterStream — AbfsInputStream.java:1158

// Before (wrong key - cache is keyed by eTag, not path):
layoutCache.deregisterStream(path);

// After (correct):
layoutCache.deregisterStream(eTag);

This prevented the active-stream count from ever being decremented, so cache entries would never expire.

2. IndexOutOfBoundsException in findEndpointForPosition — AbfsInputStream.java:565

// Before (no isEmpty check):
if (blobRanges == null) { return null; }
return blobRanges.get(0).host();

// After:
if (blobRanges == null || blobRanges.isEmpty()) { return null; }
return blobRanges.get(0).host();

3. Wrong TracingContext mutated in getBlobLayout — AbfsInputStream.java:679

// Before (modifies the caller's context copy instead of the local one):
TracingContext context = new TracingContext(tracingContext);
tracingContext.setOperation(FSOperationType.GET_BLOB_LAYOUT);  // wrong!

// After:
context.setOperation(FSOperationType.GET_BLOB_LAYOUT);

4. Non-atomic field replacement in putBlobLayout — BlobLayoutCache.java:335

// Before (replaces the AtomicBoolean reference — not thread-safe if other threads read the field):
layoutEntry.isLayoutPresent = new AtomicBoolean(false);

// After:
layoutEntry.isLayoutPresent.set(false);

5. Off-by-one in buffer index validation — ReadBufferManagerV2.java:318, 442

// Before (misses the case where index == length, causing ArrayIndexOutOfBoundsException):
if (bufferIndex > bufferPool.length)

// After:
if (bufferIndex >= bufferPool.length)

6. NPE when host is null in getMergedRangesInternal — BlobLayout.java:267

// Before (throws NPE when host is null, which is valid for gap ranges):
if (next.start() <= current.end() + 1 && next.host().equals(current.host()))

// After:
if (next.start() <= current.end() + 1 && Objects.equals(next.host(), current.host()))

---

⚠️ Design Issues (not fixed, consider addressing)

1. BlobLayoutCache singleton ignores reconfiguration — BlobLayoutCache.getInstance(evictionTime, maxCacheWeight) silently discards evictionTime/maxCacheWeight after the first call. If different filesystem instances use different settings, only the first one's values apply.

2. promiseRegistry is public final — Exposes internal mutable state. Should be package-private and accessed through controlled methods.

3. LOCK shadowing in ReadBufferManagerV2 — ReadBufferManagerV2 declares a private static final ReentrantLock LOCK that shadows protected static final ReentrantLock LOCK in parent ReadBufferManager. The two locks are completely independent objects, which can be confusing and could lead to bugs if someone calls base-class methods expecting them to synchronize on the same lock.

4. Static fetchExecutor is never shut down — AbfsInputStream.fetchExecutor is a static ThreadPoolExecutor shared by all instances. It is never shut down, relying entirely on daemon threads for cleanup. If the number of concurrent streams is high, the bounded queue (LinkedBlockingQueue(1024)) could fill up, triggering CallerRunsPolicy which would block read threads.

5. Hard 60s timeout on read path — CompletableFuture.allOf(...).get(60, TimeUnit.SECONDS) in getBlobRanges blocks the read thread for up to 60 seconds on layout fetch failures. Consider making this configurable.

6. addBlobLayoutResponse semantics — The method name implies accumulation but it replaces ranges/endpoints instead of merging them, which is misleading.

[bhattmanish98]

@copilot Please review

[Copilot]

@copilot Please review

The author addressed all 6 design issues from the previous review in commits 8df07ac5 and 92e388db. I fixed 3 remaining issues in commit 22e3eaa4:

---

✅ Issues Addressed by Author (8df07ac5, 92e388db)

1. fetchExecutor is no longer static — per-instance, created only when data locality is enabled, shut down in close()
2. promiseRegistry is now private — accessed via getPromiseRegistry()
3. LOCK shadowing removed — private static final ReentrantLock LOCK removed from ReadBufferManagerV2
4. Timeout is now configurable — fs.azure.blob.layout.fetch.timeout.millis, default 10s
5. addBlobLayoutResponse truly accumulates — ranges and endpoints are now appended, not replaced
6. Null eTag guard added — && eTag != null prevents NPE on streams without an eTag

---

🐛 Additional Fixes (22e3eaa4)

1. fetchExecutor.shutdown() → fetchExecutor.shutdownNow()

shutdown() waits for all queued layout-fetch network calls to complete before close() returns. shutdownNow() immediately interrupts them, which is the correct behavior when a stream is being closed.

2. Missing instanceof AbfsBlobClient guard in getBlobLayout

// Added guard before unsafe cast
if (!(client instanceof AbfsBlobClient)) {
    throw new IllegalStateException(
        "getBlobLayout called on a non-Blob endpoint client: "
            + client.getClass().getSimpleName());
}
AbfsRestOperation op = ((AbfsBlobClient) client).getBlobLayout(...);

Without this, if data locality is enabled on a DFS-endpoint client, the cast throws ClassCastException with no useful error message.

3. Removed dead MAX_CACHE_WEIGHT = 100_000 constant from BlobLayoutCache

This constant became unused after maxCacheWeight was made a constructor parameter.

---

⚠️ Remaining Design Concerns (not fixed)

1. Per-stream ThreadPoolExecutor with 8 core / 32 max threads — Each active stream creates its own pool. With many concurrent streams, this spawns hundreds of threads. Consider a shared pool scoped to the AzureBlobFileSystem or AbfsClient instance, or reduce the core size.
2. BlobLayoutCache singleton ignores reconfiguration — If different filesystem instances use different cache settings (eviction time, max weight), only the first one's values are used silently.
3. getPromiseRegistry() returns the mutable internal map — Still allows callers to manipulate the internal structure directly.

[bhattmanish98]

@copilot Please review

[Copilot]

@copilot Please review

The author addressed all remaining design concerns in commit 9c9df911. I fixed 3 minor issues in commit 2ef6d668.

---

✅ All Design Issues Now Addressed

1. Per-stream thread pool scalability — fetchExecutor moved to BlobLayoutCache as a shared static singleton. A JVM shutdown hook performs orderly shutdown (5s grace → shutdownNow()). No per-stream executor is created.

2. BlobLayoutCache singleton reconfiguration note — A comment was added: // evictionTime and maxCacheWeight will not be changed once INSTANCE is created. (by design; acceptable for a POC).

3. getPromiseRegistry() mutable access — Replaced with processInFlightPromises(eTag, action), which encapsulates all access to the registry inside a single ConcurrentHashMap.compute call. The internal promiseRegistry is no longer accessible from outside the class.

4. registerAndFetch simplification — The AtomicReference + raw compute pattern moved into processInFlightPromises, making registerAndFetch cleaner. buildCompositeFuture extracted as a helper with short-circuit exception propagation.

5. client instanceof AbfsBlobClient guard — Moved from getBlobLayout() to the constructor condition isDataLocalityCheckEnabled, which is the correct place (checked once at stream creation).

---

🐛 Minor Fixes (2ef6d668)

1. getFetchExecutor() made package-private — It was public static but is only used by AbfsInputStream in the same package. Removed public.
2. getFetchExecutor() and processInFlightPromises() Javadoc added — Both new methods were missing documentation.
3. Trailing newline added to BlobLayoutCache.java — The file was missing a final newline.

---

⚠️ One remaining note

processInFlightPromises still returns the mutable ConcurrentHashMap-backed CopyOnWriteArrayList to the action lambda. This is intentional (the action needs to add new InFlightPromise entries), but callers should be aware that mutations made to promiseList inside the action persist in the registry. This is correct by design and properly documented now.