RequestHedgingProxyProvider extends
        ConfiguredFailoverProxyProvider<T> { @SuppressWarnings("unchecked") @Override public synchronized ProxyInfo<T> getProxy() { if (currentUsedHandler != null) { return currentUsedHandler; } Map<String, ProxyInfo<T>> targetProxyInfos = new HashMap<>(); StringBuilder combinedInfo = new StringBuilder("["); for (int i = 0; i < proxies.size(); i++) { ProxyInfo<T> pInfo = super.getProxy(); incrementProxyIndex(); targetProxyInfos.put(pInfo.proxyInfo, pInfo); combinedInfo.append(pInfo.proxyInfo).append(','); } combinedInfo.append(']'); T wrappedProxy = (T) Proxy.newProxyInstance( RequestHedgingInvocationHandler.class.getClassLoader(), new Class<?>[]{xface}, new RequestHedgingInvocationHandler(targetProxyInfos)); currentUsedHandler = new ProxyInfo<T>(wrappedProxy, combinedInfo.toString()); return currentUsedHandler; } RequestHedgingProxyProvider(Configuration conf, URI uri,
      Class<T> xface, HAProxyFactory<T> proxyFactory); @SuppressWarnings("unchecked") @Override synchronized ProxyInfo<T> getProxy(); @Override synchronized void performFailover(T currentProxy); static final Logger LOG; }
