refresh the cache for the key Pair.of(nsId, true) at times.

KeeProMise · Aug 30, 2023 · 85e81b5 · 85e81b5
1 parent 055937c
commit 85e81b5
Show file tree

Hide file tree

Showing 4 changed files with 10 additions and 7 deletions.
diff --git a/...c/main/java/org/apache/hadoop/hdfs/server/federation/resolver/ActiveNamenodeResolver.java b/...c/main/java/org/apache/hadoop/hdfs/server/federation/resolver/ActiveNamenodeResolver.java
@@ -153,6 +153,7 @@ List<? extends FederationNamenodeContext> getNamenodesForNameserviceId(
    *
    * @param nsId name service id
    * @param namenode namenode contexts
+   * @param listObserversFirst Observer read case, observer NN will be ranked first
    */
-  void rotateCache(String nsId, FederationNamenodeContext namenode);
+  void rotateCache(String nsId, FederationNamenodeContext namenode, boolean listObserversFirst);
 }
diff --git a/...in/java/org/apache/hadoop/hdfs/server/federation/resolver/MembershipNamenodeResolver.java b/...in/java/org/apache/hadoop/hdfs/server/federation/resolver/MembershipNamenodeResolver.java
@@ -485,10 +485,12 @@ public void setRouterId(String router) {
    *
    * @param nsId name service id
    * @param namenode namenode contexts
+   * @param listObserversFirst Observer read case, observer NN will be ranked first
    */
   @Override
-  public synchronized void rotateCache(String nsId, FederationNamenodeContext namenode) {
-    cacheNS.compute(Pair.of(nsId, false), (ns, namenodeContexts) -> {
+  public synchronized void rotateCache(
+          String nsId, FederationNamenodeContext namenode, boolean listObserversFirst) {
+    cacheNS.compute(Pair.of(nsId, listObserversFirst), (ns, namenodeContexts) -> {
       if (namenodeContexts == null || namenodeContexts.size() <= 1) {
         return namenodeContexts;
       }
@@ -509,9 +511,8 @@ public synchronized void rotateCache(String nsId, FederationNamenodeContext name
         List<FederationNamenodeContext> rotatedNnContexts = new ArrayList<>(namenodeContexts);
         Collections.rotate(rotatedNnContexts, -1);
         return rotatedNnContexts;
-      }else {
-        return namenodeContexts;
       }
+      return namenodeContexts;
     });
   }
 }
diff --git a/...fs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcClient.java b/...fs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcClient.java
@@ -600,7 +600,7 @@ public Object invokeMethod(
           LOG.error("Cannot get available namenode for {} {} error: {}",
               nsId, rpcAddress, ioe.getMessage());
           if (this.namenodeResolver != null) {
-            this.namenodeResolver.rotateCache(nsId, namenode);
+            this.namenodeResolver.rotateCache(nsId, namenode, shouldUseObserver);
           }
           // Throw RetriableException so that client can retry
           throw new RetriableException(ioe);

diff --git a/.../hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/MockResolver.java b/.../hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/MockResolver.java
@@ -398,7 +398,8 @@ public void setRouterId(String router) {
   }
 
   @Override
-  public void rotateCache(String nsId, FederationNamenodeContext namenode) {
+  public void rotateCache(
+          String nsId, FederationNamenodeContext namenode, boolean listObserversFirst) {
   }
 
   /**