From fe1b5afa954726d7b37d5bf37f7937524f6e5da7 Mon Sep 17 00:00:00 2001 From: huangzhaobo Date: Wed, 24 Jun 2026 15:13:30 +0800 Subject: [PATCH 1/4] HDFS-17937. Limit fsimage transfer concurrency during checkpoint with multiple NameNodes. --- .../org/apache/hadoop/hdfs/DFSConfigKeys.java | 3 + .../hdfs/server/namenode/CheckpointConf.java | 15 +++++ .../namenode/ha/StandbyCheckpointer.java | 5 +- .../src/main/resources/hdfs-default.xml | 11 ++++ .../namenode/ha/TestStandbyCheckpoints.java | 57 +++++++++++++++++++ 5 files changed, 90 insertions(+), 1 deletion(-) 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..ccce585ee35fb7 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 = 1; 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..6b2c2b937312e3 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,11 @@ 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 0"); warnForDeprecatedConfigs(conf); } @@ -119,4 +130,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..afebac4332a584 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,10 @@ 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; + int poolSize = checkpointConf.isParallelUploadEnabled() ? + Math.min(remoteNNAddresses.size(), checkpointConf.getParallelUploadMaxThreads()) : 0; + LOG.info("Fsimage upload concurrency: {}, remote namenodes count: {}, maxThreads: {}", poolSize, + remoteNNAddresses.size(), checkpointConf.getParallelUploadMaxThreads()); 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..26a6da3191a963 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,17 @@ + + dfs.namenode.checkpoint.parallel.upload.max-threads + 1 + + 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. + + + 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..365bef38cfd8c9 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; @@ -511,6 +512,8 @@ public void testCheckpointParallelUpload() throws Exception { DFSConfigKeys.DFS_IMAGE_TRANSFER_RATE_KEY, 100); 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, NUM_NNS); } for (int i = 0; i < NUM_NNS; i++) { cluster.restartNameNode(i); @@ -829,6 +832,60 @@ public void testPutFsimagePartFailed() throws Exception { assertTrue(snnCheckpointTime2 > snnCheckpointTime1); } + @Test + @Timeout(value = 300) + public void testCheckpointParallelUploadRespectsMaxThreads() 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 From e0b979afd45ba7ee2082f86a4cfc220fbc6c8331 Mon Sep 17 00:00:00 2001 From: huangzhaobo Date: Thu, 2 Jul 2026 20:57:34 +0800 Subject: [PATCH 2/4] fix parallel upload concurrency policy & fix style --- .../java/org/apache/hadoop/hdfs/DFSConfigKeys.java | 2 +- .../hdfs/server/namenode/CheckpointConf.java | 5 +++-- .../server/namenode/ha/StandbyCheckpointer.java | 14 +++++++++++--- .../src/main/resources/hdfs-default.xml | 4 +++- .../server/namenode/ha/TestStandbyCheckpoints.java | 2 +- 5 files changed, 19 insertions(+), 8 deletions(-) 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 ccce585ee35fb7..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 @@ -264,7 +264,7 @@ public class DFSConfigKeys extends CommonConfigurationKeys { 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 = 1; + 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 6b2c2b937312e3..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 @@ -86,8 +86,9 @@ public CheckpointConf(Configuration conf) { 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 0"); + Preconditions.checkArgument(parallelUploadMaxThreads >= 0, + DFS_NAMENODE_CHECKPOINT_PARALLEL_UPLOAD_MAX_THREADS_KEY + + " must be greater than or equal to 0"); warnForDeprecatedConfigs(conf); } 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 afebac4332a584..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,10 +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 + // 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() ? - Math.min(remoteNNAddresses.size(), checkpointConf.getParallelUploadMaxThreads()) : 0; - LOG.info("Fsimage upload concurrency: {}, remote namenodes count: {}, maxThreads: {}", poolSize, - remoteNNAddresses.size(), checkpointConf.getParallelUploadMaxThreads()); + (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 26a6da3191a963..3022a58d39516d 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 @@ -1431,12 +1431,14 @@ dfs.namenode.checkpoint.parallel.upload.max-threads - 1 + 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. 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 365bef38cfd8c9..f68a6db640a56b 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 @@ -834,7 +834,7 @@ public void testPutFsimagePartFailed() throws Exception { @Test @Timeout(value = 300) - public void testCheckpointParallelUploadRespectsMaxThreads() throws Exception { + public void testCheckpointParallelUploadWithMaxThreads() throws Exception { final int maxThreads = 1; final AtomicInteger inFlightUploads = new AtomicInteger(0); final AtomicInteger peakInFlightUploads = new AtomicInteger(0); From 1350f5d89171cb615f4ec0b5a6c4cc3caa2fa5c8 Mon Sep 17 00:00:00 2001 From: huangzhaobo Date: Mon, 6 Jul 2026 15:21:41 +0800 Subject: [PATCH 3/4] remove invalid code --- .../hadoop/hdfs/server/namenode/ha/TestStandbyCheckpoints.java | 2 -- 1 file changed, 2 deletions(-) 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 f68a6db640a56b..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 @@ -512,8 +512,6 @@ public void testCheckpointParallelUpload() throws Exception { DFSConfigKeys.DFS_IMAGE_TRANSFER_RATE_KEY, 100); 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, NUM_NNS); } for (int i = 0; i < NUM_NNS; i++) { cluster.restartNameNode(i); From 6a149c734d5e774d2115aaea8cfb4bdffde204e4 Mon Sep 17 00:00:00 2001 From: huangzhaobo Date: Mon, 6 Jul 2026 21:36:03 +0800 Subject: [PATCH 4/4] add config note --- .../hadoop-hdfs/src/main/resources/hdfs-default.xml | 3 +++ 1 file changed, 3 insertions(+) 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 3022a58d39516d..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 @@ -1439,6 +1439,9 @@ 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.