Merge branch 'apache:trunk' into YARN-11374

Author: slfan1989

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/resolver/MembershipNamenodeResolver.java

@@ -193,13 +193,53 @@ private void updateNameNodeState(final String nsId,
     }
   }
 
+  /**
+   * Try to shuffle the multiple observer namenodes if listObserversFirst is true.
+   * @param inputNameNodes the input FederationNamenodeContext list. If listObserversFirst is true,
+   *                       all observers will be placed at the front of the collection.
+   * @param listObserversFirst true if we need to shuffle the multiple front observer namenodes.
+   * @return a list of FederationNamenodeContext.
+   * @param <T> a subclass of FederationNamenodeContext.
+   */
+  private <T extends FederationNamenodeContext> List<T> shuffleObserverNN(
+      List<T> inputNameNodes, boolean listObserversFirst) {
+    if (!listObserversFirst) {
+      return inputNameNodes;
+    }
+    // Get Observers first.
+    List<T> observerList = new ArrayList<>();
+    for (T t : inputNameNodes) {
+      if (t.getState() == OBSERVER) {
+        observerList.add(t);
+      } else {
+        // The inputNameNodes are already sorted, so it can break
+        // when the first non-observer is encountered.
+        break;
+      }
+    }
+    // Returns the inputNameNodes if no shuffle is required
+    if (observerList.size() <= 1) {
+      return inputNameNodes;
+    }
+
+    // Shuffle multiple Observers
+    Collections.shuffle(observerList);
+
+    List<T> ret = new ArrayList<>(inputNameNodes.size());
+    ret.addAll(observerList);
+    for (int i = observerList.size(); i < inputNameNodes.size(); i++) {
+      ret.add(inputNameNodes.get(i));
+    }
+    return Collections.unmodifiableList(ret);
+  }
+
   @Override
   public List<? extends FederationNamenodeContext> getNamenodesForNameserviceId(
       final String nsId, boolean listObserversFirst) throws IOException {
 
     List<? extends FederationNamenodeContext> ret = cacheNS.get(Pair.of(nsId, listObserversFirst));
     if (ret != null) {
-      return ret;
+      return shuffleObserverNN(ret, listObserversFirst);
     }
 
     // Not cached, generate the value

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterClientProtocol.java

@@ -702,8 +702,9 @@ public boolean truncate(String src, long newLength, String clientName)
     RemoteMethod method = new RemoteMethod("truncate",
         new Class<?>[] {String.class, long.class, String.class},
         new RemoteParam(), newLength, clientName);
+    // Truncate can return true/false, so don't expect a result
     return rpcClient.invokeSequential(locations, method, Boolean.class,
-        Boolean.TRUE);
+        null);
   }
 
   @Override

hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/resolver/TestNamenodeResolver.java

@@ -90,6 +90,98 @@ public void setup() throws IOException, InterruptedException {
     assertTrue(cleared);
   }
 
