Close Executor Service

Author: anujmodi2021

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

@@ -110,6 +110,9 @@
 import org.apache.hadoop.fs.azurebfs.services.AuthType;
 import org.apache.hadoop.fs.azurebfs.services.ExponentialRetryPolicy;
 import org.apache.hadoop.fs.azurebfs.services.ListingSupport;
+import org.apache.hadoop.fs.azurebfs.services.ReadBufferManager;
+import org.apache.hadoop.fs.azurebfs.services.ReadBufferManagerV1;
+import org.apache.hadoop.fs.azurebfs.services.ReadBufferManagerV2;
 import org.apache.hadoop.fs.azurebfs.services.SharedKeyCredentials;
 import org.apache.hadoop.fs.azurebfs.services.StaticRetryPolicy;
 import org.apache.hadoop.fs.azurebfs.services.VersionedFileStatus;
@@ -324,6 +327,10 @@ public void close() throws IOException {
       HadoopExecutors.shutdown(boundedThreadPool, LOG,
           30, TimeUnit.SECONDS);
       boundedThreadPool = null;
+      ReadBufferManagerV2 bufferManagerV2 = ReadBufferManagerV2.getBufferManager();
+      ReadBufferManagerV1 bufferManagerV1 = ReadBufferManagerV1.getBufferManager();
+      bufferManagerV1.close();
+      bufferManagerV2.close();
     } catch (InterruptedException e) {
       LOG.error("Interrupted freeing leases", e);
       Thread.currentThread().interrupt();

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

@@ -119,6 +119,12 @@ abstract void doneReading(ReadBuffer buffer,
    */
   abstract void purgeBuffersForStream(AbfsInputStream stream);
 
+  public void close() {
+    closeReadBufferManager();
+  }
+
+  abstract void closeReadBufferManager();
+
 
   // Following Methods are for testing purposes only and should not be used in production code.
 

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

@@ -33,7 +33,7 @@
  * The Read Buffer Manager for Rest AbfsClient.
  * V1 implementation of ReadBufferManager.
  */
-final class ReadBufferManagerV1 extends ReadBufferManager {
+public final class ReadBufferManagerV1 extends ReadBufferManager {
 
   private static final int NUM_BUFFERS = 16;
   private static final int NUM_THREADS = 8;
@@ -66,7 +66,7 @@ static void setReadBufferManagerConfigs(int readAheadBlockSize) {
    * Returns the singleton instance of ReadBufferManagerV1.
    * @return the singleton instance of ReadBufferManagerV1
    */
-  static ReadBufferManagerV1 getBufferManager() {
+  public static ReadBufferManagerV1 getBufferManager() {
     if (bufferManager == null) {
       LOCK.lock();
       try {
@@ -542,6 +542,13 @@ private void purgeList(AbfsInputStream stream, LinkedList<ReadBuffer> list) {
     }
   }
 
+  @Override
+  public void closeReadBufferManager() {
+    // no-op, as this is a singleton and should not be closed
+    // the buffers will be cleaned up when the JVM exits
+    LOGGER.debug("ReadBufferManagerV1 close called, but no action taken as it is a singleton.");
+  }
+
   /**
    * {@inheritDoc}
    */

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

@@ -29,6 +29,7 @@
 import java.lang.management.MemoryMXBean;
 import java.lang.management.MemoryUsage;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.Iterator;
 import java.util.LinkedList;
@@ -47,14 +48,15 @@
 
 import org.apache.hadoop.fs.azurebfs.utils.TracingContext;
 import org.apache.hadoop.classification.VisibleForTesting;
+import org.apache.hadoop.util.concurrent.HadoopExecutors;
 
 import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.ONE_HUNDRED;
 import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.ONE_MB;
 
 /**
  * The Improved Read Buffer Manager for Rest AbfsClient.
  */
-final class ReadBufferManagerV2 extends ReadBufferManager {
+public final class ReadBufferManagerV2 extends ReadBufferManager {
   // Internal constants
   private static final Logger LOGGER = LoggerFactory.getLogger(ReadBufferManagerV2.class);
   private static final ReentrantLock LOCK = new ReentrantLock();
@@ -350,6 +352,21 @@ public synchronized void purgeBuffersForStream(AbfsInputStream stream) {
     purgeList(stream, getCompletedReadList());
   }
 
+  @Override
+  public void closeReadBufferManager() {
+    printTraceLog("Closing ReadBufferManagerV2");
+    HadoopExecutors.shutdown(workerPool, LOGGER,
+        30, TimeUnit.SECONDS);
+    workerPool = null;
+    if (bufferPool != null) {
+      // help GC
+      Arrays.fill(bufferPool, null);
+      bufferPool = null;
+    }
+    setBufferManager(null); // reset the singleton instance
+    printTraceLog("ReadBufferManagerV2 closed");
+  }
+
   private boolean isAlreadyQueued(final String eTag, final long requestedOffset) {
     // returns true if any part of the buffer is already queued
     return (isInList(getReadAheadQueue(), eTag, requestedOffset)