clean up

Author: sumangala17

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

@@ -23,9 +23,7 @@
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorCompletionService;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.ForkJoinPool;
 import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.RecursiveAction;
 import java.util.concurrent.SynchronousQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
@@ -69,7 +67,6 @@ public ContentSummaryProcessor(ListingSupport abfsStore) {
 
   public ContentSummary getContentSummary(Path path)
           throws IOException, ExecutionException, InterruptedException {
-//    return gcs_fjp(path);
     try {
       processDirectoryTree(path);
       while (!queue.isEmpty() || numTasks.get() > 0) {
@@ -92,46 +89,6 @@ public ContentSummary getContentSummary(Path path)
     return builder.build();
   }
 
-  class GCS extends RecursiveAction {
-    Path path;
-    GCS (Path path) {
-      this.path = path;
-    }
-    @Override
-    protected void compute() {
-      try {
-        for (FileStatus fileStatus1 : abfsStore.listStatus(path)) {
-          if (fileStatus1.isDirectory()) {
-            processDirectory();
-            new GCS(fileStatus1.getPath()).fork();
-          } else {
-            processFile(fileStatus1);
-          }
-        }
-      } catch (IOException e) {
-        e.printStackTrace();
-      }
-    }
-  }
-
-  public ContentSummary gcs_fjp (Path path)
-      throws InterruptedException, IOException {
-    ForkJoinPool forkJoinPool = new ForkJoinPool();
-    for (FileStatus fileStatus : abfsStore.listStatus(path)) {
-      if (fileStatus.isDirectory()) {
-        processDirectory();
-        forkJoinPool.invoke(new GCS(fileStatus.getPath()));
-      } else {
-        processFile(fileStatus);
-      }
-    }
-    forkJoinPool.awaitTermination(1, TimeUnit.SECONDS);
-    ContentSummary.Builder builder = new ContentSummary.Builder()
-        .directoryCount(directoryCount.get()).fileCount(fileCount.get())
-        .length(totalBytes.get()).spaceConsumed(totalBytes.get());
-    return builder.build();
-  }
-
   /**
    * Calls listStatus on given path and populated fileStatus queue with
    * subdirectories. Is called by new tasks to process the complete subtree

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

@@ -22,7 +22,9 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Random;
+import java.util.concurrent.CompletionService;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorCompletionService;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
@@ -47,8 +49,8 @@ public class TestGetContentSummary 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_FILES_FOR_LIST_MAX_TEST =
+      DEFAULT_AZURE_LIST_MAX_RESULTS + 10;
 private static final int NUM_CONCURRENT_CALLS = 8;
 
   private final String[] directories = {"/testFolder",
@@ -156,17 +158,18 @@ public void testInvalidPath() throws Exception {
   @Test
   public void testConcurrentGetContentSummaryCalls()
       throws InterruptedException, ExecutionException, IOException {
+    AzureBlobFileSystem fs = getFileSystem();
     ExecutorService executorService = new ThreadPoolExecutor(1, MAX_THREADS, 5,
         TimeUnit.SECONDS, new SynchronousQueue<>());
-    ArrayList<Future<ContentSummary>> futures = new ArrayList<>();
+    CompletionService<ContentSummary> completionService =
+        new ExecutorCompletionService<>(executorService);
     createDirectoryStructure();
     for (int i = 0; i < NUM_CONCURRENT_CALLS; i++) {
-      Future<ContentSummary> future = executorService.submit(
-          () -> getFileSystem().getContentSummary(new Path("/testFolder")));
-      futures.add(future);
+      completionService.submit(() -> fs.getContentSummary(new Path(
+          "/testFolder")));
     }
     for (int i = 0; i < NUM_CONCURRENT_CALLS; i++) {
-      ContentSummary contentSummary = futures.get(i).get();
+      ContentSummary contentSummary = completionService.take().get();
       verifyContentSummary(contentSummary, 7, 8 * FILES_PER_DIRECTORY,
           8 * TEST_BUFFER_SIZE);
     }