Graceful shutdown FileCopyThread in FCH (#3083)

* Graceful shutdown FileCopyThread in FCH

* Graceful shutdown FileCopyThread in FCH - Patch(2)

* Graceful shutdown FileCopyThread in FCH - Patch(3)

---------

Co-authored-by: Aman Pandey <[email protected]>

Author: amanatgithub

ambry-file-transfer/src/main/java/com/github/ambry/filetransfer/DiskAwareFileCopyThreadPoolManager.java

@@ -182,6 +182,18 @@ boolean areThreadsRunning() {
   @Override
   public void shutdown() throws InterruptedException {
     isRunning = false;
+    logger.info("Shutting down DiskAwareFileCopyThreadPoolManager");
+    threadQueueLock.lock();
+    runningThreads.forEach((diskId, fileCopyThreads) -> {
+      fileCopyThreads.forEach(fileCopyThread -> {
+        try {
+          fileCopyThread.shutDown();
+        } catch (Exception e) {
+          logger.error("Error while shutting down thread {}", fileCopyThread.threadName, e);
+        }
+      });
+    });
+    threadQueueLock.unlock();
     shutdownLatch.await();
   }
 }

ambry-file-transfer/src/main/java/com/github/ambry/filetransfer/FileCopyBasedReplicationSchedulerImpl.java

@@ -45,6 +45,7 @@ class FileCopyBasedReplicationSchedulerImpl implements FileCopyBasedReplicationS
   private final FileCopyHandlerFactory fileCopyHandlerFactory;
   private final ClusterMap clusterMap;
   private final FileCopyBasedReplicationThreadPoolManager fileCopyBasedReplicationThreadPoolManager;
+  private final Thread fileCopyBasedReplicationThreadPoolManagerThread;
   private final Map<ReplicaId, Long> replicaToStartTimeMap;
 
   private final Map<ReplicaId, FileCopyStatusListener> replicaToStatusListenerMap;
@@ -77,6 +78,7 @@ public FileCopyBasedReplicationSchedulerImpl(@Nonnull FileCopyHandlerFactory fil
     this.clusterMap = clusterMap;
     this.fileCopyBasedReplicationThreadPoolManager = new DiskAwareFileCopyThreadPoolManager(dataNodeId.getDiskIds(),
         fileCopyBasedReplicationConfig.fileCopyNumberOfFileCopyThreads);
+    this.fileCopyBasedReplicationThreadPoolManagerThread = new Thread(fileCopyBasedReplicationThreadPoolManager);
     this.replicaToStartTimeMap = new ConcurrentHashMap<>();
     this.inFlightReplicas = new LinkedList<>();
     this.prioritizationManager = prioritizationManager;
@@ -90,6 +92,7 @@ public void run(){
     isRunning = true;
     logger.info("FileCopyBasedReplicationSchedulerImpl Started");
     try {
+      fileCopyBasedReplicationThreadPoolManagerThread.start();
       scheduleFileCopy();
     } catch (InterruptedException e) {
       logger.error("Failed to start FileCopy Scheduler", e);
@@ -128,8 +131,11 @@ List<ReplicaId> getNextReplicaToHydrate(DiskId diskId, int numberOfReplicasOnDis
 
   @Override
   public void shutdown() throws InterruptedException {
+    logger.info("Shutting down FileCopyBasedReplicationSchedulerImpl");
     isRunning = false;
     fileCopyBasedReplicationThreadPoolManager.shutdown();
+    fileCopyBasedReplicationThreadPoolManagerThread.join();
+    logger.info("FileCopyBasedReplicationSchedulerImpl shutdown");
   }
 
   @Override

ambry-file-transfer/src/main/java/com/github/ambry/filetransfer/FileCopyBasedReplicationThreadPoolManager.java

@@ -25,7 +25,7 @@
  * assignments of partitions for hydration to threads in the thread pool.
  */
 
-public interface FileCopyBasedReplicationThreadPoolManager {
+public interface FileCopyBasedReplicationThreadPoolManager extends Runnable {
 
   /**
    * @return the number thread pool size.

ambry-file-transfer/src/main/java/com/github/ambry/filetransfer/handler/StoreFileCopyHandler.java

@@ -118,10 +118,9 @@ void stop() {
   }
 
   /**
-   * Shutdown the file copy handler. Perform clean up steps in case of a graceful shutdown.
+   * Shutdown the file copy handler.
    */
   public void shutdown() {
-    connectionPool.shutdown();
     isRunning = false;
   }