diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/io_/retry/RetryInvocationHandler.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/io_/retry/RetryInvocationHandler.java index 604b82ea2a61..f421a1dbd1b2 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/io_/retry/RetryInvocationHandler.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/io_/retry/RetryInvocationHandler.java @@ -423,6 +423,9 @@ static boolean isRpcInvocation(Object proxy) { if (proxy instanceof ProtocolTranslator) { proxy = ((ProtocolTranslator) proxy).getUnderlyingProxyObject(); } + if (proxy instanceof RpcProxy) { + return true; + } if (!Proxy.isProxyClass(proxy.getClass())) { return false; } diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ipc_/RPC.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ipc_/RPC.java index 56d647f5eb7a..e26bae6df7a4 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ipc_/RPC.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ipc_/RPC.java @@ -352,6 +352,9 @@ public static ConnectionId getConnectionIdForProxy(Object proxy) { if (proxy instanceof ProtocolTranslator) { proxy = ((ProtocolTranslator)proxy).getUnderlyingProxyObject(); } + if (proxy instanceof RpcProxy) { + return ((RpcProxy) proxy).getConnectionId(); + } RpcInvocationHandler inv = (RpcInvocationHandler) Proxy .getInvocationHandler(proxy); return inv.getConnectionId(); diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ipc_/RpcInvocationHandler.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ipc_/RpcInvocationHandler.java index bc91d2bd16e2..f3f39b22907b 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ipc_/RpcInvocationHandler.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ipc_/RpcInvocationHandler.java @@ -17,20 +17,11 @@ */ package org.apache.hadoop.ipc_; -import java.io.Closeable; import java.lang.reflect.InvocationHandler; -import org.apache.hadoop.ipc_.Client.ConnectionId; - /** * This interface must be implemented by all InvocationHandler * implementations. */ -public interface RpcInvocationHandler extends InvocationHandler, Closeable { - - /** - * Returns the connection id associated with the InvocationHandler instance. - * @return ConnectionId - */ - ConnectionId getConnectionId(); +public interface RpcInvocationHandler extends InvocationHandler, RpcProxy { } diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ipc_/RpcProxy.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ipc_/RpcProxy.java new file mode 100644 index 000000000000..3bdb21a8ce1f --- /dev/null +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ipc_/RpcProxy.java @@ -0,0 +1,31 @@ +/* + * 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.ipc_; + +import java.io.Closeable; +import org.apache.hadoop.ipc_.Client.ConnectionId; + +/** + * Connection and lifecycle access for RPC proxies, including concrete protocol implementations. + */ +public interface RpcProxy extends Closeable { + /** + * @return the connection ID associated with this RPC proxy. + */ + ConnectionId getConnectionId(); +} diff --git a/hadoop-hdds/common/src/test/java/org/apache/hadoop/io_/retry/TestRetryProxy.java b/hadoop-hdds/common/src/test/java/org/apache/hadoop/io_/retry/TestRetryProxy.java index 4239d49a02c7..f6a10cfa7b3b 100644 --- a/hadoop-hdds/common/src/test/java/org/apache/hadoop/io_/retry/TestRetryProxy.java +++ b/hadoop-hdds/common/src/test/java/org/apache/hadoop/io_/retry/TestRetryProxy.java @@ -25,6 +25,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.fail; import static org.mockito.Mockito.any; @@ -33,6 +34,7 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; import static org.mockito.Mockito.when; import java.io.IOException; @@ -51,7 +53,10 @@ import org.apache.hadoop.io.retry.RetryPolicy.RetryAction; import org.apache.hadoop.io.retry.RetryPolicy.RetryAction.RetryDecision; import org.apache.hadoop.io_.retry.UnreliableInterface.UnreliableException; +import org.apache.hadoop.ipc_.Client.ConnectionId; import org.apache.hadoop.ipc_.ProtocolTranslator; +import org.apache.hadoop.ipc_.RPC; +import org.apache.hadoop.ipc_.RpcProxy; import org.apache.hadoop.security.AccessControlException; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -142,6 +147,24 @@ public Object getUnderlyingProxyObject() { // For non-proxy the method must return false assertFalse(RetryInvocationHandler.isRpcInvocation(new Object())); } + + @Test + public void testConcreteRpcProxy() throws Exception { + RpcProxy proxy = mock(RpcProxy.class); + assertTrue(RetryInvocationHandler.isRpcInvocation(proxy)); + verifyNoInteractions(proxy); + + ConnectionId connectionId = mock(ConnectionId.class); + when(proxy.getConnectionId()).thenReturn(connectionId); + assertSame(connectionId, RPC.getConnectionIdForProxy(proxy)); + + ProtocolTranslator translator = () -> proxy; + assertTrue(RetryInvocationHandler.isRpcInvocation(translator)); + assertSame(connectionId, RPC.getConnectionIdForProxy(translator)); + + RPC.stopProxy(proxy); + verify(proxy).close(); + } @Test public void testRetryForever() throws UnreliableException { diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java index 38a7bbbb5bb2..feed03f54a70 100644 --- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java +++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java @@ -27,22 +27,21 @@ import com.google.protobuf.ServiceException; import java.io.IOException; import java.io.InterruptedIOException; -import java.lang.reflect.InvocationTargetException; -import java.lang.reflect.Method; -import java.lang.reflect.Proxy; import java.util.List; import org.apache.hadoop.io.retry.FailoverProxyProvider; import org.apache.hadoop.io.retry.RetryPolicy; +import org.apache.hadoop.io_.retry.RetryProxy; import org.apache.hadoop.ipc_.Client.ConnectionId; import org.apache.hadoop.ipc_.RPC; -import org.apache.hadoop.ipc_.RpcInvocationHandler; import org.apache.hadoop.ipc_.RpcNoSuchProtocolException; +import org.apache.hadoop.ipc_.RpcProxy; import org.apache.hadoop.ozone.OmUtils; import org.apache.hadoop.ozone.om.exceptions.OMLeaderNotReadyException; import org.apache.hadoop.ozone.om.exceptions.OMNotLeaderException; import org.apache.hadoop.ozone.om.helpers.ReadConsistency; import org.apache.hadoop.ozone.om.protocolPB.OzoneManagerProtocolPB; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest; +import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.ReadConsistencyHint; import org.apache.ratis.protocol.exceptions.ReadException; import org.apache.ratis.protocol.exceptions.ReadIndexException; @@ -117,10 +116,7 @@ public HadoopRpcOMFollowerReadFailoverProxyProvider( final String combinedInfo = "[" + leaderProxy.getOMProxies().stream() .map(a -> a.proxyInfo) .reduce((a, b) -> a + ", " + b).orElse("") + "]"; - OzoneManagerProtocolPB wrappedProxy = (OzoneManagerProtocolPB) Proxy.newProxyInstance( - FollowerReadInvocationHandler.class.getClassLoader(), - new Class[] {OzoneManagerProtocolPB.class}, new FollowerReadInvocationHandler()); - combinedProxy = new ProxyInfo<>(wrappedProxy, combinedInfo); + combinedProxy = new ProxyInfo<>(new FollowerReadProxy(), combinedInfo); this.useFollowerRead = useFollowerRead; this.followerReadConsistency = followerReadConsistencyType.getHint(); this.leaderReadConsistency = leaderReadConsistencyType.getHint(); @@ -138,7 +134,7 @@ public ProxyInfo getProxy() { @Override public void performFailover(OzoneManagerProtocolPB currProxy) { - // Since FollowerReadInvocationHandler might user or fallback to leader-based failover logic, + // Since FollowerReadProxy might use or fall back to leader-based failover logic, // we should delegate the failover logic to the leader's failover. leaderProxy.performFailover(currProxy); } @@ -148,34 +144,65 @@ public RetryPolicy getRetryPolicy(int maxFailovers) { // for a few reasons // 1. We want to ensure that the retry policy behavior remains the same when we use the leader proxy // (when follower read is disabled or using write request) - // 2. The FollowerInvocationHandler is also written so that the thrown exception is handled by the + // 2. The FollowerReadProxy is also written so that the thrown exception is handled by the // OMFailoverProxyProviderbase's RetryPolicy return leaderProxy.getRetryPolicy(maxFailovers); } /** - * Parse the OM request from the request args. - * - * @return parsed OM request. + * Create a client that applies the default consistency hint once, before retries. + *

+ * Proxy/provider structure: + * 

{@code
+   * ReadConsistencyProxy
+   * └── RetryProxy
+   *     └── HadoopRpcOMFollowerReadFailoverProxyProvider
+   *         ├── FollowerReadProxy
+   *         └── HadoopRpcOMFailoverProxyProvider (leaderProxy)
+   * }
*/ - private static OMRequest parseOMRequest(Object[] args) throws ServiceException { - String error = null; - if (args == null) { - error = "args == null"; - } else if (args.length < 2) { - error = "args.length == " + args.length + " < 2"; - } else if (args[1] == null) { - error = "args[1] == null"; - } else if (!(args[1] instanceof OMRequest)) { - error = "Non-OMRequest: " + args[1].getClass(); + public OzoneManagerProtocolPB newProxy(int maxFailovers) { + OzoneManagerProtocolPB retryProxy = (OzoneManagerProtocolPB) RetryProxy.create( + OzoneManagerProtocolPB.class, this, getRetryPolicy(maxFailovers)); + return new ReadConsistencyProxy(retryProxy); + } + + private class ReadConsistencyProxy implements OzoneManagerProtocolPB, RpcProxy { + private final OzoneManagerProtocolPB retryProxy; + + ReadConsistencyProxy(OzoneManagerProtocolPB retryProxy) { + this.retryProxy = retryProxy; + } + + @Override + public OMResponse submitRequest(RpcController controller, OMRequest request) throws ServiceException { + return retryProxy.submitRequest(controller, applyReadConsistency(request)); } - if (error != null) { - // Throws a non-retriable exception to prevent retry and failover - // See the HddsUtils#shouldNotFailoverOnRpcException used in - // OMFailoverProxyProviderBase#shouldFailover - throwServiceException(new RpcNoSuchProtocolException("Failed to parseOMRequest: " + error)); + + private OMRequest applyReadConsistency(OMRequest request) throws ServiceException { + if (request == null) { + // Reject invalid requests before entering the retry loop. + throw new ServiceException(new RpcNoSuchProtocolException("OMRequest == null")); + } + if (!request.hasReadConsistencyHint()) { + ReadConsistencyHint hint = useFollowerRead && OmUtils.shouldSendToFollower(request) + ? followerReadConsistency : leaderReadConsistency; + if (hint != null) { + return request.toBuilder().setReadConsistencyHint(hint).build(); + } + } + return request; + } + + @Override + public ConnectionId getConnectionId() { + return RPC.getConnectionIdForProxy(retryProxy); + } + + @Override + public void close() throws IOException { + HadoopRpcOMFollowerReadFailoverProxyProvider.this.close(); } - return (OMRequest) args[1]; } @VisibleForTesting @@ -219,8 +246,7 @@ private synchronized OMProxyInfo changeProxy(OMProxyInfo } /** - * An InvocationHandler to handle incoming requests. This class's invoke - * method contains the primary logic for redirecting to followers. + * A protocol implementation that redirects incoming requests to followers. *

* If follower reads are enabled, attempt to send read operations to the * current proxy which can be either a leader or follower. If the current @@ -228,60 +254,38 @@ private synchronized OMProxyInfo changeProxy(OMProxyInfo *

* Write requests are always forwarded to the leader. */ - private class FollowerReadInvocationHandler implements RpcInvocationHandler { + private class FollowerReadProxy implements OzoneManagerProtocolPB, RpcProxy { @Override - public Object invoke(Object proxy, final Method method, final Object[] args) - throws Throwable { + public OMResponse submitRequest(RpcController controller, OMRequest omRequest) throws ServiceException { lastProxy = null; - if (method.getDeclaringClass() == Object.class) { - // If the method is not a OzoneManagerProtocolPB method (e.g. Object#toString()), - // we should invoke the method on the current proxy - return method.invoke(this, args); - } - OMRequest omRequest = parseOMRequest(args); - - // Apply default consistency hint once, before any routing decision. - // In the future, we will support per-request hints which allows client (e.g. S3 clients) - // to specify a custom request header (e.g. x-ozone-read-consistency) as a consistency hint - // for read requests. boolean isFollowerReadEligible = useFollowerRead && OmUtils.shouldSendToFollower(omRequest); - if (!omRequest.hasReadConsistencyHint()) { - final ReadConsistencyHint defaultReadConsistency = isFollowerReadEligible - ? followerReadConsistency : leaderReadConsistency; - if (defaultReadConsistency != null) { - omRequest = omRequest.toBuilder() - .setReadConsistencyHint(defaultReadConsistency) - .build(); - args[1] = omRequest; - } - } if (isFollowerReadEligible) { int failedCount = 0; for (int i = 0; useFollowerRead && i < leaderProxy.getOMProxyMap().size(); i++) { OMProxyInfo current = getCurrentProxy(); - LOG.debug("Attempting to service {} with cmdType {} using proxy {}", - method.getName(), omRequest.getCmdType(), current.proxyInfo); + LOG.debug("Attempting to service submitRequest with cmdType {} using proxy {}", + omRequest.getCmdType(), current.proxyInfo); try { - final Object retVal = method.invoke(current.getProxy(), args); + final OMResponse response = current.getProxy().submitRequest(controller, omRequest); lastProxy = current; - LOG.debug("Invocation of {} with cmdType {} using {} was successful", - method.getName(), omRequest.getCmdType(), current.proxyInfo); - return retVal; - } catch (InvocationTargetException ite) { - LOG.debug("Invocation of {} with cmdType {} using proxy {} failed", method.getName(), - omRequest.getCmdType(), current.proxyInfo, ite); - if (!(ite.getCause() instanceof Exception)) { - throwServiceException(ite.getCause()); + LOG.debug("Invocation of submitRequest with cmdType {} using {} was successful", + omRequest.getCmdType(), current.proxyInfo); + return response; + } catch (Throwable failure) { + LOG.debug("Invocation of submitRequest with cmdType {} using proxy {} failed", + omRequest.getCmdType(), current.proxyInfo, failure); + if (!(failure instanceof Exception)) { + throw toServiceException(failure); } - Exception e = (Exception) ite.getCause(); + Exception e = (Exception) failure; if (e instanceof InterruptedIOException || e instanceof InterruptedException) { // If interrupted, do not retry. LOG.warn("Invocation returned interrupted exception on [{}];", current.proxyInfo, e); - throwServiceException(e); + throw toServiceException(e); } if (e instanceof ServiceException) { @@ -305,7 +309,7 @@ public Object invoke(Object proxy, final Method method, final Object[] args) "Directly throw the exception to trigger retry", current.proxyInfo); // Throw here to trigger retry since we already communicate to the leader // If we break here instead, we will retry the same leader again without waiting - throw e; + throw toServiceException(e); } ReadIndexException readIndexException = getReadIndexException(e); @@ -333,7 +337,7 @@ public Object invoke(Object proxy, final Method method, final Object[] args) LOG.debug("Invocation with cmdType {} returned exception on [{}] that cannot be retried; " + "{} failure(s) so far", omRequest.getCmdType(), current.proxyInfo, failedCount, e); - throw e; + throw toServiceException(e); } else { failedCount++; LOG.warn( @@ -349,9 +353,9 @@ public Object invoke(Object proxy, final Method method, final Object[] args) // be that there is simply no Follower node running at all. if (failedCount > 0) { // If we get here, it means all followers have failed. - LOG.warn("{} nodes have failed for read request {} with cmdType {}." + LOG.warn("{} nodes have failed for submitRequest with cmdType {}." + " Falling back to leader.", failedCount, - omRequest.getCmdType(), method.getName()); + omRequest.getCmdType()); } else { if (LOG.isDebugEnabled()) { LOG.debug("Read falling back to leader without follower read " @@ -363,24 +367,23 @@ public Object invoke(Object proxy, final Method method, final Object[] args) // Either all followers have failed, follower reads are disabled, // or this is a write request. In any case, forward the request to // the leader OM. - LOG.debug("Using leader-based failoverProxy to service {}", method.getName()); + LOG.debug("Using leader-based failoverProxy to service submitRequest"); final OMProxyInfo currentLeaderProxy = leaderProxy.getProxy(); - Object retVal = null; try { - retVal = method.invoke(currentLeaderProxy.getProxy(), args); - } catch (InvocationTargetException e) { - LOG.debug("Exception thrown from leader-based failoverProxy", e.getCause()); + OMResponse response = currentLeaderProxy.getProxy().submitRequest(controller, omRequest); + lastProxy = currentLeaderProxy; + return response; + } catch (Throwable e) { + LOG.debug("Exception thrown from leader-based failoverProxy", e); // This exception will be handled by the OMFailoverProxyProviderBase#getRetryPolicy // (see getRetryPolicy). This ensures that the leader-only failover should still work. - throwServiceException(e.getCause()); + throw toServiceException(e); } - lastProxy = currentLeaderProxy; - return retVal; } @Override public void close() throws IOException { - + // The provider owns the underlying OM proxies. } @Override @@ -433,15 +436,10 @@ public synchronized void changeInitialProxyForTest(String initialOmNodeId) { } /** - * Throw the passed {@link Throwable} wrapped in {@link ServiceException}. - * This is required to prevent {@link java.lang.reflect.UndeclaredThrowableException} to be thrown - * since {@link OzoneManagerProtocolPB#submitRequest(RpcController, OMRequest)} only - * throws {@link ServiceException}. - * @param e exception to wrap in {@link ServiceException}. - * @throws ServiceException the exception that wraps the passed throwable. + * Preserve ServiceException instances and wrap other failures in the protocol's declared exception type. */ - private static void throwServiceException(Throwable e) throws ServiceException { - throw e instanceof ServiceException ? (ServiceException) e : new ServiceException(e); + private static ServiceException toServiceException(Throwable e) { + return e instanceof ServiceException ? (ServiceException) e : new ServiceException(e); } } diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolPB.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolPB.java index d11aa7e6aab1..22220b8744d3 100644 --- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolPB.java +++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolPB.java @@ -48,7 +48,6 @@ static OzoneManagerProtocolPB newProxy(OMFailoverProxyProviderBase omProxy : proxyProvider.getOMProxies()) { + assertNull(omProxy.getProxy()); + } assertNull(proxyProvider.getLastProxy()); + verifyNoInteractions((Object[]) proxies); } @Test - void testShortArgsArrayDoesNotThrowArrayIndex() throws Exception { + void testConnectionIdAndClose() throws Exception { setupProxyProvider(2); + ConnectionId first = mock(ConnectionId.class); + ConnectionId second = mock(ConnectionId.class); + when(((RpcProxy) proxies[0]).getConnectionId()).thenReturn(first); + when(((RpcProxy) proxies[1]).getConnectionId()).thenReturn(second); - Object combinedProxy = proxyProvider.getProxy().proxy; - InvocationHandler handler = Proxy.getInvocationHandler(combinedProxy); - Method submitRequest = OzoneManagerProtocolPB.class.getMethod( - "submitRequest", RpcController.class, OMRequest.class); + assertSame(first, RPC.getConnectionIdForProxy(retryProxy)); + omNodeAnswers[0].unreachable = true; + doRead(); + assertSame(second, RPC.getConnectionIdForProxy(retryProxy)); - ServiceException exception = assertThrows(ServiceException.class, - () -> handler.invoke(combinedProxy, submitRequest, new Object[] {null})); - assertInstanceOf(RpcNoSuchProtocolException.class, exception.getCause()); + omNodeAnswers[1].isFollowerReadSupported = false; + omNodeAnswers[0].unreachable = false; + omNodeAnswers[0].isLeader = true; + doRead(); + assertFalse(proxyProvider.isUseFollowerRead()); + assertSame(first, RPC.getConnectionIdForProxy(retryProxy)); + + RPC.stopProxy(retryProxy); + verify((RpcProxy) proxies[0]).close(); + verify((RpcProxy) proxies[1]).close(); } @Test @@ -338,6 +374,58 @@ void testNullRequest() throws Exception { ServiceException exception = assertThrows(ServiceException.class, () -> retryProxy.submitRequest(null, null)); assertInstanceOf(RpcNoSuchProtocolException.class, exception.getCause()); + verifyNoInteractions((Object[]) proxies); + } + + @Test + void testConsistencyHintSurvivesRetryAfterFollowerReadDisabled() throws Exception { + setupProxyProvider(2); + omNodeAnswers[0].isFollowerReadSupported = false; + omNodeAnswers[1].isLeader = true; + + doRead(); + + assertHandledBy(1); + assertFalse(proxyProvider.isUseFollowerRead()); + assertEquals(ReadConsistency.LINEARIZABLE_ALLOW_FOLLOWER.getHint(), + omNodeAnswers[1].lastRequest.getReadConsistencyHint()); + + doRead(); + assertEquals(ReadConsistency.DEFAULT.getHint(), omNodeAnswers[1].lastRequest.getReadConsistencyHint()); + } + + @Test + void testExplicitConsistencyAndController() throws Exception { + setupProxyProvider(2); + RpcController controller = mock(RpcController.class); + ReadConsistencyHint hint = ReadConsistency.LOCAL_LEASE.getHint(); + OMRequest request = OMRequest.newBuilder().setCmdType(Type.GetKeyInfo).setClientId("client") + .setReadConsistencyHint(hint).build(); + OMResponse response = OMResponse.newBuilder().setCmdType(Type.GetKeyInfo) + .setStatus(Status.OK).build(); + when(proxies[0].submitRequest(controller, request)).thenReturn(response); + + assertSame(response, retryProxy.submitRequest(controller, request)); + verify(proxies[0]).submitRequest(controller, request); + assertHandledBy(0); + } + + @ParameterizedTest + @EnumSource(value = Type.class, names = {"GetKeyInfo", "CreateKey"}) + void testInterruptedRequestDoesNotFailOver(Type type) throws Exception { + setupProxyProvider(2); + InterruptedIOException interrupted = new InterruptedIOException("interrupted"); + doAnswer(invocation -> { + throw interrupted; + }).when(proxies[0]).submitRequest(any(), any()); + OMRequest request = OMRequest.newBuilder().setCmdType(type).setClientId("client").build(); + + ServiceException exception = assertThrows(ServiceException.class, + () -> proxyProvider.getProxy().proxy.submitRequest(null, request)); + + assertSame(interrupted, exception.getCause()); + assertNull(proxyProvider.getLastProxy()); + verifyNoInteractions(proxies[1]); } @Test @@ -364,7 +452,7 @@ private void setupProxyProvider(int omNodeCount, OzoneConfiguration config) thro omNodeIds = new String[omNodeCount]; omNodeAnswers = new OMAnswer[omNodeCount]; StringJoiner allNodeIds = new StringJoiner(","); - final OzoneManagerProtocolPB[] proxies = new OzoneManagerProtocolPB[omNodeCount]; + proxies = new OzoneManagerProtocolPB[omNodeCount]; for (int i = 0; i < omNodeCount; i++) { String nodeId = NODE_ID_BASE_STR + (i + 1); // 1-th indexed config.set(ConfUtils.addKeySuffixes(OZONE_OM_ADDRESS_KEY, OM_SERVICE_ID, @@ -372,7 +460,7 @@ private void setupProxyProvider(int omNodeCount, OzoneConfiguration config) thro allNodeIds.add(nodeId); omNodeIds[i] = nodeId; omNodeAnswers[i] = new OMAnswer(); - proxies[i] = mock(OzoneManagerProtocolPB.class); + proxies[i] = mock(OzoneManagerProtocolPB.class, withSettings().extraInterfaces(RpcProxy.class)); doAnswer(omNodeAnswers[i].clientAnswer) .when(proxies[i]).submitRequest(any(), any()); doAnswer(omNodeAnswers[i].clientAnswer) @@ -439,10 +527,7 @@ protected List> initOmProxiesFromConfigs( proxyProvider = new HadoopRpcOMFollowerReadFailoverProxyProvider(underlyingProxyProvider); assertTrue(proxyProvider.isUseFollowerRead()); // Wrap the follower read proxy provider in retry proxy to allow automatic failover - retryProxy = (OzoneManagerProtocolPB) RetryProxy.create( - OzoneManagerProtocolPB.class, proxyProvider, - proxyProvider.getRetryPolicy(2 * omNodeCount) - ); + retryProxy = OzoneManagerProtocolPB.newProxy(proxyProvider, 2 * omNodeCount); // This is currently added to prevent IllegalStateException in // Client#setCallIdAndRetryCount since it seems that callId is set but not unset properly RetryInvocationHandler.SET_CALL_ID_FOR_TEST.set(false); @@ -510,6 +595,7 @@ private static class OMAnswer { private volatile boolean isFollowerReadSupported = true; private volatile boolean isThrowReadIndexException = false; private volatile boolean isThrowReadException = false; + private OMRequest lastRequest; private OMProtocolAnswer clientAnswer = new OMProtocolAnswer(); @@ -525,6 +611,7 @@ public OMResponse answer(InvocationOnMock invocationOnMock) throws Throwable { Thread.sleep(SLOW_RESPONSE_SLEEP_TIME); } OMRequest omRequest = invocationOnMock.getArgument(1); + lastRequest = omRequest; switch (omRequest.getCmdType()) { case CreateKey: if (!isLeader) {