anmolanmol1234 commented on code in PR #8830:
URL: https://github.com/apache/hadoop/pull/8830#discussion_r4522796900
##########
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:
Data locality is an optimisation, but this blocks the calling thread for up
to FS_AZURE_BLOB_LAYOUT_FETCH_TIMEOUT_MILLIS, which defaults to 10 seconds. In
readInternal it is also called once per read-ahead window inside the while
loop. On failure or timeout nothing is remembered, because the promise is
removed and the gap is retried. With a slow or degraded layout call, every read
pays the full wait again. Please use a much shorter default wait, and add a
short per-ETag backoff or negative cache after a timeout or failure. That would
keep a layout problem from becoming a read-latency problem.
--
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]
