Merge branch 'apache:trunk' into YARN-11350-V2

Author: slfan1989

LICENSE-binary

@@ -244,7 +244,7 @@ com.nimbusds:nimbus-jose-jwt:9.8.1
 com.squareup.okhttp3:okhttp:4.10.0
 com.squareup.okio:okio:3.2.0
 com.zaxxer:HikariCP:4.0.3
-commons-beanutils:commons-beanutils:1.9.3
+commons-beanutils:commons-beanutils:1.9.4
 commons-cli:commons-cli:1.2
 commons-codec:commons-codec:1.11
 commons-collections:commons-collections:3.2.2

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/VectoredReadUtils.java

@@ -307,9 +307,16 @@ public static ByteBuffer sliceTo(ByteBuffer readData, long readOffset,
                                    FileRange request) {
     int offsetChange = (int) (request.getOffset() - readOffset);
     int requestLength = request.getLength();
+    // Create a new buffer that is backed by the original contents
+    // The buffer will have position 0 and the same limit as the original one
     readData = readData.slice();
+    // Change the offset and the limit of the buffer as the reader wants to see
+    // only relevant data
     readData.position(offsetChange);
     readData.limit(offsetChange + requestLength);
+    // Create a new buffer after the limit change so that only that portion of the data is
+    // returned to the reader.
+    readData = readData.slice();
     return readData;
   }
 

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/TestVectoredReadUtils.java

