You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ozone.apache.org by iv...@apache.org on 2026/09/23 02:14:27 UTC
(ozone) branch master updated: HDDS-16529. Replace reflective OM follower routing with typed dispatch (#11289)
This is an automated email from the ASF dual-hosted git repository.
ivandika3 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new 54ad584c7b7 HDDS-16529. Replace reflective OM follower routing with typed dispatch (#11289)
54ad584c7b7 is described below
commit 54ad584c7b74495740a0f08b5a32e38465e3e56b
Author: Peter Lee <pe...@gmail.com>
AuthorDate: Wed Sep 23 10:14:21 2026 +0800
HDDS-16529. Replace reflective OM follower routing with typed dispatch (#11289)
---
.../hadoop/io_/retry/RetryInvocationHandler.java | 3 +
.../src/main/java/org/apache/hadoop/ipc_/RPC.java | 3 +
.../apache/hadoop/ipc_/RpcInvocationHandler.java | 11 +-
.../main/java/org/apache/hadoop/ipc_/RpcProxy.java | 31 ++++
.../apache/hadoop/io_/retry/TestRetryProxy.java | 23 +++
...doopRpcOMFollowerReadFailoverProxyProvider.java | 176 ++++++++++-----------
.../om/protocolPB/OzoneManagerProtocolPB.java | 3 +-
...doopRpcOMFollowerReadFailoverProxyProvider.java | 123 +++++++++++---
8 files changed, 254 insertions(+), 119 deletions(-)
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 604b82ea2a6..f421a1dbd1b 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 56d647f5eb7..e26bae6df7a 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 bc91d2bd16e..f3f39b22907 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 00000000000..3bdb21a8ce1
--- /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 4239d49a02c..f6a10cfa7b3 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 38a7bbbb5bb..feed03f54a7 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<OzoneManagerProtocolPB> 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.
+ * <p>
+ * Proxy/provider structure:
+ * <pre>{@code
+ * ReadConsistencyProxy
+ * └── RetryProxy
+ * └── HadoopRpcOMFollowerReadFailoverProxyProvider
+ * ├── FollowerReadProxy
+ * └── HadoopRpcOMFailoverProxyProvider (leaderProxy)
+ * }</pre>
*/
- 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<OzoneManagerProtocolPB> 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.
* <p>
* 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<OzoneManagerProtocolPB> changeProxy(OMProxyInfo
* <p>
* 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<OzoneManagerProtocolPB> 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<OzoneManagerProtocolPB> 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 d11aa7e6aab..22220b8744d 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<OzoneManagerP
static OzoneManagerProtocolPB newProxy(HadoopRpcOMFollowerReadFailoverProxyProvider
followerReadFailoverProxyProvider, int maxFailovers) {
- return (OzoneManagerProtocolPB) RetryProxy.create(OzoneManagerProtocolPB.class, followerReadFailoverProxyProvider,
- followerReadFailoverProxyProvider.getRetryPolicy(maxFailovers));
+ return followerReadFailoverProxyProvider.newProxy(maxFailovers);
}
}
diff --git a/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/ha/TestHadoopRpcOMFollowerReadFailoverProxyProvider.java b/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/ha/TestHadoopRpcOMFollowerReadFailoverProxyProvider.java
index 403dfe8ef3c..9c9f573eea5 100644
--- a/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/ha/TestHadoopRpcOMFollowerReadFailoverProxyProvider.java
+++ b/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/ha/TestHadoopRpcOMFollowerReadFailoverProxyProvider.java
@@ -29,18 +29,21 @@
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.withSettings;
import com.google.protobuf.RpcController;
import com.google.protobuf.ServiceException;
import java.io.IOException;
-import java.lang.reflect.InvocationHandler;
-import java.lang.reflect.Method;
-import java.lang.reflect.Proxy;
+import java.io.InterruptedIOException;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Collection;
@@ -51,20 +54,25 @@
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.io_.retry.RetryInvocationHandler;
-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_.RemoteException;
import org.apache.hadoop.ipc_.RpcNoSuchProtocolException;
+import org.apache.hadoop.ipc_.RpcProxy;
import org.apache.hadoop.ozone.ClientVersion;
import org.apache.hadoop.ozone.OmUtils;
import org.apache.hadoop.ozone.ha.ConfUtils;
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.CreateKeyRequest;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.GetKeyInfoRequest;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.KeyArgs;
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.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Status;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.util.Time;
@@ -72,6 +80,8 @@
import org.apache.ratis.protocol.exceptions.ReadException;
import org.apache.ratis.protocol.exceptions.ReadIndexException;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
@@ -87,6 +97,7 @@ public class TestHadoopRpcOMFollowerReadFailoverProxyProvider {
private OzoneManagerProtocolPB retryProxy;
private String[] omNodeIds;
private OMAnswer[] omNodeAnswers;
+ private OzoneManagerProtocolPB[] proxies;
@Test
void testWriteOperationOnLeader() throws Exception {
@@ -96,6 +107,7 @@ void testWriteOperationOnLeader() throws Exception {
doWrite();
assertHandledBy(2);
+ assertEquals(ReadConsistency.DEFAULT.getHint(), omNodeAnswers[2].lastRequest.getReadConsistencyHint());
assertTrue(proxyProvider.isUseFollowerRead());
// Although the write request is forwarded to the leader,
// the follower read proxy provider should still point to first OM follower
@@ -315,21 +327,45 @@ void testObjectMethodsOnProxy() throws Exception {
void testObjectMethodsDoNotSelectProxy() throws Exception {
setupProxyProvider(2);
+ OzoneManagerProtocolPB routingProxy = proxyProvider.getProxy().proxy;
+ assertEquals(routingProxy.getClass().getName() + "@" + Integer.toHexString(System.identityHashCode(routingProxy)),
+ routingProxy.toString());
+ assertEquals(routingProxy.hashCode(), routingProxy.hashCode());
+ assertNotEquals((Object) routingProxy, new Object());
+ assertEquals(retryProxy.getClass().getName() + "@" + Integer.toHexString(System.identityHashCode(retryProxy)),
+ retryProxy.toString());
+ assertEquals(retryProxy.hashCode(), retryProxy.hashCode());
+ assertNotEquals((Object) retryProxy, new Object());
+ for (OMProxyInfo<OzoneManagerProtocolPB> 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<OMProxyInfo<OzoneManagerProtocolPB>> 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) {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@ozone.apache.org
For additional commands, e-mail: commits-help@ozone.apache.org