+  @Test
+  public void testShuffleObserverNNs() throws Exception {
+    // Add an active entry to the store
+    NamenodeStatusReport activeReport = createNamenodeReport(
+        NAMESERVICES[0], NAMENODES[0], HAServiceState.ACTIVE);
+    assertTrue(namenodeResolver.registerNamenode(activeReport));
+
+    // Add a standby entry to the store
+    NamenodeStatusReport standbyReport = createNamenodeReport(
+        NAMESERVICES[0], NAMENODES[1], HAServiceState.STANDBY);
+    assertTrue(namenodeResolver.registerNamenode(standbyReport));
+
+    // Load cache
+    stateStore.refreshCaches(true);
+
+    // Get namenodes from state store.
+    List<? extends FederationNamenodeContext> withoutObserver =
+        namenodeResolver.getNamenodesForNameserviceId(NAMESERVICES[0], true);
+    assertEquals(2, withoutObserver.size());
+    assertEquals(FederationNamenodeServiceState.ACTIVE, withoutObserver.get(0).getState());
+    assertEquals(FederationNamenodeServiceState.STANDBY, withoutObserver.get(1).getState());
+
+    // Get namenodes from cache.
+    withoutObserver = namenodeResolver.getNamenodesForNameserviceId(NAMESERVICES[0], true);
+    assertEquals(2, withoutObserver.size());
+    assertEquals(FederationNamenodeServiceState.ACTIVE, withoutObserver.get(0).getState());
+    assertEquals(FederationNamenodeServiceState.STANDBY, withoutObserver.get(1).getState());
+
+    // Add an observer entry to the store
+    NamenodeStatusReport observerReport1 = createNamenodeReport(
+        NAMESERVICES[0], NAMENODES[2], HAServiceState.OBSERVER);
+    assertTrue(namenodeResolver.registerNamenode(observerReport1));
+
+    // Load cache
+    stateStore.refreshCaches(true);
+
+    // Get namenodes from state store.
+    List<? extends FederationNamenodeContext> observerList =
+        namenodeResolver.getNamenodesForNameserviceId(NAMESERVICES[0], true);
+    assertEquals(3, observerList.size());
+    assertEquals(FederationNamenodeServiceState.OBSERVER, observerList.get(0).getState());
+    assertEquals(FederationNamenodeServiceState.ACTIVE, observerList.get(1).getState());
+    assertEquals(FederationNamenodeServiceState.STANDBY, observerList.get(2).getState());
+
+    // Get namenodes from cache.
+    observerList = namenodeResolver.getNamenodesForNameserviceId(NAMESERVICES[0], true);
+    assertEquals(3, observerList.size());
+    assertEquals(FederationNamenodeServiceState.OBSERVER, observerList.get(0).getState());
+    assertEquals(FederationNamenodeServiceState.ACTIVE, observerList.get(1).getState());
+    assertEquals(FederationNamenodeServiceState.STANDBY, observerList.get(2).getState());
+
+    // Add one new observer entry to the store
+    NamenodeStatusReport observerReport2 = createNamenodeReport(
+        NAMESERVICES[0], NAMENODES[3], HAServiceState.OBSERVER);
+    assertTrue(namenodeResolver.registerNamenode(observerReport2));
+
+    // Load cache
+    stateStore.refreshCaches(true);
+
+    // Get namenodes from state store.
+    List<? extends FederationNamenodeContext> observerList2 =
+        namenodeResolver.getNamenodesForNameserviceId(NAMESERVICES[0], true);
+    assertEquals(4, observerList2.size());
+    assertEquals(FederationNamenodeServiceState.OBSERVER, observerList2.get(0).getState());
+    assertEquals(FederationNamenodeServiceState.OBSERVER, observerList2.get(1).getState());
+    assertEquals(FederationNamenodeServiceState.ACTIVE, observerList2.get(2).getState());
+    assertEquals(FederationNamenodeServiceState.STANDBY, observerList2.get(3).getState());
+
+    // Get namenodes from cache.
+    observerList2 = namenodeResolver.getNamenodesForNameserviceId(NAMESERVICES[0], true);
+    assertEquals(4, observerList2.size());
+    assertEquals(FederationNamenodeServiceState.OBSERVER, observerList2.get(0).getState());
+    assertEquals(FederationNamenodeServiceState.OBSERVER, observerList2.get(1).getState());
+    assertEquals(FederationNamenodeServiceState.ACTIVE, observerList2.get(2).getState());
+    assertEquals(FederationNamenodeServiceState.STANDBY, observerList2.get(3).getState());
+
+    // Test shuffler
+    List<? extends FederationNamenodeContext> observerList3;
+    boolean hit = false;
+    for (int i = 0; i < 1000; i++) {
+      observerList3 = namenodeResolver.getNamenodesForNameserviceId(NAMESERVICES[0], true);
+      assertEquals(FederationNamenodeServiceState.OBSERVER, observerList3.get(0).getState());
+      assertEquals(FederationNamenodeServiceState.OBSERVER, observerList3.get(1).getState());
+      if (observerList3.get(0).getNamenodeId().equals(observerList2.get(1).getNamenodeId()) &&
+          observerList3.get(1).getNamenodeId().equals(observerList2.get(0).getNamenodeId())) {
+        hit = true;
+        break;
+      }
+    }
+    assertTrue(hit);
+  }
+
   @Test
   public void testStateStoreDisconnected() throws Exception {
 

hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterAllResolver.java

@@ -33,6 +33,7 @@
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.DistributedFileSystem;
 import org.apache.hadoop.hdfs.server.federation.RouterConfigBuilder;
 import org.apache.hadoop.hdfs.server.federation.MiniRouterDFSCluster.NamenodeContext;
 import org.apache.hadoop.hdfs.server.federation.MiniRouterDFSCluster.RouterContext;
@@ -46,6 +47,7 @@
 import org.apache.hadoop.hdfs.server.federation.store.protocol.GetMountTableEntriesRequest;
 import org.apache.hadoop.hdfs.server.federation.store.protocol.GetMountTableEntriesResponse;
 import org.apache.hadoop.hdfs.server.federation.store.records.MountTable;
+import org.apache.hadoop.hdfs.server.namenode.TestFileTruncate;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -191,6 +193,18 @@ private void testAll(final String path) throws Exception {
     assertDirsEverywhere(path, 9);
     assertFilesDistributed(path, 15);
 
+    // Test truncate
+    String testTruncateFile = path + "/dir2/dir22/dir220/file-truncate.txt";
+    createTestFile(routerFs, testTruncateFile);
+    Path testTruncateFilePath = new Path(testTruncateFile);
+    routerFs.truncate(testTruncateFilePath, 10);
+    TestFileTruncate.checkBlockRecovery(testTruncateFilePath,
+        (DistributedFileSystem) routerFs);
+    assertEquals("Truncate file fails", 10,
+        routerFs.getFileStatus(testTruncateFilePath).getLen());
+    assertDirsEverywhere(path, 9);
+    assertFilesDistributed(path, 16);
+
     // Removing a directory should remove it from every subcluster
     routerFs.delete(new Path(path + "/dir2/dir22/dir220"), true);
     assertDirsEverywhere(path, 8);

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSEditLog.java

@@ -1654,18 +1654,31 @@ synchronized void logEdit(final int length, final byte[] data) {
     endTransaction(start);
   }
 
+  void recoverUnclosedStreams() throws IOException {
+    recoverUnclosedStreams(false);
+  }
+
   /**
    * Run recovery on all journals to recover any unclosed segments
    */
-  synchronized void recoverUnclosedStreams() {
+  synchronized void recoverUnclosedStreams(boolean terminateOnFailure) throws IOException {
     Preconditions.checkState(
         state == State.BETWEEN_LOG_SEGMENTS,
         "May not recover segments - wrong state: %s", state);
     try {
       journalSet.recoverUnfinalizedSegments();
     } catch (IOException ex) {
-      // All journals have failed, it is handled in logSync.
-      // TODO: are we sure this is OK?
+      if (terminateOnFailure) {
+        final String msg = "Unable to recover log segments: "
+            + "too few journals successfully recovered.";
+        LOG.error(msg, ex);
+        synchronized (journalSetLock) {
+          IOUtils.cleanupWithLogger(LOG, journalSet);
+        }
+        terminate(1, msg);
+      } else {
+        throw ex;
+      }
     }
   }
 

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java

@@ -1389,7 +1389,7 @@ void startActiveServices() throws IOException {
         // During startup, we're already open for write during initialization.
         editLog.initJournalsForWrite();
         // May need to recover
-        editLog.recoverUnclosedStreams();
+        editLog.recoverUnclosedStreams(true);
 
         LOG.info("Catching up to latest edits from old active before " +
             "taking over writer role in edits logs");

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/ha/EditLogTailer.java

@@ -311,7 +311,8 @@ public Void run() throws Exception {
                 startTime - lastLoadTimeMs);
             // It is already under the name system lock and the checkpointer
             // thread is already stopped. No need to acquire any other lock.
-            editsTailed = doTailEdits();
+            // HDFS-16689. Disable inProgress to use the streaming mechanism
+            editsTailed = doTailEdits(false);
           } catch (InterruptedException e) {
             throw new IOException(e);
           } finally {
@@ -323,9 +324,13 @@ public Void run() throws Exception {
       }
     });
   }
-  
+
   @VisibleForTesting
   public long doTailEdits() throws IOException, InterruptedException {
+    return doTailEdits(inProgressOk);
+  }
+
+  private long doTailEdits(boolean enableInProgress) throws IOException, InterruptedException {
     Collection<EditLogInputStream> streams;
     FSImage image = namesystem.getFSImage();
 
@@ -334,7 +339,7 @@ public long doTailEdits() throws IOException, InterruptedException {
     long startTime = timer.monotonicNow();
     try {
       streams = editLog.selectInputStreams(lastTxnId + 1, 0,
-          null, inProgressOk, true);
+          null, enableInProgress, true);
     } catch (IOException ioe) {
       // This is acceptable. If we try to tail edits in the middle of an edits
       // log roll, i.e. the last one has been finalized but the new inprogress

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/qjournal/TestNNWithQJM.java

@@ -33,7 +33,6 @@
 import org.apache.hadoop.ipc.RemoteException;
 import org.apache.hadoop.test.GenericTestUtils;
 import org.apache.hadoop.util.ExitUtil;
-import org.apache.hadoop.util.ExitUtil.ExitException;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -197,10 +196,9 @@ public void testMismatchedNNIsRejected() throws Exception {
           .manageNameDfsDirs(false).format(false).checkExitOnShutdown(false)
           .build();
       fail("New NN with different namespace should have been rejected");
-    } catch (ExitException ee) {
+    } catch (IOException ioe) {
       GenericTestUtils.assertExceptionContains(
-          "Unable to start log segment 1: too few journals", ee);
-      assertTrue("Didn't terminate properly ", ExitUtil.terminateCalled());
+          "recoverUnfinalizedSegments failed for too many journals", ioe);
     }
   }
 }