HDFS-17588. RBF: RouterObserverReadProxyProvider should perform an msync before executing the first read.

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

@@ -82,6 +82,15 @@ public class RouterObserverReadProxyProvider<T> extends AbstractNNFailoverProxyP
    */
   private volatile long lastMsyncTimeMs = -1;
 
+  /**
+   * A client using RouterObserverReadProxyProvider should first sync with the
+   * Active NameNode on startup. This ensures that the client reads data which
+   * is consistent with the state of the world as of the time of its
+   * instantiation. This variable will be true after this initial sync has
+   * been performed.
+   */
+  private volatile boolean msynced = false;
+
   public RouterObserverReadProxyProvider(Configuration conf, URI uri, Class<T> xface,
       HAProxyFactory<T> factory) {
     this(conf, uri, xface, factory, new IPFailoverProxyProvider<>(conf, uri, xface, factory));
@@ -154,6 +163,23 @@ private ClientProtocol getProxyAsClientProtocol(T proxy) {
     return (ClientProtocol) proxy;
   }
 
+  /**
+   * This will call {@link ClientProtocol#msync()} on the active NameNode
+   * (via the {@link #innerProxy}) to initialize the state of this client.
+   * Calling it multiple times is a no-op; only the first will perform an
+   * msync.
+   *
+   * @see #msynced
+   */
+  private synchronized void initializeMsync() throws IOException {
+    if (msynced) {
+      return; // No need for an msync
+    }
+    getProxyAsClientProtocol(innerProxy.getProxy().proxy).msync();
+    msynced = true;
+    lastMsyncTimeMs = Time.monotonicNow();
+  }
+
   /**
    * This will call {@link ClientProtocol#msync()} on the active NameNode
    * (via the {@link #innerProxy}) to update the state of this client, only
@@ -209,7 +235,13 @@ public void close() throws IOException {
     @Override
     public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
       if (observerReadEnabled && isRead(method)) {
-        autoMsyncIfNecessary();
+        if (!msynced) {
+          // An msync() must first be performed to ensure that this client is
+          // up-to-date with the active's state. This will only be done once.
+          initializeMsync();
+        } else {
+          autoMsyncIfNecessary();
+        }
       }
 
       try {

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

@@ -815,6 +815,55 @@ public void testPeriodicStateRefreshUsingActiveNamenode(ConfigSetting configSett
         initialLengthOfRootListing + 10, rootFolderAfterMkdir.length);
   }
 
+  @EnumSource(ConfigSetting.class)
+  @ParameterizedTest
+  public void testAutoMsyncDefault(ConfigSetting configSetting) throws Exception {
+    Configuration clientConfiguration = getConfToEnableObserverReads(configSetting);
+    fileSystem = routerContext.getFileSystem(clientConfiguration);
+
+    List<? extends FederationNamenodeContext> namenodes = routerContext
+        .getRouter().getNamenodeResolver()
+        .getNamenodesForNameserviceId(cluster.getNameservices().get(0), true);
+    assertEquals("First namenode should be observer", namenodes.get(0).getState(),
+        FederationNamenodeServiceState.OBSERVER);
+    Path path = new Path("/");
+
+    long rpcCountForActive;
+    long rpcCountForObserver;
+
+    // Send read requests
+    int numListings = 15;
+    for (int i = 0; i < numListings; i++) {
+      fileSystem.listFiles(path, false);
+    }
+    fileSystem.close();
+
+    rpcCountForActive = routerContext.getRouter().getRpcServer()
+        .getRPCMetrics().getActiveProxyOps();
+
+    rpcCountForObserver = routerContext.getRouter().getRpcServer()
+        .getRPCMetrics().getObserverProxyOps();
+
+    switch (configSetting) {
+      case USE_NAMENODE_PROXY_FLAG:
+        // First read goes to active.
+        assertEquals("Calls sent to the active", 1, rpcCountForActive);
+        // The rest of the reads are sent to the observer.
+        assertEquals("Reads sent to observer", numListings - 1, rpcCountForObserver);
+        break;
+      case USE_ROUTER_OBSERVER_READ_PROXY_PROVIDER:
+      case USE_ROUTER_OBSERVER_READ_CONFIGURED_FAILOVER_PROXY_PROVIDER:
+        // An msync is sent to each active namenode.
+        assertEquals("Msyncs sent to the active namenodes",
+            NUM_NAMESERVICES * 1, rpcCountForActive);
+        // All reads should be sent of the observer.
+        assertEquals("Reads sent to observer", numListings, rpcCountForObserver);
+        break;
+      default:
+        Assertions.fail("Unknown config setting: " + configSetting);
+    }
+  }
+
   @EnumSource(ConfigSetting.class)
   @ParameterizedTest
   public void testAutoMsyncEqualsZero(ConfigSetting configSetting) throws Exception {