[
https://issues.apache.org/jira/browse/HADOOP-19946?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18141856#comment-18141856
]
ASF GitHub Bot commented on HADOOP-19946:
-----------------------------------------
anmolanmol1234 commented on code in PR #8830:
URL: https://github.com/apache/hadoop/pull/8830#discussion_r4522700429
##########
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:
catch (Exception) also catches InterruptedException from future.get(), and
the interrupt flag is not restored. A task kill or cancel on this thread is
lost, and the next blocking call may proceed as if nothing happened. Please
catch InterruptedException separately and call
Thread.currentThread().interrupt(). Timeouts and ExecutionExceptions can stay
on the fall-back-and-continue path. The comment also says the failed promises
are already removed. On a timeout the promise is still in flight, so the
comment is inaccurate for that case.
> ABFS: Enable data locality aware reads for Azure Blob Storage
> -------------------------------------------------------------
>
> Key: HADOOP-19946
> URL: https://issues.apache.org/jira/browse/HADOOP-19946
> Project: Hadoop Common
> Issue Type: New Feature
> Components: fs/azure
> Reporter: Manish Bhatt
> Assignee: Manish Bhatt
> Priority: Major
> Labels: pull-request-available
>
> Introduce data locality support in ABFS to improve read performance for Azure
> Blob Storage by routing read requests to the endpoint that hosts the required
> data range. The goal is to reduce latency, improve throughput, and make read
> operations more efficient for large files and distributed workloads. This
> work adds blob layout discovery, range-to-endpoint mapping, locality-aware
> range selection, and caching of layout metadata to avoid repeated lookups. It
> also enables segmented read-ahead aligned with locality boundaries and adds
> safeguards for concurrency, eviction, timeout handling, and failure
> propagation needed for reliable production-scale reads.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
