From d33d44b36c7b23da4c6aa7521f6555f78fe01f88 Mon Sep 17 00:00:00 2001 From: caozhiqiang Date: Fri, 8 Aug 2025 19:04:01 +0800 Subject: [PATCH 1/2] Fix serial fsimage transfer during checkpoint with multiple namenodes --- .../server/namenode/ha/StandbyCheckpointer.java | 7 ++++--- .../namenode/ha/TestStandbyCheckpoints.java | 15 +++++++++++++++ 2 files changed, 19 insertions(+), 3 deletions(-) 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 e240921f67066c..5da3703571371a 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 @@ -248,9 +248,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 - ExecutorService executor = new ThreadPoolExecutor(0, activeNNAddresses.size(), 100, - TimeUnit.MILLISECONDS, new LinkedBlockingQueue(activeNNAddresses.size()), - uploadThreadFactory); + ExecutorService executor = + new ThreadPoolExecutor(activeNNAddresses.size(), activeNNAddresses.size(), 100, + TimeUnit.MILLISECONDS, new LinkedBlockingQueue(activeNNAddresses.size()), + uploadThreadFactory); // for right now, just match the upload to the nn address by convention. There is no need to // directly tie them together by adding a pair class. HashMap> uploads = 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 6f36d5e19bcdd0..b2b2d26c3ad27f 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 @@ -458,6 +458,21 @@ public void testCheckpointCancellationDuringUpload() throws Exception { cluster.transitionToStandby(0); cluster.transitionToActive(1); + GenericTestUtils.waitFor(new Supplier() { + @Override + public Boolean get() { + int transferThreadCount = 0; + ThreadMXBean threadBean = ManagementFactory.getThreadMXBean(); + ThreadInfo[] threads = threadBean.getThreadInfo( + threadBean.getAllThreadIds(), 1); + for (ThreadInfo thread: threads) { + if (thread.getThreadName().startsWith("TransferFsImageUpload")) { + transferThreadCount++; + } + } + return transferThreadCount == NUM_NNS - 1; + } + }, 1000, 30000); // Wait to make sure background TransferFsImageUpload thread was cancelled. // This needs to be done before the next test in the suite starts, so that a From 02be5f5eb439d812300d51ae99c026f3570df2c8 Mon Sep 17 00:00:00 2001 From: caozhiqiang Date: Fri, 10 Oct 2025 21:29:17 +0800 Subject: [PATCH 2/2] Add config to enable parallel upload fsimage with multiple observer namenodes --- .../org/apache/hadoop/hdfs/DFSConfigKeys.java | 3 + .../hdfs/server/namenode/CheckpointConf.java | 13 ++++ .../namenode/ha/StandbyCheckpointer.java | 8 +- .../src/main/resources/hdfs-default.xml | 11 +++ .../namenode/ha/TestStandbyCheckpoints.java | 77 +++++++++++++++---- 5 files changed, 92 insertions(+), 20 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 c72bc50c0fb76e..df9e3907bdaf2c 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 @@ -259,6 +259,9 @@ public class DFSConfigKeys extends CommonConfigurationKeys { public static final long DFS_NAMENODE_CHECKPOINT_TXNS_DEFAULT = 1000000; public static final String DFS_NAMENODE_CHECKPOINT_MAX_RETRIES_KEY = "dfs.namenode.checkpoint.max-retries"; public static final int DFS_NAMENODE_CHECKPOINT_MAX_RETRIES_DEFAULT = 3; + 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_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 4df170d7716017..a5f8049c95066e 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 @@ -54,6 +54,12 @@ public class CheckpointConf { */ private double quietMultiplier; + /** + * Whether enable the standby namenode to upload fsiamge to multiple other namenodes in + * parallel, in the cluster with observer namenodes. + */ + private final boolean parallelUploadEnabled; + public CheckpointConf(Configuration conf) { checkpointCheckPeriod = conf.getTimeDuration( DFS_NAMENODE_CHECKPOINT_CHECK_PERIOD_KEY, @@ -68,6 +74,9 @@ public CheckpointConf(Configuration conf) { legacyOivImageDir = conf.get(DFS_NAMENODE_LEGACY_OIV_IMAGE_DIR_KEY); quietMultiplier = conf.getDouble(DFS_NAMENODE_CHECKPOINT_QUIET_MULTIPLIER_KEY, DFS_NAMENODE_CHECKPOINT_QUIET_MULTIPLIER_DEFAULT); + parallelUploadEnabled = conf.getBoolean( + DFS_NAMENODE_CHECKPOINT_PARALLEL_UPLOAD_ENABLED_KEY, + DFS_NAMENODE_CHECKPOINT_PARALLEL_UPLOAD_ENABLED_DEFAULT); warnForDeprecatedConfigs(conf); } @@ -106,4 +115,8 @@ public String getLegacyOivImageDir() { public double getQuietPeriod() { return this.checkpointPeriod * this.quietMultiplier; } + + public boolean isParallelUploadEnabled() { + return parallelUploadEnabled; + } } 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 5da3703571371a..52e1d7cdfbe398 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 @@ -248,10 +248,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 - ExecutorService executor = - new ThreadPoolExecutor(activeNNAddresses.size(), activeNNAddresses.size(), 100, - TimeUnit.MILLISECONDS, new LinkedBlockingQueue(activeNNAddresses.size()), - uploadThreadFactory); + int poolSize = checkpointConf.isParallelUploadEnabled() ? activeNNAddresses.size() : 0; + ExecutorService executor = new ThreadPoolExecutor(poolSize, activeNNAddresses.size(), 100, + TimeUnit.MILLISECONDS, new LinkedBlockingQueue(activeNNAddresses.size()), + uploadThreadFactory); // for right now, just match the upload to the nn address by convention. There is no need to // directly tie them together by adding a pair class. HashMap> uploads = 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 b9d8b67dc122aa..2b889dd2adc5c9 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 @@ -1397,6 +1397,17 @@ + + dfs.namenode.checkpoint.parallel.upload.enabled + false + + If true, the CheckpointNode will upload the checkpoint image to multiple other + NameNodes in parallel, in the cluster with observer namenodes. You should + make sure the network bandwidth is sufficient. + If false, the fsimage will be uploaded serially to multiple namenodes. + + + 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 b2b2d26c3ad27f..ba1093b136bf34 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 @@ -458,21 +458,6 @@ public void testCheckpointCancellationDuringUpload() throws Exception { cluster.transitionToStandby(0); cluster.transitionToActive(1); - GenericTestUtils.waitFor(new Supplier() { - @Override - public Boolean get() { - int transferThreadCount = 0; - ThreadMXBean threadBean = ManagementFactory.getThreadMXBean(); - ThreadInfo[] threads = threadBean.getThreadInfo( - threadBean.getAllThreadIds(), 1); - for (ThreadInfo thread: threads) { - if (thread.getThreadName().startsWith("TransferFsImageUpload")) { - transferThreadCount++; - } - } - return transferThreadCount == NUM_NNS - 1; - } - }, 1000, 30000); // Wait to make sure background TransferFsImageUpload thread was cancelled. // This needs to be done before the next test in the suite starts, so that a @@ -498,7 +483,67 @@ public Boolean get() { // Assert that former active did not accept the canceled checkpoint file. assertEquals(0, nns[0].getFSImage().getMostRecentCheckpointTxId()); } - + + /** + * Test standby namenode upload fsiamge to multiple other namenodes in parallel, in the + * cluster with observer namenodes. + */ + @Test + @Timeout(value = 300) + public void testCheckpointParallelUpload() throws Exception { + // Set dfs.namenode.checkpoint.txns differently on the first NN to avoid it + // doing checkpoint when it becomes a standby + cluster.getConfiguration(0).setInt( + DFSConfigKeys.DFS_NAMENODE_CHECKPOINT_TXNS_KEY, 1000); + + // don't compress, we want a big image + for (int i = 0; i < NUM_NNS; i++) { + cluster.getConfiguration(i).setBoolean( + DFSConfigKeys.DFS_IMAGE_COMPRESS_KEY, false); + } + + // Throttle SBN upload to make it hang during upload to ANN, and enable parallel upload fsimage. + for (int i = 1; i < NUM_NNS; i++) { + cluster.getConfiguration(i).setLong( + DFSConfigKeys.DFS_IMAGE_TRANSFER_RATE_KEY, 100); + cluster.getConfiguration(i).setBoolean( + DFSConfigKeys.DFS_NAMENODE_CHECKPOINT_PARALLEL_UPLOAD_ENABLED_KEY, true); + } + for (int i = 0; i < NUM_NNS; i++) { + cluster.restartNameNode(i); + } + + // update references to each of the nns + 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)); + } + cluster.transitionToStandby(0); + cluster.transitionToActive(1); + + GenericTestUtils.waitFor(new Supplier() { + @Override + public Boolean get() { + int transferThreadCount = 0; + ThreadMXBean threadBean = ManagementFactory.getThreadMXBean(); + ThreadInfo[] threads = threadBean.getThreadInfo( + threadBean.getAllThreadIds(), 1); + for (ThreadInfo thread: threads) { + if (thread.getThreadName().startsWith("TransferFsImageUpload")) { + transferThreadCount++; + } + } + return transferThreadCount == NUM_NNS - 1; + } + }, 1000, 30000); + } + /** * Make sure that clients will receive StandbyExceptions even when a * checkpoint is in progress on the SBN, and therefore the StandbyCheckpointer