Fix serial fsimage transfer during checkpoint with multiple namenodes

Author: lfxy

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<Runnable>(activeNNAddresses.size()),
-        uploadThreadFactory);
+    ExecutorService executor =
+        new ThreadPoolExecutor(activeNNAddresses.size(), activeNNAddresses.size(), 100,
+            TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(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<String, Future<TransferFsImage.TransferResult>> uploads =

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<Boolean>() {
+      @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