HDFS-13536. [PROVIDED Storage] HA for InMemoryAliasMap. Contributed by Virajith Jalaparti.

Author: Inigo Goiri

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

@@ -396,7 +396,7 @@ static String concatSuffixes(String... suffixes) {
    * @param keys Set of keys to look for in the order of preference
    * @return a map(nameserviceId to map(namenodeId to InetSocketAddress))
    */
-  static Map<String, Map<String, InetSocketAddress>> getAddresses(
+  public static Map<String, Map<String, InetSocketAddress>> getAddresses(
       Configuration conf, String defaultAddress, String... keys) {
     Collection<String> nameserviceIds = getNameServiceIds(conf);
     return getAddressesForNsIds(conf, nameserviceIds, defaultAddress, keys);
@@ -426,7 +426,7 @@ static Map<String, Map<String, InetSocketAddress>> getAddressesForNsIds(
     return ret;
   }
 
-  static Map<String, InetSocketAddress> getAddressesForNameserviceId(
+  public static Map<String, InetSocketAddress> getAddressesForNameserviceId(
       Configuration conf, String nsId, String defaultValue, String... keys) {
     Collection<String> nnIds = getNameNodeIds(conf, nsId);
     Map<String, InetSocketAddress> ret = Maps.newLinkedHashMap();

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

@@ -184,6 +184,9 @@ public interface HdfsClientConfigKeys {
       "dfs.namenode.snapshot.capture.openfiles";
   boolean DFS_NAMENODE_SNAPSHOT_CAPTURE_OPENFILES_DEFAULT = false;
 
+  String DFS_PROVIDED_ALIASMAP_INMEMORY_RPC_ADDRESS =
+      "dfs.provided.aliasmap.inmemory.dnrpc-address";
+
   /**
    * These are deprecated config keys to client code.
    */

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

@@ -37,6 +37,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import static org.apache.hadoop.hdfs.client.HdfsClientConfigKeys.DFS_NAMENODE_RPC_ADDRESS_KEY;
+
 /**
  * A FailoverProxyProvider implementation which allows one to configure
  * multiple URIs to connect to during fail-over. A random configured address is
@@ -60,6 +62,11 @@ public class ConfiguredFailoverProxyProvider<T> extends
 
   public ConfiguredFailoverProxyProvider(Configuration conf, URI uri,
       Class<T> xface, HAProxyFactory<T> factory) {
+    this(conf, uri, xface, factory, DFS_NAMENODE_RPC_ADDRESS_KEY);
+  }
+
+  public ConfiguredFailoverProxyProvider(Configuration conf, URI uri,
+      Class<T> xface, HAProxyFactory<T> factory, String addressKey) {
     this.xface = xface;
     this.conf = new Configuration(conf);
     int maxRetries = this.conf.getInt(
@@ -81,7 +88,7 @@ public ConfiguredFailoverProxyProvider(Configuration conf, URI uri,
       ugi = UserGroupInformation.getCurrentUser();
 
       Map<String, Map<String, InetSocketAddress>> map =
-          DFSUtilClient.getHaNnRpcAddresses(conf);
+          DFSUtilClient.getAddresses(conf, null, addressKey);
       Map<String, InetSocketAddress> addressesInNN = map.get(uri.getHost());
 
       if (addressesInNN == null || addressesInNN.size() == 0) {

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

@@ -0,0 +1,38 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs.server.namenode.ha;
+
+import org.apache.hadoop.conf.Configuration;
+
+import java.net.URI;
+
+import static org.apache.hadoop.hdfs.client.HdfsClientConfigKeys.DFS_PROVIDED_ALIASMAP_INMEMORY_RPC_ADDRESS;
+
+/**
+ * A {@link ConfiguredFailoverProxyProvider} implementation used to connect
+ * to an InMemoryAliasMap.
+ */
+public class InMemoryAliasMapFailoverProxyProvider<T>
+    extends ConfiguredFailoverProxyProvider<T> {
+
+  public InMemoryAliasMapFailoverProxyProvider(
+      Configuration conf, URI uri, Class<T> xface, HAProxyFactory<T> factory) {
+    super(conf, uri, xface, factory,
+        DFS_PROVIDED_ALIASMAP_INMEMORY_RPC_ADDRESS);
+  }
+}

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java

@@ -86,8 +86,11 @@ public class DFSConfigKeys extends CommonConfigurationKeys {
       HdfsClientConfigKeys.DeprecatedKeys.DFS_NAMENODE_BACKUP_HTTP_ADDRESS_KEY;
   public static final String  DFS_NAMENODE_BACKUP_HTTP_ADDRESS_DEFAULT = "0.0.0.0:50105";
   public static final String  DFS_NAMENODE_BACKUP_SERVICE_RPC_ADDRESS_KEY = "dfs.namenode.backup.dnrpc-address";
-  public static final String DFS_PROVIDED_ALIASMAP_INMEMORY_RPC_ADDRESS = "dfs.provided.aliasmap.inmemory.dnrpc-address";
+  public static final String DFS_PROVIDED_ALIASMAP_INMEMORY_RPC_ADDRESS =
+      HdfsClientConfigKeys.DFS_PROVIDED_ALIASMAP_INMEMORY_RPC_ADDRESS;
   public static final String DFS_PROVIDED_ALIASMAP_INMEMORY_RPC_ADDRESS_DEFAULT = "0.0.0.0:50200";
+  public static final String DFS_PROVIDED_ALIASMAP_INMEMORY_RPC_BIND_HOST = "dfs.provided.aliasmap.inmemory.rpc.bind-host";
+
   public static final String DFS_PROVIDED_ALIASMAP_INMEMORY_LEVELDB_DIR = "dfs.provided.aliasmap.inmemory.leveldb.dir";
   public static final String DFS_PROVIDED_ALIASMAP_INMEMORY_BATCH_SIZE = "dfs.provided.aliasmap.inmemory.batch-size";
   public static final int DFS_PROVIDED_ALIASMAP_INMEMORY_BATCH_SIZE_DEFAULT = 500;

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSUtil.java

@@ -1130,7 +1130,42 @@ private static String getNameServiceId(Configuration conf, String addressKey) {
 
     return getSuffixIDs(conf, addressKey, null, nnId, LOCAL_ADDRESS_MATCHER)[0];
   }
-  
+
+  /**
+   * Determine the {@link InetSocketAddress} to bind to, for any service.
+   * In case of HA or federation, the address is assumed to specified as
+   * {@code confKey}.NAMESPACEID.NAMENODEID as appropriate.
+   *
+   * @param conf configuration.
+   * @param confKey configuration key (prefix if HA/federation) used to
+   *        specify the address for the service.
+   * @param defaultValue default value for the address.
+   * @param bindHostKey configuration key (prefix if HA/federation)
+   *        specifying host to bind to.
+   * @return the address to bind to.
+   */
+  public static InetSocketAddress getBindAddress(Configuration conf,
+      String confKey, String defaultValue, String bindHostKey) {
+    InetSocketAddress address;
+    String nsId = DFSUtil.getNamenodeNameServiceId(conf);
+    String bindHostActualKey;
+    if (nsId != null) {
+      String namenodeId = HAUtil.getNameNodeId(conf, nsId);
+      address = DFSUtilClient.getAddressesForNameserviceId(
+          conf, nsId, null, confKey).get(namenodeId);
+      bindHostActualKey = DFSUtil.addKeySuffixes(bindHostKey, nsId, namenodeId);
+    } else {
+      address = NetUtils.createSocketAddr(conf.get(confKey, defaultValue));
+      bindHostActualKey = bindHostKey;
+    }
+
+    String bindHost = conf.get(bindHostActualKey);
+    if (bindHost == null || bindHost.isEmpty()) {
+      bindHost = address.getHostName();
+    }
+    return new InetSocketAddress(bindHost, address.getPort());
+  }
+
   /**
    * Returns nameservice Id and namenode Id when the local host matches the
    * configuration parameter {@code addressKey}.<nameservice Id>.<namenode Id>

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/NameNodeProxies.java

@@ -31,10 +31,13 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hdfs.NameNodeProxiesClient.ProxyAndInfo;
 import org.apache.hadoop.hdfs.protocol.ClientProtocol;
+import org.apache.hadoop.hdfs.protocolPB.AliasMapProtocolPB;
+import org.apache.hadoop.hdfs.protocolPB.InMemoryAliasMapProtocolClientSideTranslatorPB;
 import org.apache.hadoop.hdfs.protocolPB.JournalProtocolPB;
 import org.apache.hadoop.hdfs.protocolPB.JournalProtocolTranslatorPB;
 import org.apache.hadoop.hdfs.protocolPB.NamenodeProtocolPB;
 import org.apache.hadoop.hdfs.protocolPB.NamenodeProtocolTranslatorPB;
+import org.apache.hadoop.hdfs.server.aliasmap.InMemoryAliasMapProtocol;
 import org.apache.hadoop.hdfs.server.namenode.ha.AbstractNNFailoverProxyProvider;
 import org.apache.hadoop.hdfs.server.namenode.ha.NameNodeHAProxyFactory;
 import org.apache.hadoop.hdfs.server.protocol.JournalProtocol;
@@ -184,6 +187,8 @@ public static <T> ProxyAndInfo<T> createNonHAProxy(
           conf, ugi);
     } else if (xface == RefreshCallQueueProtocol.class) {
       proxy = (T) createNNProxyWithRefreshCallQueueProtocol(nnAddr, conf, ugi);
+    } else if (xface == InMemoryAliasMapProtocol.class) {
+      proxy = (T) createNNProxyWithInMemoryAliasMapProtocol(nnAddr, conf, ugi);
     } else {
       String message = "Unsupported protocol found when creating the proxy " +
           "connection to NameNode: " +
@@ -194,7 +199,15 @@ public static <T> ProxyAndInfo<T> createNonHAProxy(
 
     return new ProxyAndInfo<T>(proxy, dtService, nnAddr);
   }
-  
+
+  private static InMemoryAliasMapProtocol createNNProxyWithInMemoryAliasMapProtocol(
+      InetSocketAddress address, Configuration conf, UserGroupInformation ugi)
+      throws IOException {
+    AliasMapProtocolPB proxy = (AliasMapProtocolPB) createNameNodeProxy(
+        address, conf, ugi, AliasMapProtocolPB.class, 30000);
+    return new InMemoryAliasMapProtocolClientSideTranslatorPB(proxy);
+  }
+
   private static JournalProtocol createNNProxyWithJournalProtocol(
       InetSocketAddress address, Configuration conf, UserGroupInformation ugi)
       throws IOException {

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/InMemoryAliasMapProtocolClientSideTranslatorPB.java

@@ -20,27 +20,38 @@
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.HAUtil;
+import org.apache.hadoop.hdfs.NameNodeProxies;
 import org.apache.hadoop.hdfs.protocol.Block;
+import org.apache.hadoop.hdfs.protocol.HdfsConstants;
 import org.apache.hadoop.hdfs.protocol.ProvidedStorageLocation;
 import org.apache.hadoop.hdfs.server.aliasmap.InMemoryAliasMap;
 import org.apache.hadoop.hdfs.server.aliasmap.InMemoryAliasMapProtocol;
 import org.apache.hadoop.hdfs.server.common.FileRegion;
+import org.apache.hadoop.hdfs.server.namenode.ha.AbstractNNFailoverProxyProvider;
+import org.apache.hadoop.hdfs.server.namenode.ha.InMemoryAliasMapFailoverProxyProvider;
 import org.apache.hadoop.ipc.ProtobufHelper;
-import org.apache.hadoop.ipc.ProtobufRpcEngine;
 import org.apache.hadoop.ipc.RPC;
 import org.apache.hadoop.net.NetUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import javax.annotation.Nonnull;
+import java.io.Closeable;
 import java.io.IOException;
-import java.net.InetSocketAddress;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.ArrayList;
+import java.util.Collection;
 import java.util.List;
 import java.util.Optional;
 import java.util.stream.Collectors;
 
 import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_PROVIDED_ALIASMAP_INMEMORY_RPC_ADDRESS;
-import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_PROVIDED_ALIASMAP_INMEMORY_RPC_ADDRESS_DEFAULT;
+import static org.apache.hadoop.hdfs.DFSUtil.addKeySuffixes;
+import static org.apache.hadoop.hdfs.DFSUtil.createUri;
+import static org.apache.hadoop.hdfs.DFSUtilClient.getNameServiceIds;
+import static org.apache.hadoop.hdfs.client.HdfsClientConfigKeys.Failover.PROXY_PROVIDER_KEY_PREFIX;
 import static org.apache.hadoop.hdfs.protocol.proto.AliasMapProtocolProtos.*;
 import static org.apache.hadoop.hdfs.protocol.proto.HdfsProtos.*;
 
@@ -52,30 +63,69 @@
 @InterfaceAudience.Private
 @InterfaceStability.Unstable
 public class InMemoryAliasMapProtocolClientSideTranslatorPB
-    implements InMemoryAliasMapProtocol {
+    implements InMemoryAliasMapProtocol, Closeable {
 
   private static final Logger LOG =
       LoggerFactory
           .getLogger(InMemoryAliasMapProtocolClientSideTranslatorPB.class);
 
   private AliasMapProtocolPB rpcProxy;
 
-  public InMemoryAliasMapProtocolClientSideTranslatorPB(Configuration conf) {
-    String addr = conf.getTrimmed(DFS_PROVIDED_ALIASMAP_INMEMORY_RPC_ADDRESS,
-        DFS_PROVIDED_ALIASMAP_INMEMORY_RPC_ADDRESS_DEFAULT);
-    InetSocketAddress aliasMapAddr = NetUtils.createSocketAddr(addr);
+  public InMemoryAliasMapProtocolClientSideTranslatorPB(
+      AliasMapProtocolPB rpcProxy) {
+    this.rpcProxy = rpcProxy;
+  }
 
-    RPC.setProtocolEngine(conf, AliasMapProtocolPB.class,
-        ProtobufRpcEngine.class);
-    LOG.info("Connecting to address: " + addr);
-    try {
-      rpcProxy = RPC.getProxy(AliasMapProtocolPB.class,
-          RPC.getProtocolVersion(AliasMapProtocolPB.class), aliasMapAddr, null,
-          conf, NetUtils.getDefaultSocketFactory(conf), 0);
-    } catch (IOException e) {
-      throw new RuntimeException(
-          "Error in connecting to " + addr + " Got: " + e);
+  public static Collection<InMemoryAliasMapProtocol> init(Configuration conf) {
+    Collection<InMemoryAliasMapProtocol> aliasMaps = new ArrayList<>();
+    // Try to connect to all configured nameservices as it is not known which
+    // nameservice supports the AliasMap.
+    for (String nsId : getNameServiceIds(conf)) {
+      try {
+        URI namenodeURI = null;
+        Configuration newConf = new Configuration(conf);
+        if (HAUtil.isHAEnabled(conf, nsId)) {
+          // set the failover-proxy provider if HA is enabled.
+          newConf.setClass(
+              addKeySuffixes(PROXY_PROVIDER_KEY_PREFIX, nsId),
+              InMemoryAliasMapFailoverProxyProvider.class,
+              AbstractNNFailoverProxyProvider.class);
+          namenodeURI = new URI(HdfsConstants.HDFS_URI_SCHEME + "://" + nsId);
+        } else {
+          String key =
+              addKeySuffixes(DFS_PROVIDED_ALIASMAP_INMEMORY_RPC_ADDRESS, nsId);
+          String addr = conf.get(key);
+          if (addr != null) {
+            namenodeURI = createUri(HdfsConstants.HDFS_URI_SCHEME,
+                NetUtils.createSocketAddr(addr));
+          }
+        }
+        if (namenodeURI != null) {
+          aliasMaps.add(NameNodeProxies
+              .createProxy(newConf, namenodeURI, InMemoryAliasMapProtocol.class)
+              .getProxy());
+          LOG.info("Connected to InMemoryAliasMap at {}", namenodeURI);
+        }
+      } catch (IOException | URISyntaxException e) {
+        LOG.warn("Exception in connecting to InMemoryAliasMap for nameservice "
+            + "{}: {}", nsId, e);
+      }
     }
+    // if a separate AliasMap is configured using
+    // DFS_PROVIDED_ALIASMAP_INMEMORY_RPC_ADDRESS, try to connect it.
+    if (conf.get(DFS_PROVIDED_ALIASMAP_INMEMORY_RPC_ADDRESS) != null) {
+      URI uri = createUri("hdfs", NetUtils.createSocketAddr(
+          conf.get(DFS_PROVIDED_ALIASMAP_INMEMORY_RPC_ADDRESS)));
+      try {
+        aliasMaps.add(NameNodeProxies
+            .createProxy(conf, uri, InMemoryAliasMapProtocol.class).getProxy());
+        LOG.info("Connected to InMemoryAliasMap at {}", uri);
+      } catch (IOException e) {
+        LOG.warn("Exception in connecting to InMemoryAliasMap at {}: {}", uri,
+            e);
+      }
+    }
+    return aliasMaps;
   }
 
   @Override
@@ -168,7 +218,12 @@ public String getBlockPoolId() throws IOException {
     }
   }
 
-  public void stop() {
-    RPC.stopProxy(rpcProxy);
+  @Override
+  public void close() throws IOException {
+    LOG.info("Stopping rpcProxy in" +
+        "InMemoryAliasMapProtocolClientSideTranslatorPB");
+    if (rpcProxy != null) {
+      RPC.stopProxy(rpcProxy);
+    }
   }
 }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/aliasmap/InMemoryAliasMapProtocol.java

@@ -21,6 +21,7 @@
 import org.apache.hadoop.hdfs.protocol.Block;
 import org.apache.hadoop.hdfs.protocol.ProvidedStorageLocation;
 import org.apache.hadoop.hdfs.server.common.FileRegion;
+import org.apache.hadoop.io.retry.Idempotent;
 
 import javax.annotation.Nonnull;
 import java.io.IOException;
@@ -69,6 +70,7 @@ public Optional<Block> getNextBlock() {
    * FileRegions and the next marker.
    * @throws IOException
    */
+  @Idempotent
   InMemoryAliasMap.IterationResult list(Optional<Block> marker)
       throws IOException;
 
@@ -80,6 +82,7 @@ InMemoryAliasMap.IterationResult list(Optional<Block> marker)
    * @throws IOException
    */
   @Nonnull
+  @Idempotent
   Optional<ProvidedStorageLocation> read(@Nonnull Block block)
       throws IOException;
 
@@ -90,6 +93,7 @@ Optional<ProvidedStorageLocation> read(@Nonnull Block block)
    * @param providedStorageLocation
    * @throws IOException
    */
+  @Idempotent
   void write(@Nonnull Block block,
       @Nonnull ProvidedStorageLocation providedStorageLocation)
       throws IOException;
@@ -99,5 +103,6 @@ void write(@Nonnull Block block,
    * @return the block pool id associated with the Namenode running
    * the in-memory alias map.
    */
+  @Idempotent
   String getBlockPoolId() throws IOException;
 }