review comments part 2: move executor->abfsStore

Author: sumangala17

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

@@ -127,9 +127,6 @@ public class AzureBlobFileSystem extends FileSystem
   private String clientCorrelationId;
   private TracingHeaderFormat tracingHeaderFormat;
   private Listener listener;
-  private final ExecutorService contentSummaryExecutorService = Executors.newCachedThreadPool();
-//  private final ExecutorService contentSummaryExecutorService = new ThreadPoolExecutor(
-//      0, 16, 5, TimeUnit.SECONDS, new SynchronousQueue<>());
 
   @Override
   public void initialize(URI uri, Configuration configuration)
@@ -453,8 +450,7 @@ public ContentSummary getContentSummary(Path path) throws IOException {
       TracingContext tracingContext = new TracingContext(clientCorrelationId,
           fileSystemId, FSOperationType.GET_CONTENT_SUMMARY, true,
           tracingHeaderFormat, listener);
-      return (new ContentSummaryProcessor(abfsStore,
-          contentSummaryExecutorService)).getContentSummary(path,
+      return (new ContentSummaryProcessor(abfsStore)).getContentSummary(path,
           tracingContext);
     } catch (InterruptedException e) {
       LOG.debug("Thread interrupted");

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

@@ -51,6 +51,8 @@
 import java.util.Set;
 import java.util.WeakHashMap;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 
 import org.apache.hadoop.thirdparty.com.google.common.annotations.VisibleForTesting;
 import org.apache.hadoop.thirdparty.com.google.common.base.Preconditions;
@@ -166,6 +168,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();
 
   /**
    * The set of directories where we should store files as append blobs.
@@ -1669,6 +1672,10 @@ private AbfsPerfInfo startTracking(String callerName, String calleeName) {
     return new AbfsPerfInfo(abfsPerfTracker, callerName, calleeName);
   }
 
+  public ExecutorService getContentSummaryExecutorService() {
+    return contentSummaryExecutorService;
+  }
+
   private static class VersionedFileStatus extends FileStatus {
     private final String version;
 

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

@@ -29,14 +29,20 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 
-import org.apache.hadoop.fs.azurebfs.utils.TracingContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import org.apache.hadoop.fs.ContentSummary;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.azurebfs.AzureBlobFileSystemStore;
+import org.apache.hadoop.fs.azurebfs.utils.TracingContext;
 
+/**
+ * Class to carry out parallelized recursive listing on a given path to
+ * collect directory and file count/size information, as part of the
+ * implementation for the Filesystem method getContentSummary
+ */
 public class ContentSummaryProcessor {
   private static final int MAX_THREAD_COUNT = 16;
   private static final int POLL_TIMEOUT = 100;
@@ -56,33 +62,34 @@ public class ContentSummaryProcessor {
    * @param abfsStore Instance of AzureBlobFileSystemStore, used to make
    * listStatus calls to server
    */
-  public ContentSummaryProcessor(ListingSupport abfsStore,
-      ExecutorService executorService) {
+  public ContentSummaryProcessor(ListingSupport abfsStore) {
     this.abfsStore = abfsStore;
-    this.executorService = executorService;
+    this.executorService = ((AzureBlobFileSystemStore) abfsStore).getContentSummaryExecutorService();
     completionService = new ExecutorCompletionService<>(this.executorService);
   }
 
-  public ContentSummary getContentSummary(Path path, TracingContext tracingContext)
-          throws IOException, ExecutionException, InterruptedException {
-    try {
-      processDirectoryTree(path, tracingContext);
-      while (!queue.isEmpty() || numTasks.get() > 0) {
-        try {
-          completionService.take().get();
-        } finally {
-          numTasks.decrementAndGet();
-          LOG.debug("FileStatus queue size = {}, number of submitted unfinished tasks = {}, active thread count = {}",
-              queue.size(), numTasks, ((ThreadPoolExecutor) executorService).getActiveCount());
-        }
+  public ContentSummary getContentSummary(Path path,
+      TracingContext tracingContext)
+      throws IOException, ExecutionException, InterruptedException {
+
+    processDirectoryTree(path, tracingContext);
+    while (!queue.isEmpty() || numTasks.get() > 0) {
+      try {
+        completionService.take().get();
+      } finally {
+        numTasks.decrementAndGet();
+        LOG.debug(
+            "FileStatus queue size = {}, number of submitted unfinished tasks"
+                + " = {}, active thread count = {}",
+            queue.size(), numTasks,
+            ((ThreadPoolExecutor) executorService).getActiveCount());
       }
-    } finally {
-      executorService.shutdownNow();
-      LOG.debug("Executor shutdown");
     }
+
     LOG.debug("Processed content summary of subtree under given path");
-    ContentSummary.Builder builder = new ContentSummary.Builder()
-        .directoryCount(directoryCount.get()).fileCount(fileCount.get())
+    ContentSummary.Builder builder =
+        new ContentSummary.Builder().directoryCount(
+            directoryCount.get()).fileCount(fileCount.get())
         .length(totalBytes.get()).spaceConsumed(totalBytes.get());
     return builder.build();
   }

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

@@ -208,7 +208,7 @@ private void populateDirWithFiles(Path directory, int numFiles)
     final List<Future<Void>> tasks = new ArrayList<>();
     ExecutorService es = Executors.newFixedThreadPool(10);
     for (int i = 0; i < numFiles; i++) {
-      final Path fileName = new Path(directory + "/test" + i);
+      final Path fileName = new Path(directory, String.format("test-%02d", i));
       tasks.add(es.submit(() -> {
         touch(fileName);
         return null;
@@ -218,7 +218,7 @@ private void populateDirWithFiles(Path directory, int numFiles)
       task.get();
     }
     try (FSDataOutputStream out = getFileSystem().append(
-        new Path(directory + "/test0"))) {
+        new Path(directory + "/test-00"))) {
       out.write(b);
     }
     es.shutdownNow();