@@ -61,6 +61,9 @@ public void testSliceTo() {
             .describedAs("Slicing on the same offset shouldn't " +
                     "create a new buffer")
             .isEqualTo(slice);
+    Assertions.assertThat(slice.position())
+        .describedAs("Slicing should return buffers starting from position 0")
+        .isEqualTo(0);
 
     // try slicing a range
     final int offset = 100;
@@ -77,6 +80,9 @@ public void testSliceTo() {
             .describedAs("Slicing should use the same underlying " +
                     "data")
             .isEqualTo(slice.array());
+    Assertions.assertThat(slice.position())
+        .describedAs("Slicing should return buffers starting from position 0")
+        .isEqualTo(0);
     // test the contents of the slice
     intBuffer = slice.asIntBuffer();
     for(int i=0; i < sliceLength / Integer.BYTES; ++i) {

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/NameNodeProxiesClient.java

@@ -349,6 +349,12 @@ public static ClientProtocol createProxyWithAlignmentContext(
       boolean withRetries, AtomicBoolean fallbackToSimpleAuth,
       AlignmentContext alignmentContext)
       throws IOException {
+    if (alignmentContext == null &&
+        conf.getBoolean(HdfsClientConfigKeys.DFS_RBF_OBSERVER_READ_ENABLE,
+            HdfsClientConfigKeys.DFS_RBF_OBSERVER_READ_ENABLE_DEFAULT)) {
+      alignmentContext = new ClientGSIContext();
+    }
+
     RPC.setProtocolEngine(conf, ClientNamenodeProtocolPB.class,
         ProtobufRpcEngine2.class);
 

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/client/HdfsClientConfigKeys.java

@@ -78,6 +78,8 @@ public interface HdfsClientConfigKeys {
   int     DFS_NAMENODE_HTTPS_PORT_DEFAULT = 9871;
   String  DFS_NAMENODE_HTTPS_ADDRESS_KEY = "dfs.namenode.https-address";
   String DFS_HA_NAMENODES_KEY_PREFIX = "dfs.ha.namenodes";
+  String DFS_RBF_OBSERVER_READ_ENABLE = "dfs.client.rbf.observer.read.enable";
+  boolean DFS_RBF_OBSERVER_READ_ENABLE_DEFAULT = false;
   int DFS_NAMENODE_RPC_PORT_DEFAULT = 8020;
   String DFS_NAMENODE_KERBEROS_PRINCIPAL_KEY =
       "dfs.namenode.kerberos.principal";

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

@@ -226,13 +226,14 @@ public ConnectionContext getConnection(UserGroupInformation ugi,
           this.pools.put(connectionId, pool);
           this.connectionPoolToNamespaceMap.put(connectionId, nsId);
         }
-        long clientStateId = RouterStateIdContext.getClientStateIdFromCurrentCall(nsId);
-        pool.getPoolAlignmentContext().advanceClientStateId(clientStateId);
       } finally {
         writeLock.unlock();
       }
     }
 
+    long clientStateId = RouterStateIdContext.getClientStateIdFromCurrentCall(nsId);
+    pool.getPoolAlignmentContext().advanceClientStateId(clientStateId);
+
     ConnectionContext conn = pool.getConnection();
 
     // Add a new connection to the pool if it wasn't usable

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

@@ -20,6 +20,8 @@
 
 import java.io.IOException;
 import java.util.concurrent.atomic.LongAccumulator;
+
+import org.apache.hadoop.classification.VisibleForTesting;
 import org.apache.hadoop.ipc.AlignmentContext;
 import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos;
 
@@ -99,4 +101,9 @@ public boolean isCoordinatedCall(String protocolName, String method) {
   public void advanceClientStateId(Long clientStateId) {
     poolLocalStateId.accumulate(clientStateId);
   }
+
+  @VisibleForTesting
+  public long getPoolLocalStateId() {
+    return this.poolLocalStateId.get();
+  }
 }

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/StateStoreService.java

@@ -272,6 +272,15 @@ public <T extends RecordStore<?>> T getRegisteredRecordStore(
     return null;
   }
 
+  /**
+   * Get the list of all RecordStores.
+   * @return a list of each RecordStore.
+   */
+  @SuppressWarnings("unchecked")
+  public <T extends RecordStore<? extends BaseRecord>> List<T> getRecordStores() {
+    return new ArrayList<>((Collection<T>) recordStores.values());
+  }
+
   /**
    * List of records supported by this State Store.
    *

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreFileBaseImpl.java

@@ -85,7 +85,8 @@ protected abstract <T extends BaseRecord> BufferedReader getReader(
    * @param path Path of the record to write.
    * @return Writer for the record.
    */
-  protected abstract <T extends BaseRecord> BufferedWriter getWriter(
+  @VisibleForTesting
+  public abstract <T extends BaseRecord> BufferedWriter getWriter(
       String path);
 
   /**
@@ -348,25 +349,18 @@ public <T extends BaseRecord> boolean putAll(
     for (Entry<String, T> entry : toWrite.entrySet()) {
       String recordPath = entry.getKey();
       String recordPathTemp = recordPath + "." + now() + TMP_MARK;
-      BufferedWriter writer = getWriter(recordPathTemp);
-      try {
+      boolean recordWrittenSuccessfully = true;
+      try (BufferedWriter writer = getWriter(recordPathTemp)) {
         T record = entry.getValue();
         String line = serializeString(record);
         writer.write(line);
       } catch (IOException e) {
         LOG.error("Cannot write {}", recordPathTemp, e);
+        recordWrittenSuccessfully = false;
         success = false;
-      } finally {
-        if (writer != null) {
-          try {
-            writer.close();
-          } catch (IOException e) {
-            LOG.error("Cannot close the writer for {}", recordPathTemp, e);
-          }
-        }
       }
       // Commit
-      if (!rename(recordPathTemp, recordPath)) {
+      if (recordWrittenSuccessfully && !rename(recordPathTemp, recordPath)) {
         LOG.error("Failed committing record into {}", recordPath);
         success = false;
       }

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreFileImpl.java

@@ -31,6 +31,7 @@
 import java.util.List;
 
 import org.apache.commons.lang3.ArrayUtils;
+import org.apache.hadoop.classification.VisibleForTesting;
 import org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys;
 import org.apache.hadoop.hdfs.server.federation.store.records.BaseRecord;
 import org.slf4j.Logger;
@@ -125,7 +126,8 @@ protected <T extends BaseRecord> BufferedReader getReader(String filename) {
   }
 
   @Override
-  protected <T extends BaseRecord> BufferedWriter getWriter(String filename) {
+  @VisibleForTesting
+  public <T extends BaseRecord> BufferedWriter getWriter(String filename) {
     BufferedWriter writer = null;
     try {
       LOG.debug("Writing file: {}", filename);

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreFileSystemImpl.java

@@ -28,13 +28,14 @@
 import java.util.Collections;
 import java.util.List;
 
+import org.apache.hadoop.classification.VisibleForTesting;
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.FileUtil;
 import org.apache.hadoop.fs.Options;
 import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hdfs.DistributedFileSystem;
 import org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys;
 import org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreDriver;
 import org.apache.hadoop.hdfs.server.federation.store.records.BaseRecord;
@@ -82,17 +83,8 @@ protected boolean mkdir(String path) {
   @Override
   protected boolean rename(String src, String dst) {
     try {
-      if (fs instanceof DistributedFileSystem) {
-        DistributedFileSystem dfs = (DistributedFileSystem)fs;
-        dfs.rename(new Path(src), new Path(dst), Options.Rename.OVERWRITE);
-        return true;
-      } else {
-        // Replace should be atomic but not available
-        if (fs.exists(new Path(dst))) {
-          fs.delete(new Path(dst), true);
-        }
-        return fs.rename(new Path(src), new Path(dst));
-      }
+      FileUtil.rename(fs, new Path(src), new Path(dst), Options.Rename.OVERWRITE);
+      return true;
     } catch (Exception e) {
       LOG.error("Cannot rename {} to {}", src, dst, e);
       return false;
@@ -148,7 +140,8 @@ protected <T extends BaseRecord> BufferedReader getReader(String pathName) {
   }
 
   @Override
-  protected <T extends BaseRecord> BufferedWriter getWriter(String pathName) {
+  @VisibleForTesting
+  public <T extends BaseRecord> BufferedWriter getWriter(String pathName) {
     BufferedWriter writer = null;
     Path path = new Path(pathName);
     try {

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/tools/federation/RouterAdmin.java

@@ -18,6 +18,7 @@
 package org.apache.hadoop.hdfs.tools.federation;
 
 import java.io.IOException;
+import java.io.PrintStream;
 import java.net.InetSocketAddress;
 import java.util.Arrays;
 import java.util.Collection;
@@ -26,6 +27,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.TreeMap;
 import java.util.regex.Pattern;
 
 import org.apache.hadoop.classification.InterfaceAudience.Private;
@@ -46,6 +48,10 @@
 import org.apache.hadoop.hdfs.server.federation.router.RouterClient;
 import org.apache.hadoop.hdfs.server.federation.router.RouterQuotaUsage;
 import org.apache.hadoop.hdfs.server.federation.router.RouterStateManager;
+import org.apache.hadoop.hdfs.server.federation.store.CachedRecordStore;
+import org.apache.hadoop.hdfs.server.federation.store.RecordStore;
+import org.apache.hadoop.hdfs.server.federation.store.StateStoreService;
+import org.apache.hadoop.hdfs.server.federation.store.StateStoreUtils;
 import org.apache.hadoop.hdfs.server.federation.store.protocol.AddMountTableEntryRequest;
 import org.apache.hadoop.hdfs.server.federation.store.protocol.AddMountTableEntryResponse;
 import org.apache.hadoop.hdfs.server.federation.store.protocol.DisableNameserviceRequest;
@@ -70,7 +76,9 @@
 import org.apache.hadoop.hdfs.server.federation.store.protocol.RemoveMountTableEntryResponse;
 import org.apache.hadoop.hdfs.server.federation.store.protocol.UpdateMountTableEntryRequest;
 import org.apache.hadoop.hdfs.server.federation.store.protocol.UpdateMountTableEntryResponse;
+import org.apache.hadoop.hdfs.server.federation.store.records.BaseRecord;
 import org.apache.hadoop.hdfs.server.federation.store.records.MountTable;
+import org.apache.hadoop.hdfs.server.federation.store.records.impl.pb.PBRecord;
 import org.apache.hadoop.ipc.ProtobufRpcEngine2;
 import org.apache.hadoop.ipc.RPC;
 import org.apache.hadoop.ipc.RefreshResponse;
@@ -97,6 +105,7 @@
 public class RouterAdmin extends Configured implements Tool {
 
   private static final Logger LOG = LoggerFactory.getLogger(RouterAdmin.class);
+  private static final String DUMP_COMMAND = "-dumpState";
 
   private RouterClient client;
 
@@ -133,7 +142,7 @@ private String getUsage(String cmd) {
       String[] commands =
           {"-add", "-update", "-rm", "-ls", "-getDestination", "-setQuota",
               "-setStorageTypeQuota", "-clrQuota", "-clrStorageTypeQuota",
-              "-safemode", "-nameservice", "-getDisabledNameservices",
+              DUMP_COMMAND, "-safemode", "-nameservice", "-getDisabledNameservices",
               "-refresh", "-refreshRouterArgs",
               "-refreshSuperUserGroupsConfiguration", "-refreshCallQueue"};
       StringBuilder usage = new StringBuilder();
@@ -187,6 +196,8 @@ private String getUsage(String cmd) {
       return "\t[-refreshSuperUserGroupsConfiguration]";
     } else if (cmd.equals("-refreshCallQueue")) {
       return "\t[-refreshCallQueue]";
+    } else if (cmd.equals(DUMP_COMMAND)) {
+      return "\t[" + DUMP_COMMAND + "]";
     }
     return getUsage(null);
   }
@@ -224,7 +235,8 @@ private void validateMax(String[] arg) {
       if (arg.length > 1) {
         throw new IllegalArgumentException("No arguments allowed");
       }
-    } else if (arg[0].equals("-refreshCallQueue")) {
+    } else if (arg[0].equals("-refreshCallQueue") ||
+        arg[0].equals(DUMP_COMMAND)) {
       if (arg.length > 1) {
         throw new IllegalArgumentException("No arguments allowed");
       }
@@ -286,6 +298,15 @@ private boolean validateMin(String[] argv) {
     return true;
   }
 
+  /**
+   * Does this command run in the local process?
+   * @param cmd the string of the command
+   * @return is this a local command?
+   */
+  boolean isLocalCommand(String cmd) {
+    return DUMP_COMMAND.equals(cmd);
+  }
+
   @Override
   public int run(String[] argv) throws Exception {
     if (argv.length < 1) {
@@ -303,6 +324,10 @@ public int run(String[] argv) throws Exception {
       System.err.println("Not enough parameters specificed for cmd " + cmd);
       printUsage(cmd);
       return exitCode;
+    } else if (isLocalCommand(argv[0])) {
+      if (DUMP_COMMAND.equals(argv[0])) {
+        return dumpStateStore(getConf(), System.out) ? 0 : -1;
+      }
     }
     String address = null;
     // Initialize RouterClient
@@ -1301,6 +1326,49 @@ private int refreshCallQueue() throws IOException {
     return returnCode;
   }
 
+  /**
+   * Dumps the contents of the StateStore to stdout.
+   * @return true if it was successful
+   */
+  public static boolean dumpStateStore(Configuration conf,
+                                PrintStream output) throws IOException {
+    StateStoreService service = new StateStoreService();
+    conf.setBoolean(RBFConfigKeys.DFS_ROUTER_METRICS_ENABLE, false);
+    service.init(conf);
+    service.loadDriver();
+    if (!service.isDriverReady()) {
+      System.err.println("Can't initialize driver");
+      return false;
+    }
+    // Get the stores sorted by name
+    Map<String, RecordStore<? extends BaseRecord>> stores = new TreeMap<>();
+    for(RecordStore<? extends BaseRecord> store: service.getRecordStores()) {
+      String recordName = StateStoreUtils.getRecordName(store.getRecordClass());
+      stores.put(recordName, store);
+    }
+    for (Entry<String, RecordStore<? extends BaseRecord>> pair: stores.entrySet()) {
+      String recordName = pair.getKey();
+      RecordStore<? extends BaseRecord> store = pair.getValue();
+      output.println("---- " + recordName + " ----");
+      if (store instanceof CachedRecordStore) {
+        for (Object record: ((CachedRecordStore) store).getCachedRecords()) {
+          if (record instanceof BaseRecord && record instanceof PBRecord) {
+            BaseRecord baseRecord = (BaseRecord) record;
+            // Generate the pseudo-json format of the protobuf record
+            String recordString = ((PBRecord) record).getProto().toString();
+            // Indent each line
+            recordString = "    " + recordString.replaceAll("\n", "\n    ");
+            output.println(String.format("  %s:", baseRecord.getPrimaryKey()));
+            output.println(recordString);
+          }
+        }
+        output.println();
+      }
+    }
+    service.stop();
+    return true;
+  }
+
   /**
    * Normalize a path for that filesystem.
    *
@@ -1341,4 +1409,4 @@ public FsPermission getMode() {
       return mode;
     }
   }
-}
+}

hadoop-hdfs-project/hadoop-hdfs-rbf/src/site/markdown/HDFSRouterFederation.md

@@ -328,6 +328,17 @@ To trigger a runtime-refresh of the resource specified by \<key\> on \<host:ipc\
 
     [hdfs]$ $HADOOP_HOME/bin/hdfs dfsrouteradmin -refreshRouterArgs <host:ipc_port> <key> [arg1..argn]
 
+### Router state dump
+
+To diagnose the current state of the routers, you can use the
+dumpState command. It generates a text dump of the records in the
+State Store. Since it uses the configuration to find and read the
+state store, it is often easiest to use the machine where the routers
+run. The command runs locally, so the routers do not have to be up to
+use this command.
+
+    [hdfs]$ $HADOOP_HOME/bin/hdfs dfsrouteradmin -dumpState
+
 Client configuration
 --------------------