You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by cl...@apache.org on 2018/12/14 18:42:15 UTC
hadoop git commit: HDFS-14116. ObserverReadProxyProvider should work
with protocols other than ClientProtocol. Contributed by Chao Sun.
Repository: hadoop
Updated Branches:
refs/heads/HDFS-12943 9a9e3663f -> 12bdbf2be
HDFS-14116. ObserverReadProxyProvider should work with protocols other than ClientProtocol. Contributed by Chao Sun.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/12bdbf2b
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/12bdbf2b
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/12bdbf2b
Branch: refs/heads/HDFS-12943
Commit: 12bdbf2be582cc74f720189421c60244c4bdd927
Parents: 9a9e366
Author: Chen Liang <cl...@apache.org>
Authored: Fri Dec 14 10:39:49 2018 -0800
Committer: Chen Liang <cl...@apache.org>
Committed: Fri Dec 14 10:39:49 2018 -0800
----------------------------------------------------------------------
.../hadoop/io/retry/FailoverProxyProvider.java | 15 +++++-
.../ha/AbstractNNFailoverProxyProvider.java | 48 +++++++++++++++++++-
.../namenode/ha/IPFailoverProxyProvider.java | 11 +----
.../namenode/ha/ObserverReadProxyProvider.java | 19 ++++----
.../org/apache/hadoop/hdfs/DFSTestUtil.java | 20 --------
.../server/namenode/NNThroughputBenchmark.java | 5 +-
.../ha/TestObserverReadProxyProvider.java | 18 ++++++++
7 files changed, 93 insertions(+), 43 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/12bdbf2b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/retry/FailoverProxyProvider.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/retry/FailoverProxyProvider.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/retry/FailoverProxyProvider.java
index f2fa3af..7a3e77e 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/retry/FailoverProxyProvider.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/retry/FailoverProxyProvider.java
@@ -18,8 +18,10 @@
package org.apache.hadoop.io.retry;
import java.io.Closeable;
+import java.io.IOException;
import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.ipc.RPC;
/**
* An implementer of this interface is capable of providing proxy objects for
@@ -30,7 +32,7 @@ import org.apache.hadoop.classification.InterfaceStability;
*/
@InterfaceStability.Evolving
public interface FailoverProxyProvider<T> extends Closeable {
- static class ProxyInfo<T> {
+ static class ProxyInfo<T> implements Closeable {
public T proxy;
/*
* The information (e.g., the IP address) of the current proxy object. It
@@ -55,6 +57,17 @@ public interface FailoverProxyProvider<T> extends Closeable {
public String toString() {
return proxyName() + " over " + proxyInfo;
}
+
+ @Override
+ public void close() throws IOException {
+ if (proxy != null) {
+ if (proxy instanceof Closeable) {
+ ((Closeable) proxy).close();
+ } else {
+ RPC.stopProxy(proxy);
+ }
+ }
+ }
}
/**
http://git-wip-us.apache.org/repos/asf/hadoop/blob/12bdbf2b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/server/namenode/ha/AbstractNNFailoverProxyProvider.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/server/namenode/ha/AbstractNNFailoverProxyProvider.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/server/namenode/ha/AbstractNNFailoverProxyProvider.java
index 572cb1c..d25c0c6 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/server/namenode/ha/AbstractNNFailoverProxyProvider.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/server/namenode/ha/AbstractNNFailoverProxyProvider.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.hdfs.server.namenode.ha;
+import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.URI;
@@ -28,13 +29,17 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
+import com.google.common.annotations.VisibleForTesting;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
import org.apache.hadoop.ha.HAServiceProtocol.HAServiceState;
import org.apache.hadoop.hdfs.DFSUtilClient;
import org.apache.hadoop.hdfs.HAUtilClient;
+import org.apache.hadoop.hdfs.NameNodeProxiesClient;
import org.apache.hadoop.hdfs.client.HdfsClientConfigKeys;
+import org.apache.hadoop.hdfs.protocol.ClientProtocol;
import org.apache.hadoop.io.retry.FailoverProxyProvider;
+import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.security.UserGroupInformation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -110,7 +115,7 @@ public abstract class AbstractNNFailoverProxyProvider<T> implements
/**
* ProxyInfo to a NameNode. Includes its address.
*/
- public static class NNProxyInfo<T> extends ProxyInfo<T> {
+ public static class NNProxyInfo<T> extends ProxyInfo<T> implements Closeable {
private InetSocketAddress address;
/**
* The currently known state of the NameNode represented by this ProxyInfo.
@@ -119,6 +124,9 @@ public abstract class AbstractNNFailoverProxyProvider<T> implements
*/
private HAServiceState cachedState;
+ /** Proxy to get NameNode's HA service state. */
+ private ClientProtocol serviceStateProxy;
+
public NNProxyInfo(InetSocketAddress address) {
super(null, address.toString());
this.address = address;
@@ -128,6 +136,10 @@ public abstract class AbstractNNFailoverProxyProvider<T> implements
return address;
}
+ public void refreshCachedState() throws IOException {
+ cachedState = serviceStateProxy.getHAServiceState();
+ }
+
public void setCachedState(HAServiceState state) {
cachedState = state;
}
@@ -135,6 +147,23 @@ public abstract class AbstractNNFailoverProxyProvider<T> implements
public HAServiceState getCachedState() {
return cachedState;
}
+
+ @VisibleForTesting
+ void setClientProxyForTesting(ClientProtocol cp) {
+ this.serviceStateProxy = cp;
+ }
+
+ @Override
+ public void close() throws IOException {
+ super.close();
+ if (serviceStateProxy != null) {
+ if (serviceStateProxy instanceof Closeable) {
+ ((Closeable) serviceStateProxy).close();
+ } else {
+ RPC.stopProxy(serviceStateProxy);
+ }
+ }
+ }
}
@Override
@@ -146,8 +175,8 @@ public abstract class AbstractNNFailoverProxyProvider<T> implements
* Create a proxy if it has not been created yet.
*/
protected NNProxyInfo<T> createProxyIfNeeded(NNProxyInfo<T> pi) {
+ assert pi.getAddress() != null : "Proxy address is null";
if (pi.proxy == null) {
- assert pi.getAddress() != null : "Proxy address is null";
try {
pi.proxy = factory.createProxy(conf,
pi.getAddress(), xface, ugi, false, getFallbackToSimpleAuth());
@@ -157,6 +186,21 @@ public abstract class AbstractNNFailoverProxyProvider<T> implements
throw new RuntimeException(ioe);
}
}
+ if (pi.serviceStateProxy == null) {
+ if (pi.proxy instanceof ClientProtocol) {
+ pi.serviceStateProxy = (ClientProtocol) pi.proxy;
+ } else {
+ try {
+ pi.serviceStateProxy = NameNodeProxiesClient
+ .createNonHAProxyWithClientProtocol(pi.address, conf, ugi,
+ false, getFallbackToSimpleAuth());
+ } catch (IOException ioe) {
+ LOG.error("{} Failed to create client RPC proxy to NameNode at {}",
+ this.getClass().getSimpleName(), pi.address, ioe);
+ throw new RuntimeException(ioe);
+ }
+ }
+ }
return pi;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/12bdbf2b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/server/namenode/ha/IPFailoverProxyProvider.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/server/namenode/ha/IPFailoverProxyProvider.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/server/namenode/ha/IPFailoverProxyProvider.java
index 3eb181d..b5f4951 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/server/namenode/ha/IPFailoverProxyProvider.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/server/namenode/ha/IPFailoverProxyProvider.java
@@ -17,13 +17,11 @@
*/
package org.apache.hadoop.hdfs.server.namenode.ha;
-import java.io.Closeable;
import java.io.IOException;
import java.net.URI;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdfs.DFSUtilClient;
-import org.apache.hadoop.ipc.RPC;
/**
* A NNFailoverProxyProvider implementation which works on IP failover setup.
@@ -67,14 +65,7 @@ public class IPFailoverProxyProvider<T> extends
*/
@Override
public synchronized void close() throws IOException {
- if (nnProxyInfo.proxy == null) {
- return;
- }
- if (nnProxyInfo.proxy instanceof Closeable) {
- ((Closeable)nnProxyInfo.proxy).close();
- } else {
- RPC.stopProxy(nnProxyInfo.proxy);
- }
+ nnProxyInfo.close();
}
/**
http://git-wip-us.apache.org/repos/asf/hadoop/blob/12bdbf2b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/server/namenode/ha/ObserverReadProxyProvider.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/server/namenode/ha/ObserverReadProxyProvider.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/server/namenode/ha/ObserverReadProxyProvider.java
index 5d56c59..1457222 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/server/namenode/ha/ObserverReadProxyProvider.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/server/namenode/ha/ObserverReadProxyProvider.java
@@ -30,7 +30,6 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.ha.HAServiceProtocol.HAServiceState;
import org.apache.hadoop.hdfs.ClientGSIContext;
import org.apache.hadoop.hdfs.client.HdfsClientConfigKeys;
-import org.apache.hadoop.hdfs.protocol.ClientProtocol;
import org.apache.hadoop.io.retry.AtMostOnce;
import org.apache.hadoop.io.retry.Idempotent;
import org.apache.hadoop.io.retry.RetryPolicies;
@@ -58,13 +57,14 @@ import com.google.common.annotations.VisibleForTesting;
* Read and write requests will still be sent to active NN if reading from
* observer is turned off.
*/
-public class ObserverReadProxyProvider<T extends ClientProtocol>
+public class ObserverReadProxyProvider<T>
extends AbstractNNFailoverProxyProvider<T> {
private static final Logger LOG = LoggerFactory.getLogger(
ObserverReadProxyProvider.class);
- /** Client-side context for syncing with the NameNode server side. */
- private final AlignmentContext alignmentContext;
+ /** Client-side context for syncing with the NameNode server side.
+ * This will be null if protocol is not ClientProtocol. */
+ private AlignmentContext alignmentContext;
/** The inner proxy provider used for active/standby failover. */
private final AbstractNNFailoverProxyProvider<T> failoverProxy;
@@ -114,8 +114,11 @@ public class ObserverReadProxyProvider<T extends ClientProtocol>
AbstractNNFailoverProxyProvider<T> failoverProxy) {
super(conf, uri, xface, factory);
this.failoverProxy = failoverProxy;
- this.alignmentContext = new ClientGSIContext();
- ((ClientHAProxyFactory<T>) factory).setAlignmentContext(alignmentContext);
+
+ if (factory instanceof ClientHAProxyFactory<?>) {
+ this.alignmentContext = new ClientGSIContext();
+ ((ClientHAProxyFactory<T>) factory).setAlignmentContext(alignmentContext);
+ }
// Don't bother configuring the number of retries and such on the retry
// policy since it is mainly only used for determining whether or not an
@@ -145,6 +148,7 @@ public class ObserverReadProxyProvider<T extends ClientProtocol>
this.observerReadEnabled = true;
}
+ @VisibleForTesting
public AlignmentContext getAlignmentContext() {
return alignmentContext;
}
@@ -212,8 +216,7 @@ public class ObserverReadProxyProvider<T extends ClientProtocol>
currentIndex = (currentIndex + 1) % nameNodeProxies.size();
currentProxy = createProxyIfNeeded(nameNodeProxies.get(currentIndex));
try {
- HAServiceState state = currentProxy.proxy.getHAServiceState();
- currentProxy.setCachedState(state);
+ currentProxy.refreshCachedState();
} catch (IOException e) {
LOG.info("Failed to connect to {}. Setting cached state to Standby",
currentProxy.getAddress(), e);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/12bdbf2b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/DFSTestUtil.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/DFSTestUtil.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/DFSTestUtil.java
index 97609fb..5be2205 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/DFSTestUtil.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/DFSTestUtil.java
@@ -71,7 +71,6 @@ import java.util.Random;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicBoolean;
import com.google.common.base.Charsets;
import com.google.common.base.Joiner;
@@ -186,7 +185,6 @@ import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.net.unix.DomainSocket;
import org.apache.hadoop.net.unix.TemporarySocketDirectory;
import org.apache.hadoop.security.AccessControlException;
-import org.apache.hadoop.security.RefreshUserMappingsProtocol;
import org.apache.hadoop.security.ShellBasedUnixGroupsMapping;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.token.Token;
@@ -2019,24 +2017,6 @@ public class DFSTestUtil {
}
/**
- * Get the RefreshUserMappingsProtocol RPC proxy for the NN associated with
- * this DFSClient object
- *
- * @param nameNodeUri the URI of the NN to get a proxy for.
- *
- * @return the RefreshUserMappingsProtocol RPC proxy associated with this
- * DFSClient object
- */
- @VisibleForTesting
- public static RefreshUserMappingsProtocol getRefreshUserMappingsProtocolProxy(
- Configuration conf, URI nameNodeUri) throws IOException {
- final AtomicBoolean nnFallbackToSimpleAuth = new AtomicBoolean(false);
- return NameNodeProxies.createProxy(conf,
- nameNodeUri, RefreshUserMappingsProtocol.class,
- nnFallbackToSimpleAuth).getProxy();
- }
-
- /**
* Set the datanode dead
*/
public static void setDatanodeDead(DatanodeInfo dn) {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/12bdbf2b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/NNThroughputBenchmark.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/NNThroughputBenchmark.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/NNThroughputBenchmark.java
index a9c4578..7d2ca89 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/NNThroughputBenchmark.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/NNThroughputBenchmark.java
@@ -28,6 +28,7 @@ import java.util.List;
import com.google.common.base.Preconditions;
+import org.apache.hadoop.hdfs.NameNodeProxies;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.hadoop.conf.Configuration;
@@ -1521,8 +1522,8 @@ public class NNThroughputBenchmark implements Tool {
clientProto = dfs.getClient().getNamenode();
dataNodeProto = new DatanodeProtocolClientSideTranslatorPB(
DFSUtilClient.getNNAddress(nnUri), config);
- refreshUserMappingsProto =
- DFSTestUtil.getRefreshUserMappingsProtocolProxy(config, nnUri);
+ refreshUserMappingsProto = NameNodeProxies.createProxy(
+ config, nnUri, RefreshUserMappingsProtocol.class).getProxy();
getBlockPoolId(dfs);
}
// run each benchmark
http://git-wip-us.apache.org/repos/asf/hadoop/blob/12bdbf2b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/ha/TestObserverReadProxyProvider.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/ha/TestObserverReadProxyProvider.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/ha/TestObserverReadProxyProvider.java
index caf7d00..c5c2863 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/ha/TestObserverReadProxyProvider.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/ha/TestObserverReadProxyProvider.java
@@ -27,12 +27,14 @@ import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.permission.FsAction;
+import org.apache.hadoop.hdfs.NameNodeProxies;
import org.apache.hadoop.hdfs.client.HdfsClientConfigKeys;
import org.apache.hadoop.hdfs.protocol.ClientProtocol;
import org.apache.hadoop.hdfs.protocol.LocatedBlock;
import org.apache.hadoop.ipc.ObserverRetryOnActiveException;
import org.apache.hadoop.ipc.RemoteException;
import org.apache.hadoop.ipc.StandbyException;
+import org.apache.hadoop.security.RefreshUserMappingsProtocol;
import org.apache.hadoop.security.UserGroupInformation;
import org.junit.Before;
import org.junit.Test;
@@ -109,6 +111,10 @@ public class TestObserverReadProxyProvider {
URI uri, String addressKey) {
List<NNProxyInfo<ClientProtocol>> nnProxies =
super.getProxyAddresses(uri, addressKey);
+ for (NNProxyInfo<ClientProtocol> nnProxy : nnProxies) {
+ String address = nnProxy.getAddress().toString();
+ nnProxy.setClientProxyForTesting(proxyMap.get(address));
+ }
return nnProxies;
}
};
@@ -289,6 +295,18 @@ public class TestObserverReadProxyProvider {
assertHandledBy(1);
}
+ /**
+ * Test that ObserverReadProxyProvider also works with NameNodeProtocols.
+ * See HDFS-14116 for details.
+ */
+ @Test
+ public void testNameNodeProtocol() throws Exception {
+ setupProxyProvider(3);
+ conf.set(HdfsClientConfigKeys.Failover.PROXY_PROVIDER_KEY_PREFIX
+ + "." + ns, ObserverReadProxyProvider.class.getName());
+ NameNodeProxies.createProxy(conf, nnURI, RefreshUserMappingsProtocol.class);
+ }
+
private void doRead() throws Exception {
doRead(proxyProvider.getProxy().proxy);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org