@Test public void testHedgingWhenOneFails() throws Exception { final ClientProtocol goodMock = Mockito.mock(ClientProtocol.class); Mockito.when(goodMock.getStats()).thenAnswer(new Answer<long[]>() { @Override public long[] answer(InvocationOnMock invocation) throws Throwable { Thread.sleep(1000); return new long[]{1}; } }); final ClientProtocol badMock = Mockito.mock(ClientProtocol.class); Mockito.when(badMock.getStats()).thenThrow(new IOException("Bad mock !!")); RequestHedgingProxyProvider<ClientProtocol> provider = new RequestHedgingProxyProvider<>(conf, nnUri, ClientProtocol.class, createFactory(badMock, goodMock)); long[] stats = provider.getProxy().proxy.getStats(); Assert.assertTrue(stats.length == 1); Mockito.verify(badMock).getStats(); 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<long[]>() { @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<long[]>() { @Override public long[] answer(InvocationOnMock invocation) throws Throwable { badCount.incrementAndGet(); throw new IOException("Bad mock !!"); } }); RequestHedgingProxyProvider<ClientProtocol> 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()); 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<long[]>() { 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<long[]>() { @Override public long[] answer(InvocationOnMock invocation) throws Throwable { throw new IOException("Bad Mock! This is Standby!"); } }); RequestHedgingProxyProvider<ClientProtocol> provider = new RequestHedgingProxyProvider<>(conf, nnUri, ClientProtocol.class, createFactory(badMock, goodMock)); ClientProtocol proxy = provider.getProxy().proxy; proxy.getStats(); try { proxy.getStats(); } catch (IOException e) { assertExceptionContains("Expected Exception Info", e); } }
@Test public void testHedgingWhenOneIsSlow() throws Exception { final ClientProtocol goodMock = Mockito.mock(ClientProtocol.class); Mockito.when(goodMock.getStats()).thenAnswer(new Answer<long[]>() { @Override public long[] answer(InvocationOnMock invocation) throws Throwable { Thread.sleep(1000); return new long[]{1}; } }); final ClientProtocol badMock = Mockito.mock(ClientProtocol.class); Mockito.when(badMock.getStats()).thenThrow(new IOException("Bad mock !!")); RequestHedgingProxyProvider<ClientProtocol> provider = new RequestHedgingProxyProvider<>(conf, nnUri, ClientProtocol.class, createFactory(goodMock, badMock)); long[] stats = provider.getProxy().proxy.getStats(); Assert.assertTrue(stats.length == 1); Assert.assertEquals(1, stats[0]); Mockito.verify(badMock).getStats(); Mockito.verify(goodMock).getStats(); }
@Test public void testHedgingWhenBothFail() throws Exception { ClientProtocol badMock = Mockito.mock(ClientProtocol.class); Mockito.when(badMock.getStats()).thenThrow(new IOException("Bad mock !!")); ClientProtocol worseMock = Mockito.mock(ClientProtocol.class); Mockito.when(worseMock.getStats()).thenThrow( new IOException("Worse mock !!")); RequestHedgingProxyProvider<ClientProtocol> provider = new RequestHedgingProxyProvider<>(conf, nnUri, ClientProtocol.class, createFactory(badMock, worseMock)); try { provider.getProxy().proxy.getStats(); Assert.fail("Should fail since both namenodes throw IOException !!"); } catch (Exception e) { Assert.assertTrue(e instanceof MultiException); } Mockito.verify(badMock).getStats(); Mockito.verify(worseMock).getStats(); }
@Test public void testHedgingWhenFileNotFoundException() throws Exception { ClientProtocol active = Mockito.mock(ClientProtocol.class); Mockito .when(active.getBlockLocations(anyString(), anyLong(), anyLong())) .thenThrow(new RemoteException("java.io.FileNotFoundException", "File does not exist!")); ClientProtocol standby = Mockito.mock(ClientProtocol.class); Mockito .when(standby.getBlockLocations(anyString(), anyLong(), anyLong())) .thenThrow( new RemoteException("org.apache.hadoop.ipc.StandbyException", "Standby NameNode")); RequestHedgingProxyProvider<ClientProtocol> provider = new RequestHedgingProxyProvider<>(conf, nnUri, ClientProtocol.class, createFactory(active, standby)); try { provider.getProxy().proxy.getBlockLocations("/tmp/test.file", 0L, 20L); Assert.fail("Should fail since the active namenode throws" + " FileNotFoundException!"); } catch (MultiException me) { for (Exception ex : me.getExceptions().values()) { Exception rEx = ((RemoteException) ex).unwrapRemoteException(); if (rEx instanceof StandbyException) { continue; } Assert.assertTrue(rEx instanceof FileNotFoundException); } } Mockito.verify(active).getBlockLocations(anyString(), anyLong(), anyLong()); Mockito.verify(standby).getBlockLocations(anyString(), anyLong(), anyLong()); }
@Test public void testHedgingWhenConnectException() throws Exception { ClientProtocol active = Mockito.mock(ClientProtocol.class); Mockito.when(active.getStats()).thenThrow(new ConnectException()); ClientProtocol standby = Mockito.mock(ClientProtocol.class); Mockito.when(standby.getStats()) .thenThrow( new RemoteException("org.apache.hadoop.ipc.StandbyException", "Standby NameNode")); RequestHedgingProxyProvider<ClientProtocol> provider = new RequestHedgingProxyProvider<>(conf, nnUri, ClientProtocol.class, createFactory(active, standby)); try { provider.getProxy().proxy.getStats(); Assert.fail("Should fail since the active namenode throws" + " ConnectException!"); } catch (MultiException me) { for (Exception ex : me.getExceptions().values()) { if (ex instanceof RemoteException) { Exception rEx = ((RemoteException) ex) .unwrapRemoteException(); Assert.assertTrue("Unexpected RemoteException: " + rEx.getMessage(), rEx instanceof StandbyException); } else { Assert.assertTrue(ex instanceof ConnectException); } } } Mockito.verify(active).getStats(); Mockito.verify(standby).getStats(); }
@Test public void testHedgingWhenConnectAndEOFException() throws Exception { ClientProtocol active = Mockito.mock(ClientProtocol.class); Mockito.when(active.getStats()).thenThrow(new EOFException()); ClientProtocol standby = Mockito.mock(ClientProtocol.class); Mockito.when(standby.getStats()).thenThrow(new ConnectException()); RequestHedgingProxyProvider<ClientProtocol> provider = new RequestHedgingProxyProvider<>(conf, nnUri, ClientProtocol.class, createFactory(active, standby)); try { provider.getProxy().proxy.getStats(); Assert.fail("Should fail since both active and standby namenodes throw" + " Exceptions!"); } catch (MultiException me) { for (Exception ex : me.getExceptions().values()) { if (!(ex instanceof ConnectException) && !(ex instanceof EOFException)) { Assert.fail("Unexpected Exception " + ex.getMessage()); } } } Mockito.verify(active).getStats(); Mockito.verify(standby).getStats(); }