use iterator + rejected-ex handler

Author: sumangala17

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AbfsConfiguration.java

@@ -992,7 +992,7 @@ public ExponentialRetryPolicy getOauthTokenFetchRetryPolicy() {
 
   public int getWriteMaxConcurrentRequestCount() {
     if (this.writeMaxConcurrentRequestCount < 1) {
-      return 4 * Runtime.getRuntime().availableProcessors();
+      return 4 * getAvailableProcessorCount();
     }
     return this.writeMaxConcurrentRequestCount;
   }
@@ -1013,6 +1013,10 @@ public String getClientProvidedEncryptionKey() {
     return rawConfig.get(accSpecEncKey, null);
   }
 
+  public static int getAvailableProcessorCount() {
+    return Runtime.getRuntime().availableProcessors();
+  }
+
   @VisibleForTesting
   void setReadBufferSize(int bufferSize) {
     this.readBufferSize = bufferSize;

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AzureBlobFileSystem.java

@@ -1169,9 +1169,8 @@ public RemoteIterator<FileStatus> listStatusIterator(Path path)
     if (abfsStore.getAbfsConfiguration().enableAbfsListIterator()) {
       TracingContext tracingContext = new TracingContext(clientCorrelationId,
           fileSystemId, FSOperationType.LISTSTATUS, true, tracingHeaderFormat, listener);
-      AbfsListStatusRemoteIterator abfsLsItr =
-          new AbfsListStatusRemoteIterator(getFileStatus(path, tracingContext), abfsStore,
-              tracingContext);
+      AbfsListStatusRemoteIterator abfsLsItr = new AbfsListStatusRemoteIterator(
+          path, abfsStore, tracingContext);
       return RemoteIterators.typeCastingRemoteIterator(abfsLsItr);
     } else {
       return super.listStatusIterator(path);

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AzureBlobFileSystemStore.java

@@ -53,6 +53,11 @@
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.hadoop.thirdparty.com.google.common.annotations.VisibleForTesting;
 import org.apache.hadoop.thirdparty.com.google.common.base.Preconditions;
@@ -168,7 +173,7 @@ public class AzureBlobFileSystemStore implements Closeable, ListingSupport {
   private final IdentityTransformerInterface identityTransformer;
   private final AbfsPerfTracker abfsPerfTracker;
   private final AbfsCounters abfsCounters;
-  private final ExecutorService contentSummaryExecutorService = Executors.newCachedThreadPool();
+  private final ThreadPoolExecutor contentSummaryExecutorService;
 
   /**
    * The set of directories where we should store files as append blobs.
@@ -239,6 +244,17 @@ public AzureBlobFileSystemStore(URI uri, boolean isSecureScheme,
       this.appendBlobDirSet = new HashSet<>(Arrays.asList(
           abfsConfiguration.getAppendBlobDirs().split(AbfsHttpConstants.COMMA)));
     }
+    contentSummaryExecutorService = new ThreadPoolExecutor(0,
+        4 * AbfsConfiguration.getAvailableProcessorCount(), 60,
+        TimeUnit.SECONDS, new SynchronousQueue<>());
+    contentSummaryExecutorService.setRejectedExecutionHandler(
+        (runnable, threadPoolExecutor) -> {
+          try {
+            contentSummaryExecutorService.getQueue().put(runnable);
+          } catch (InterruptedException e) {
+            LOG.debug("Could not submit GetContentSummary task to thread pool");
+          }
+        });
   }
 
   /**

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsListStatusRemoteIterator.java

@@ -33,6 +33,7 @@
 import org.slf4j.LoggerFactory;
 
 import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.RemoteIterator;
 import org.apache.hadoop.fs.azurebfs.utils.TracingContext;
 
@@ -46,7 +47,7 @@ public class AbfsListStatusRemoteIterator
   private static final int MAX_QUEUE_SIZE = 10;
   private static final long POLL_WAIT_TIME_IN_MS = 250;
 
-  private final FileStatus fileStatus;
+  private final Path path;
   private final ListingSupport listingSupport;
   private final ArrayBlockingQueue<Object> iteratorsQueue;
   private final TracingContext tracingContext;
@@ -56,9 +57,9 @@ public class AbfsListStatusRemoteIterator
   private String continuation;
   private Iterator<FileStatus> currIterator;
 
-  public AbfsListStatusRemoteIterator(final FileStatus fileStatus,
+  public AbfsListStatusRemoteIterator(final Path path,
       final ListingSupport listingSupport, TracingContext tracingContext) {
-    this.fileStatus = fileStatus;
+    this.path = path;
     this.listingSupport = listingSupport;
     this.tracingContext = tracingContext;
     iteratorsQueue = new ArrayBlockingQueue<>(MAX_QUEUE_SIZE);
@@ -147,7 +148,7 @@ private void addNextBatchIteratorToQueue()
       throws IOException, InterruptedException {
     List<FileStatus> fileStatuses = new ArrayList<>();
     continuation = listingSupport
-        .listStatus(fileStatus.getPath(), null, fileStatuses, FETCH_ALL_FALSE,
+        .listStatus(path, null, fileStatuses, FETCH_ALL_FALSE,
             continuation, tracingContext);
     if (!fileStatuses.isEmpty()) {
       iteratorsQueue.put(fileStatuses.iterator());

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/ContentSummaryProcessor.java

@@ -95,24 +95,25 @@ public ContentSummary getContentSummary(Path path,
   }
 
   /**
-   * Calls listStatus on given path and populated fileStatus queue with
+   * Calls listStatus on given fileStatus and populated fileStatus queue with
    * subdirectories. Is called by new tasks to process the complete subtree
-   * under a given path
-   * @param path: Path to a file or directory
+   * under a given fileStatus
+   * @param fileStatus : Path to a file or directory
    * @throws IOException: listStatus error
    * @throws InterruptedException: error while inserting into queue
    */
-  private void processDirectoryTree(Path path, TracingContext tracingContext)
-      throws IOException, InterruptedException {
-    FileStatus[] fileStatuses = abfsStore.listStatus(path, tracingContext);
-
-    for (FileStatus fileStatus : fileStatuses) {
-      if (fileStatus.isDirectory()) {
-        queue.put(fileStatus);
+  private void processDirectoryTree(Path fileStatus,
+      TracingContext tracingContext) throws IOException, InterruptedException {
+    AbfsListStatusRemoteIterator iterator = new AbfsListStatusRemoteIterator(
+        fileStatus, abfsStore, tracingContext);
+    while (iterator.hasNext()) {
+      FileStatus status = iterator.next();
+      if (status.isDirectory()) {
+        queue.put(status);
         processDirectory();
         conditionalSubmitTaskToExecutor(tracingContext);
       } else {
-        processFile(fileStatus);
+        processFile(status);
       }
     }
   }
@@ -141,7 +142,8 @@ private synchronized void conditionalSubmitTaskToExecutor(TracingContext tracing
         FileStatus fileStatus1;
         while ((fileStatus1 = queue.poll(POLL_TIMEOUT, TimeUnit.MILLISECONDS))
                 != null) {
-          processDirectoryTree(fileStatus1.getPath(), new TracingContext(tracingContext));
+          processDirectoryTree(fileStatus1.getPath(),
+              new TracingContext(tracingContext));
         }
         return null;
       });

hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAbfsListStatusRemoteIterator.java

@@ -64,9 +64,8 @@ public void testAbfsIteratorWithHasNext() throws Exception {
         testDir, "testListPath");
 
     ListingSupport listngSupport = Mockito.spy(getFileSystem().getAbfsStore());
-    RemoteIterator<FileStatus> fsItr = new AbfsListStatusRemoteIterator(
-        getFileSystem().getFileStatus(testDir), listngSupport,
-        getTestTracingContext(getFileSystem(), true));
+    RemoteIterator<FileStatus> fsItr = new AbfsListStatusRemoteIterator(testDir,
+        listngSupport, getTestTracingContext(getFileSystem(), true));
     Assertions.assertThat(fsItr)
         .describedAs("RemoteIterator should be instance of "
             + "AbfsListStatusRemoteIterator by default")
@@ -101,12 +100,10 @@ public void testAbfsIteratorWithoutHasNext() throws Exception {
     final List<String> fileNames = createFilesUnderDirectory(TEST_FILES_NUMBER,
         testDir, "testListPath");
 
-    ListingSupport listngSupport = Mockito.spy(getFileSystem().getAbfsStore());
-    RemoteIterator<FileStatus> fsItr = new AbfsListStatusRemoteIterator(
-        getFileSystem().getFileStatus(testDir), listngSupport,
-        getTestTracingContext(getFileSystem(), true));
-    Assertions.assertThat(fsItr)
-        .describedAs("RemoteIterator should be instance of "
+    ListingSupport listingSupport = Mockito.spy(getFileSystem().getAbfsStore());
+    RemoteIterator<FileStatus> fsItr = new AbfsListStatusRemoteIterator(testDir,
+        listingSupport, getTestTracingContext(getFileSystem(), true));
+    Assertions.assertThat(fsItr).describedAs("RemoteIterator should be instance of "
             + "AbfsListStatusRemoteIterator by default")
         .isInstanceOf(AbfsListStatusRemoteIterator.class);
     int itrCount = 0;
@@ -130,7 +127,7 @@ public void testAbfsIteratorWithoutHasNext() throws Exception {
             + "there should be no more elements in the fileNames")
         .isEqualTo(0);
     int minNumberOfInvokations = TEST_FILES_NUMBER / 10;
-    verify(listngSupport, Mockito.atLeast(minNumberOfInvokations))
+    verify(listingSupport, Mockito.atLeast(minNumberOfInvokations))
         .listStatus(any(Path.class), nullable(String.class),
             anyList(), anyBoolean(),
             nullable(String.class),
@@ -209,7 +206,7 @@ public void testNextWhenNoMoreElementsPresent() throws Exception {
     Path testDir = createTestDirectory();
     setPageSize(10);
     RemoteIterator fsItr =
-        new AbfsListStatusRemoteIterator(getFileSystem().getFileStatus(testDir),
+        new AbfsListStatusRemoteIterator(testDir,
             getFileSystem().getAbfsStore(),
             getTestTracingContext(getFileSystem(), true));
     fsItr = Mockito.spy(fsItr);
@@ -257,9 +254,8 @@ public void testIOException() throws Exception {
 
     String exceptionMessage = "test exception";
     ListingSupport lsSupport =getMockListingSupport(exceptionMessage);
-    RemoteIterator fsItr =
-        new AbfsListStatusRemoteIterator(getFileSystem().getFileStatus(testDir),
-        lsSupport, getTestTracingContext(getFileSystem(), true));
+    RemoteIterator<FileStatus> fsItr = new AbfsListStatusRemoteIterator(testDir, lsSupport,
+        getTestTracingContext(getFileSystem(), true));
 
     Assertions.assertThatThrownBy(() -> fsItr.next())
         .describedAs(
@@ -270,7 +266,7 @@ public void testIOException() throws Exception {
   }
 
   @Test
-  public void testNonExistingPath() throws Throwable {
+  public void testNonExistingPath() {
     Path nonExistingDir = new Path("nonExistingPath");
     Assertions.assertThatThrownBy(
         () -> getFileSystem().listStatusIterator(nonExistingDir)).describedAs(

hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/ITestGetContentSummary.java

@@ -51,9 +51,9 @@ public class ITestGetContentSummary extends AbstractAbfsIntegrationTest {
   private static final int TEST_BUFFER_SIZE = 20;
   private static final int FILES_PER_DIRECTORY = 2;
   private static final int MAX_THREADS = 16;
-  private static final int NUM_FILES_FOR_LIST_MAX_TEST =10;
-//      DEFAULT_AZURE_LIST_MAX_RESULTS + 10;
-private static final int NUM_CONCURRENT_CALLS = 8;
+  private static final int NUM_FILES_FOR_LIST_MAX_TEST =
+      DEFAULT_AZURE_LIST_MAX_RESULTS + 10;
+  private static final int NUM_CONCURRENT_CALLS = 8;
 
   private final String[] directories = {"/testFolder",
       "/testFolderII",