Apache
/
hadoop
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HDFS-14160. [SBN read] ObserverReadInvocationHandler should implement RpcInvocationHandler. Contributed by Konstantin V Shvachko.

This commit is contained in:
Konstantin V Shvachko 2018-12-19 12:39:57 -08:00 committed by Chen Liang
parent eb7d7fa5e2
commit 2be93bd54b
4 changed files with 32 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/server/namenode/ha/ObserverReadProxyProvider.java
View File

@ -19,7 +19,6 @@ package org.apache.hadoop.hdfs.server.namenode.ha;
import java.io.Closeable;
import java.io.IOException;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
@ -39,9 +38,11 @@ import org.apache.hadoop.io.retry.RetryPolicies;
import org.apache.hadoop.io.retry.RetryPolicy;
import org.apache.hadoop.io.retry.RetryPolicy.RetryAction;
import org.apache.hadoop.ipc.AlignmentContext;
import org.apache.hadoop.ipc.Client.ConnectionId;
import org.apache.hadoop.ipc.ObserverRetryOnActiveException;
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.ipc.RemoteException;
import org.apache.hadoop.ipc.RpcInvocationHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -239,7 +240,7 @@ public class ObserverReadProxyProvider<T extends ClientProtocol>
*
* Write requests are always forwarded to the active.
*/
private class ObserverReadInvocationHandler implements InvocationHandler {
private class ObserverReadInvocationHandler implements RpcInvocationHandler {
@Override
public Object invoke(Object proxy, final Method method, final Object[] args)
@ -322,6 +323,14 @@ public class ObserverReadProxyProvider<T extends ClientProtocol>
lastProxy = activeProxy;
return retVal;
}
@Override
public void close() throws IOException {}
@Override
public ConnectionId getConnectionId() {
return RPC.getConnectionIdForProxy(getCurrentProxy().proxy);
}
}
@Override

2
hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestFsck.java
View File

@ -165,7 +165,7 @@ public class TestFsck {
private static final String LINE_SEPARATOR =
System.getProperty("line.separator");
static String runFsck(Configuration conf, int expectedErrCode,
public static String runFsck(Configuration conf, int expectedErrCode,
boolean checkErrorCode, String... path)
throws Exception {
ByteArrayOutputStream bStream = new ByteArrayOutputStream();

10
hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/ha/HATestUtil.java
View File

@ -176,10 +176,14 @@ public abstract class HATestUtil {
MiniDFSCluster cluster, Configuration conf,
Class<P> classFPP, boolean isObserverReadEnabled)
throws IOException, URISyntaxException {
conf = new Configuration(conf);
setupHAConfiguration(cluster, conf, 0, classFPP);
String logicalName = conf.get(DFSConfigKeys.DFS_NAMESERVICES);
URI nnUri = new URI(HdfsConstants.HDFS_URI_SCHEME + "://" + logicalName);
conf.set(HdfsClientConfigKeys.Failover.PROXY_PROVIDER_KEY_PREFIX
+ "." + logicalName, classFPP.getName());
conf.set("fs.defaultFS", nnUri.toString());
DistributedFileSystem dfs = (DistributedFileSystem)
FileSystem.get(getLogicalUri(cluster), conf);
FileSystem.get(nnUri, conf);
@SuppressWarnings("unchecked")
P provider = (P) ((RetryInvocationHandler<?>) Proxy.getInvocationHandler(
dfs.getClient().getNamenode())).getProxyProvider();

13
hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/ha/TestObserverNode.java
View File

@ -49,6 +49,7 @@ import org.apache.hadoop.hdfs.protocol.LocatedBlocks;
import org.apache.hadoop.hdfs.qjournal.MiniQJMHACluster;
import org.apache.hadoop.hdfs.server.blockmanagement.BlockManager;
import org.apache.hadoop.hdfs.server.namenode.NameNodeAdapter;
import org.apache.hadoop.hdfs.server.namenode.TestFsck;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
@ -339,6 +340,18 @@ public class TestObserverNode {
Mockito.reset(bmSpy);
}
@Test
public void testFsckWithObserver() throws Exception {
setObserverRead(true);
dfs.create(testPath, (short)1).close();
assertSentTo(0);
final String result = TestFsck.runFsck(conf, 0, true, "/");
LOG.info("result=" + result);
assertTrue(result.contains("Status: HEALTHY"));
}
private void assertSentTo(int nnIdx) throws IOException {
assertTrue("Request was not sent to the expected namenode " + nnIdx,
HATestUtil.isSentToAnyOfNameNodes(dfs, dfsCluster, nnIdx));