diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/server/namenode/ha/RequestHedgingProxyProvider.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/server/namenode/ha/RequestHedgingProxyProvider.java index 7b9cd64dc5..c074bb80cd 100644 --- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/server/namenode/ha/RequestHedgingProxyProvider.java +++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/server/namenode/ha/RequestHedgingProxyProvider.java @@ -79,6 +79,20 @@ public RequestHedgingInvocationHandler( public Object invoke(Object proxy, final Method method, final Object[] args) throws Throwable { + if (currentUsedProxy != null) { + try { + Object retVal = method.invoke(currentUsedProxy.proxy, args); + LOG.debug("Invocation successful on [{}]", + currentUsedProxy.proxyInfo); + return retVal; + } catch (InvocationTargetException ex) { + Exception unwrappedException = unwrapInvocationTargetException(ex); + logProxyException(unwrappedException, currentUsedProxy.proxyInfo); + LOG.trace("Unsuccessful invocation on [{}]", + currentUsedProxy.proxyInfo); + throw unwrappedException; + } + } Map, ProxyInfo> proxyMap = new HashMap<>(); int numAttempts = 0; diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/test/java/org/apache/hadoop/hdfs/server/namenode/ha/TestRequestHedgingProxyProvider.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/test/java/org/apache/hadoop/hdfs/server/namenode/ha/TestRequestHedgingProxyProvider.java index 8d6b02dc15..8913f1a5ea 100644 --- a/hadoop-hdfs-project/hadoop-hdfs-client/src/test/java/org/apache/hadoop/hdfs/server/namenode/ha/TestRequestHedgingProxyProvider.java +++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/test/java/org/apache/hadoop/hdfs/server/namenode/ha/TestRequestHedgingProxyProvider.java @@ -43,10 +43,15 @@ import org.junit.Before; import org.junit.BeforeClass; import org.junit.Test; + +import static org.apache.hadoop.test.GenericTestUtils.assertExceptionContains; +import static org.junit.Assert.assertEquals; import org.mockito.Matchers; import org.mockito.Mockito; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; +import static org.mockito.Mockito.when; +import static org.mockito.Mockito.mock; import com.google.common.collect.Lists; @@ -99,6 +104,80 @@ public long[] answer(InvocationOnMock invocation) throws Throwable { Mockito.verify(goodMock).getStats(); } + @Test + public void testRequestNNAfterOneSuccess() throws Exception { + final AtomicInteger goodCount = new AtomicInteger(0); + final AtomicInteger badCount = new AtomicInteger(0); + final ClientProtocol goodMock = mock(ClientProtocol.class); + when(goodMock.getStats()).thenAnswer(new Answer() { + @Override + public long[] answer(InvocationOnMock invocation) throws Throwable { + goodCount.incrementAndGet(); + Thread.sleep(1000); + return new long[]{1}; + } + }); + final ClientProtocol badMock = mock(ClientProtocol.class); + when(badMock.getStats()).thenAnswer(new Answer() { + @Override + public long[] answer(InvocationOnMock invocation) throws Throwable { + badCount.incrementAndGet(); + throw new IOException("Bad mock !!"); + } + }); + + RequestHedgingProxyProvider provider = + new RequestHedgingProxyProvider<>(conf, nnUri, ClientProtocol.class, + createFactory(badMock, goodMock)); + ClientProtocol proxy = provider.getProxy().proxy; + proxy.getStats(); + assertEquals(1, goodCount.get()); + assertEquals(1, badCount.get()); + // We will only use the successful proxy after a successful invocation. + proxy.getStats(); + assertEquals(2, goodCount.get()); + assertEquals(1, badCount.get()); + } + + @Test + public void testExceptionInfo() throws Exception { + final ClientProtocol goodMock = mock(ClientProtocol.class); + when(goodMock.getStats()).thenAnswer(new Answer() { + private boolean first = true; + @Override + public long[] answer(InvocationOnMock invocation) + throws Throwable { + if (first) { + Thread.sleep(1000); + first = false; + return new long[] {1}; + } else { + throw new IOException("Expected Exception Info"); + } + } + }); + final ClientProtocol badMock = mock(ClientProtocol.class); + when(badMock.getStats()).thenAnswer(new Answer() { + @Override + public long[] answer(InvocationOnMock invocation) + throws Throwable { + throw new IOException("Bad Mock! This is Standby!"); + } + }); + + RequestHedgingProxyProvider provider = + new RequestHedgingProxyProvider<>(conf, nnUri, ClientProtocol.class, + createFactory(badMock, goodMock)); + ClientProtocol proxy = provider.getProxy().proxy; + proxy.getStats(); + // Test getting the exception when the successful proxy encounters one. + try { + proxy.getStats(); + } catch (IOException e) { + assertExceptionContains("Expected Exception Info", e); + } + } + @Test public void testHedgingWhenOneIsSlow() throws Exception { final ClientProtocol goodMock = Mockito.mock(ClientProtocol.class);