diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java index 2fb2de5746fafa..214b11a43936fe 100755 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java @@ -262,6 +262,9 @@ public class DFSConfigKeys extends CommonConfigurationKeys { public static final String DFS_NAMENODE_CHECKPOINT_PARALLEL_UPLOAD_ENABLED_KEY = "dfs.namenode.checkpoint.parallel.upload.enabled"; public static final boolean DFS_NAMENODE_CHECKPOINT_PARALLEL_UPLOAD_ENABLED_DEFAULT = false; + public static final String DFS_NAMENODE_CHECKPOINT_PARALLEL_UPLOAD_MAX_THREADS_KEY = + "dfs.namenode.checkpoint.parallel.upload.max-threads"; + public static final int DFS_NAMENODE_CHECKPOINT_PARALLEL_UPLOAD_MAX_THREADS_DEFAULT = 0; public static final String DFS_NAMENODE_MISSING_CHECKPOINT_PERIODS_BEFORE_SHUTDOWN_KEY = "dfs.namenode.missing.checkpoint.periods.before.shutdown"; public static final int DFS_NAMENODE_MISSING_CHECKPOINT_PERIODS_BEFORE_SHUTDOWN_DEFAULT = 3; public static final String DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_KEY = diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/CheckpointConf.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/CheckpointConf.java index a5f8049c95066e..65534384a2f0e2 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/CheckpointConf.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/CheckpointConf.java @@ -23,6 +23,7 @@ import org.slf4j.LoggerFactory; import org.apache.hadoop.classification.InterfaceAudience; import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.util.Preconditions; import org.apache.hadoop.thirdparty.com.google.common.collect.ImmutableList; @@ -60,6 +61,11 @@ public class CheckpointConf { */ private final boolean parallelUploadEnabled; + /** + * Max number of concurrent fsimage uploads from standby checkpointer. + */ + private final int parallelUploadMaxThreads; + public CheckpointConf(Configuration conf) { checkpointCheckPeriod = conf.getTimeDuration( DFS_NAMENODE_CHECKPOINT_CHECK_PERIOD_KEY, @@ -77,6 +83,12 @@ public CheckpointConf(Configuration conf) { parallelUploadEnabled = conf.getBoolean( DFS_NAMENODE_CHECKPOINT_PARALLEL_UPLOAD_ENABLED_KEY, DFS_NAMENODE_CHECKPOINT_PARALLEL_UPLOAD_ENABLED_DEFAULT); + parallelUploadMaxThreads = conf.getInt( + DFS_NAMENODE_CHECKPOINT_PARALLEL_UPLOAD_MAX_THREADS_KEY, + DFS_NAMENODE_CHECKPOINT_PARALLEL_UPLOAD_MAX_THREADS_DEFAULT); + Preconditions.checkArgument(parallelUploadMaxThreads >= 0, + DFS_NAMENODE_CHECKPOINT_PARALLEL_UPLOAD_MAX_THREADS_KEY + + " must be greater than or equal to 0"); warnForDeprecatedConfigs(conf); } @@ -119,4 +131,8 @@ public double getQuietPeriod() { public boolean isParallelUploadEnabled() { return parallelUploadEnabled; } + + public int getParallelUploadMaxThreads() { + return parallelUploadMaxThreads; + } } diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/ha/StandbyCheckpointer.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/ha/StandbyCheckpointer.java index 0e8c067322ba3d..5c91ee5eb52fa9 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/ha/StandbyCheckpointer.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/ha/StandbyCheckpointer.java @@ -247,7 +247,18 @@ private void doCheckpoint() throws InterruptedException, IOException { // Do this in a separate thread to avoid blocking transition to active, but don't allow more // than the expected number of tasks to run or queue up // See HDFS-4816 - int poolSize = checkpointConf.isParallelUploadEnabled() ? remoteNNAddresses.size() : 0; + // If parallel upload is abled, Parallel upload concurrency policy: + // - If maxThreads is 0, concurrency equals the number of remote NameNodes. + // - If maxThreads is greater than 0, concurrency is the smaller of + // the remote Namenodes count and maxThreads. + int maxThreads = checkpointConf.getParallelUploadMaxThreads(); + int poolSize = checkpointConf.isParallelUploadEnabled() ? + (maxThreads == 0 ? + remoteNNAddresses.size() : + Math.min(remoteNNAddresses.size(), maxThreads)) : + 0; + LOG.info("Fsimage upload concurrency: {}, remote namenodes count: {}", poolSize, + remoteNNAddresses.size()); ExecutorService executor = new ThreadPoolExecutor(poolSize, remoteNNAddresses.size(), 100, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(remoteNNAddresses.size()), uploadThreadFactory); diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml b/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml index 1c56231fe26c61..b9121ef886ac65 100755 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml @@ -1429,6 +1429,22 @@ + + dfs.namenode.checkpoint.parallel.upload.max-threads + 0 + + This config sets the maximum number of concurrent threads used to upload fsimage + from StandbyCheckpointer to other NameNodes. + It takes effect only when dfs.namenode.checkpoint.parallel.upload.enabled is true. + A smaller value reduces upload concurrency pressure on both the local standby. + If set to 0, the concurrency equals the number of remote NameNodes. + If greater than 0, the concurrency is the smaller of the remote NameNodes count and maxThreads. + Note: A lower max-threads value may slow fsimage uploads and extend the effective + checkpoint interval (dfs.namenode.checkpoint.period), tune this setting according to + your deployment environment. + + + dfs.namenode.checkpoint.check.quiet-multiplier 1.5 diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/ha/TestStandbyCheckpoints.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/ha/TestStandbyCheckpoints.java index 1b03ec946893b9..c6b49590bac55c 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/ha/TestStandbyCheckpoints.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/ha/TestStandbyCheckpoints.java @@ -29,6 +29,7 @@ import java.util.List; import java.util.Random; import java.util.function.Supplier; +import java.util.concurrent.atomic.AtomicInteger; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; @@ -829,6 +830,60 @@ public void testPutFsimagePartFailed() throws Exception { assertTrue(snnCheckpointTime2 > snnCheckpointTime1); } + @Test + @Timeout(value = 300) + public void testCheckpointParallelUploadWithMaxThreads() throws Exception { + final int maxThreads = 1; + final AtomicInteger inFlightUploads = new AtomicInteger(0); + final AtomicInteger peakInFlightUploads = new AtomicInteger(0); + + for (int i = 1; i < NUM_NNS; i++) { + cluster.getConfiguration(i) + .setBoolean(DFSConfigKeys.DFS_NAMENODE_CHECKPOINT_PARALLEL_UPLOAD_ENABLED_KEY, true); + cluster.getConfiguration(i) + .setInt(DFSConfigKeys.DFS_NAMENODE_CHECKPOINT_PARALLEL_UPLOAD_MAX_THREADS_KEY, + maxThreads); + cluster.getConfiguration(i).setLong(DFSConfigKeys.DFS_IMAGE_TRANSFER_RATE_KEY, 100); + } + + CheckpointFaultInjector oldInjector = CheckpointFaultInjector.getInstance(); + CheckpointFaultInjector.set(new CheckpointFaultInjector() { + @Override + public void duringUploadInProgess() throws InterruptedException { + int currentInFlight = inFlightUploads.incrementAndGet(); + peakInFlightUploads.updateAndGet(previousPeak -> Math.max(previousPeak, currentInFlight)); + try { + Thread.sleep(100); + } finally { + inFlightUploads.decrementAndGet(); + } + } + }); + + try { + for (int i = 0; i < NUM_NNS; i++) { + cluster.restartNameNode(i); + } + setNNs(); + + cluster.transitionToActive(0); + doEdits(0, 100); + + for (int i = 1; i < NUM_NNS; i++) { + HATestUtil.waitForStandbyToCatchUp(nns[0], nns[i]); + HATestUtil.waitForCheckpoint(cluster, i, ImmutableList.of(104)); + } + + assertTrue(peakInFlightUploads.get() > 0, + "Expected at least one fsimage upload to be observed."); + assertTrue(peakInFlightUploads.get() <= maxThreads, + "Observed upload concurrency " + peakInFlightUploads.get() + + " exceeds configured max threads " + maxThreads); + } finally { + CheckpointFaultInjector.set(oldInjector); + } + } + /** * A codec which just slows down the saving of the image significantly * by sleeping a few milliseconds on every write. This makes it easy to