HDDS-14020. Use ForkJoinPool instead of a ScheduledThreadPoolExecutor in BackgroundService

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/BackgroundServiceScheduler.java

@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hdds.utils;
+
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import org.apache.ratis.util.ReferenceCountedObject;
+import org.apache.ratis.util.function.UncheckedAutoCloseableSupplier;
+
+/**
+ * Utility class to manage a shared background service using a {@link ScheduledExecutorService}
+ * which is provided with a single-threaded {@link ScheduledThreadPoolExecutor}.
+ * This class manages the lifecycle and reference counting for the executor
+ * to ensure proper resource cleanup.
+ *
+ * The executor is lazily initialized on the first invocation of the {@code get()} method.
+ * It is shut down and released when no longer referenced, ensuring efficient use
+ * of system resources. The shutdown process includes cleaning the reference to the executor.
+ *
+ * This class is thread-safe.
+ */
+final class BackgroundServiceScheduler {
+  private static ReferenceCountedObject<ScheduledExecutorService> executor;
+
+  private BackgroundServiceScheduler() {
+
+  }
+
+  public static synchronized UncheckedAutoCloseableSupplier<ScheduledExecutorService> get() {
+    if (executor == null) {
+      ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(1);
+      executor = ReferenceCountedObject.wrap(scheduler, () -> { }, (shutdown) -> {
+        if (shutdown) {
+          synchronized (BackgroundServiceScheduler.class) {
+            scheduler.shutdown();
+            executor = null;
+          }
+        }
+      });
+    }
+    return executor.retainAndReleaseOnClose();
+  }
+}

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/diskbalancer/DiskBalancerService.java

@@ -37,6 +37,7 @@
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.ForkJoinPool;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
@@ -258,11 +259,15 @@ private void applyDiskBalancerInfo(DiskBalancerInfo diskBalancerInfo)
     setStopAfterDiskEven(diskBalancerInfo.isStopAfterDiskEven());
     setVersion(diskBalancerInfo.getVersion());
 
-    // Default executorService is ScheduledThreadPoolExecutor, so we can
-    // update the poll size by setting corePoolSize.
-    if ((getExecutorService() instanceof ScheduledThreadPoolExecutor)) {
-      ((ScheduledThreadPoolExecutor) getExecutorService())
+    Object executorService = getExecutorService();
+    if (executorService instanceof ScheduledThreadPoolExecutor) {
+      // Update the pool size by setting corePoolSize for ScheduledThreadPoolExecutor
+      ((ScheduledThreadPoolExecutor) executorService)
           .setCorePoolSize(parallelThread);
+    } else if (executorService instanceof ForkJoinPool) {
+      // For ForkJoinPool, dynamic resizing is not supported and requires service restart
+      LOG.warn("ForkJoinPool doesn't support dynamic pool size changes. " +
+          "Service restart is required for pool size change to take effect.");
     }
   }
 

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/BlockDeletingServiceTestImpl.java

@@ -68,7 +68,7 @@ int getTimesOfProcessed() {
   // Override the implementation to start a single on-call control thread.
   @Override
   public void start() {
-    PeriodicalTask svc = new PeriodicalTask();
+    PeriodicalTask svc = new PeriodicalTask(null);
     // In test mode, relies on a latch countdown to runDeletingTasks tasks.
     Runnable r = () -> {
       while (true) {

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/diskbalancer/DiskBalancerServiceTestImpl.java

@@ -65,7 +65,7 @@ public int getTimesOfProcessed() {
   // Override the implementation to start a single on-call control thread.
   @Override
   public void start() {
-    PeriodicalTask svc = new PeriodicalTask();
+    PeriodicalTask svc = new PeriodicalTask(null);
     // In test mode, relies on a latch countdown to runDeletingTasks tasks.
     Runnable r = () -> {
       while (true) {