szetszwo commented on code in PR #9641: URL: https://github.com/apache/ozone/pull/9641#discussion_r2713992535
########## hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java: ########## @@ -0,0 +1,414 @@ +/* + * 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.ozone.om.ha; + +import static org.apache.hadoop.ozone.om.ha.OMFailoverProxyProviderBase.getLeaderNotReadyException; +import static org.apache.hadoop.ozone.om.ha.OMFailoverProxyProviderBase.getNotLeaderException; + +import com.google.common.annotations.VisibleForTesting; +import com.google.protobuf.Message; +import com.google.protobuf.RpcController; +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.hdds.conf.ConfigurationSource; +import org.apache.hadoop.io.retry.FailoverProxyProvider; +import org.apache.hadoop.io.retry.RetryPolicy; +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.ozone.OmUtils; +import org.apache.hadoop.ozone.om.exceptions.OMLeaderNotReadyException; +import org.apache.hadoop.ozone.om.exceptions.OMNotLeaderException; +import org.apache.hadoop.ozone.om.protocolPB.OzoneManagerProtocolPB; +import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest; +import org.apache.hadoop.security.UserGroupInformation; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A {@link org.apache.hadoop.io.retry.FailoverProxyProvider} implementation + * that supports reading from follower OM(s) (i.e. non-leader OMs also includes + * OM listeners). + * <p> + * This constructs a wrapper proxy might send the read request to follower + * OM(s), if follower read is enabled. It will try to send read requests + * to the first OM node. If RPC failed, it will try to failover to the next OM node. + * It will fail back to the leader OM after it has exhausted all the OMs. + * TODO: Currently the logic does not prioritize forwarding to followers since + * it requires an extra RPC latency to check the OM role info. + * In the future, we can try to try to pick the followers before forwarding + * the request to the leader (similar to ObserverReadProxyProvider). + * <p> + * Read and write requests will still be sent to leader OM if reading from + * follower is disabled. + */ +public class HadoopRpcOMFollowerReadFailoverProxyProvider<T> implements FailoverProxyProvider<T> { + @VisibleForTesting + public static final Logger LOG = LoggerFactory.getLogger(HadoopRpcOMFollowerReadFailoverProxyProvider.class); + + private final Class<T> protocolClass; + + /** The inner proxy provider used for leader-based failover. */ + private final HadoopRpcOMFailoverProxyProvider<T> failoverProxy; + + /** The combined proxy which redirects to other proxies as necessary. */ + private final ProxyInfo<T> combinedProxy; + + /** + * Whether reading from follower is enabled. If this is false, all read + * requests will still go to OM leader. + */ + private volatile boolean followerReadEnabled; + + /** + * The current index of the underlying leader-based proxy provider's omNodesInOrder currently being used. + * Should only be accessed in synchronized methods. + */ + private int currentIndex = -1; + + /** + * The proxy currently being used to send the read request. + * Should only be accessed in synchronized methods. + */ + private OMProxyInfo<T> currentProxy; + + /** The last proxy that has been used. Only used for testing. */ + private volatile OMProxyInfo<T> lastProxy = null; + + public HadoopRpcOMFollowerReadFailoverProxyProvider( + ConfigurationSource configuration, UserGroupInformation ugi, String omServiceId, Class<T> protocol) + throws IOException { + this(omServiceId, protocol, + new HadoopRpcOMFailoverProxyProvider<>(configuration, ugi, omServiceId, protocol)); + } + + @SuppressWarnings("unchecked") + public HadoopRpcOMFollowerReadFailoverProxyProvider(String omServiceId, Class<T> protocol, + HadoopRpcOMFailoverProxyProvider<T> failoverProxy) throws IOException { + this.protocolClass = protocol; + this.failoverProxy = failoverProxy; + + // Create a wrapped proxy containing all the proxies. Since this combined + // proxy is just redirecting to other proxies, all invocations can share it. + StringBuilder combinedInfo = new StringBuilder("["); + for (int i = 0; i < failoverProxy.getOMProxies().size(); i++) { + if (i > 0) { + combinedInfo.append(','); + } + combinedInfo.append(failoverProxy.getOMProxies().get(i).proxyInfo); + } + combinedInfo.append(']'); Review Comment: We may use "map/reduce". ```java final String combinedInfo = "[" + failoverProxy.getOMProxies().stream() .map(a -> a.proxyInfo) .reduce((a, b) -> a + ", " + b).orElse("") + "]"; ``` ########## hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java: ########## @@ -0,0 +1,414 @@ +/* + * 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.ozone.om.ha; + +import static org.apache.hadoop.ozone.om.ha.OMFailoverProxyProviderBase.getLeaderNotReadyException; +import static org.apache.hadoop.ozone.om.ha.OMFailoverProxyProviderBase.getNotLeaderException; + +import com.google.common.annotations.VisibleForTesting; +import com.google.protobuf.Message; +import com.google.protobuf.RpcController; +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.hdds.conf.ConfigurationSource; +import org.apache.hadoop.io.retry.FailoverProxyProvider; +import org.apache.hadoop.io.retry.RetryPolicy; +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.ozone.OmUtils; +import org.apache.hadoop.ozone.om.exceptions.OMLeaderNotReadyException; +import org.apache.hadoop.ozone.om.exceptions.OMNotLeaderException; +import org.apache.hadoop.ozone.om.protocolPB.OzoneManagerProtocolPB; +import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest; +import org.apache.hadoop.security.UserGroupInformation; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A {@link org.apache.hadoop.io.retry.FailoverProxyProvider} implementation + * that supports reading from follower OM(s) (i.e. non-leader OMs also includes + * OM listeners). + * <p> + * This constructs a wrapper proxy might send the read request to follower + * OM(s), if follower read is enabled. It will try to send read requests + * to the first OM node. If RPC failed, it will try to failover to the next OM node. + * It will fail back to the leader OM after it has exhausted all the OMs. + * TODO: Currently the logic does not prioritize forwarding to followers since + * it requires an extra RPC latency to check the OM role info. + * In the future, we can try to try to pick the followers before forwarding + * the request to the leader (similar to ObserverReadProxyProvider). + * <p> + * Read and write requests will still be sent to leader OM if reading from + * follower is disabled. + */ +public class HadoopRpcOMFollowerReadFailoverProxyProvider<T> implements FailoverProxyProvider<T> { + @VisibleForTesting + public static final Logger LOG = LoggerFactory.getLogger(HadoopRpcOMFollowerReadFailoverProxyProvider.class); + + private final Class<T> protocolClass; + + /** The inner proxy provider used for leader-based failover. */ + private final HadoopRpcOMFailoverProxyProvider<T> failoverProxy; + + /** The combined proxy which redirects to other proxies as necessary. */ + private final ProxyInfo<T> combinedProxy; + + /** + * Whether reading from follower is enabled. If this is false, all read + * requests will still go to OM leader. + */ + private volatile boolean followerReadEnabled; + + /** + * The current index of the underlying leader-based proxy provider's omNodesInOrder currently being used. + * Should only be accessed in synchronized methods. + */ + private int currentIndex = -1; + + /** + * The proxy currently being used to send the read request. + * Should only be accessed in synchronized methods. + */ + private OMProxyInfo<T> currentProxy; + + /** The last proxy that has been used. Only used for testing. */ + private volatile OMProxyInfo<T> lastProxy = null; + + public HadoopRpcOMFollowerReadFailoverProxyProvider( + ConfigurationSource configuration, UserGroupInformation ugi, String omServiceId, Class<T> protocol) + throws IOException { + this(omServiceId, protocol, + new HadoopRpcOMFailoverProxyProvider<>(configuration, ugi, omServiceId, protocol)); + } + + @SuppressWarnings("unchecked") Review Comment: Move it to `wrappedProxy`. Then it would only suppress the warning there but not the entire method. ```diff +++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java @@ -103,7 +103,6 @@ public HadoopRpcOMFollowerReadFailoverProxyProvider( new HadoopRpcOMFailoverProxyProvider<>(configuration, ugi, omServiceId, protocol)); } - @SuppressWarnings("unchecked") public HadoopRpcOMFollowerReadFailoverProxyProvider(String omServiceId, Class<T> protocol, HadoopRpcOMFailoverProxyProvider<T> failoverProxy) throws IOException { this.protocolClass = protocol; @@ -119,6 +118,7 @@ public HadoopRpcOMFollowerReadFailoverProxyProvider(String omServiceId, Class<T> combinedInfo.append(failoverProxy.getOMProxies().get(i).proxyInfo); } combinedInfo.append(']'); + @SuppressWarnings("unchecked") T wrappedProxy = (T) Proxy.newProxyInstance( FollowerReadInvocationHandler.class.getClassLoader(), new Class<?>[] {protocol}, new FollowerReadInvocationHandler()); ``` ########## hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java: ########## @@ -0,0 +1,414 @@ +/* + * 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.ozone.om.ha; + +import static org.apache.hadoop.ozone.om.ha.OMFailoverProxyProviderBase.getLeaderNotReadyException; +import static org.apache.hadoop.ozone.om.ha.OMFailoverProxyProviderBase.getNotLeaderException; + +import com.google.common.annotations.VisibleForTesting; +import com.google.protobuf.Message; +import com.google.protobuf.RpcController; +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.hdds.conf.ConfigurationSource; +import org.apache.hadoop.io.retry.FailoverProxyProvider; +import org.apache.hadoop.io.retry.RetryPolicy; +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.ozone.OmUtils; +import org.apache.hadoop.ozone.om.exceptions.OMLeaderNotReadyException; +import org.apache.hadoop.ozone.om.exceptions.OMNotLeaderException; +import org.apache.hadoop.ozone.om.protocolPB.OzoneManagerProtocolPB; +import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest; +import org.apache.hadoop.security.UserGroupInformation; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A {@link org.apache.hadoop.io.retry.FailoverProxyProvider} implementation + * that supports reading from follower OM(s) (i.e. non-leader OMs also includes + * OM listeners). + * <p> + * This constructs a wrapper proxy might send the read request to follower + * OM(s), if follower read is enabled. It will try to send read requests + * to the first OM node. If RPC failed, it will try to failover to the next OM node. + * It will fail back to the leader OM after it has exhausted all the OMs. + * TODO: Currently the logic does not prioritize forwarding to followers since + * it requires an extra RPC latency to check the OM role info. + * In the future, we can try to try to pick the followers before forwarding + * the request to the leader (similar to ObserverReadProxyProvider). + * <p> + * Read and write requests will still be sent to leader OM if reading from + * follower is disabled. + */ +public class HadoopRpcOMFollowerReadFailoverProxyProvider<T> implements FailoverProxyProvider<T> { + @VisibleForTesting + public static final Logger LOG = LoggerFactory.getLogger(HadoopRpcOMFollowerReadFailoverProxyProvider.class); + + private final Class<T> protocolClass; + + /** The inner proxy provider used for leader-based failover. */ + private final HadoopRpcOMFailoverProxyProvider<T> failoverProxy; + + /** The combined proxy which redirects to other proxies as necessary. */ + private final ProxyInfo<T> combinedProxy; + + /** + * Whether reading from follower is enabled. If this is false, all read + * requests will still go to OM leader. + */ + private volatile boolean followerReadEnabled; + + /** + * The current index of the underlying leader-based proxy provider's omNodesInOrder currently being used. + * Should only be accessed in synchronized methods. + */ + private int currentIndex = -1; + + /** + * The proxy currently being used to send the read request. + * Should only be accessed in synchronized methods. + */ + private OMProxyInfo<T> currentProxy; + + /** The last proxy that has been used. Only used for testing. */ + private volatile OMProxyInfo<T> lastProxy = null; + + public HadoopRpcOMFollowerReadFailoverProxyProvider( + ConfigurationSource configuration, UserGroupInformation ugi, String omServiceId, Class<T> protocol) + throws IOException { + this(omServiceId, protocol, + new HadoopRpcOMFailoverProxyProvider<>(configuration, ugi, omServiceId, protocol)); + } + + @SuppressWarnings("unchecked") + public HadoopRpcOMFollowerReadFailoverProxyProvider(String omServiceId, Class<T> protocol, + HadoopRpcOMFailoverProxyProvider<T> failoverProxy) throws IOException { + this.protocolClass = protocol; + this.failoverProxy = failoverProxy; + + // Create a wrapped proxy containing all the proxies. Since this combined + // proxy is just redirecting to other proxies, all invocations can share it. + StringBuilder combinedInfo = new StringBuilder("["); + for (int i = 0; i < failoverProxy.getOMProxies().size(); i++) { + if (i > 0) { + combinedInfo.append(','); + } + combinedInfo.append(failoverProxy.getOMProxies().get(i).proxyInfo); + } + combinedInfo.append(']'); + T wrappedProxy = (T) Proxy.newProxyInstance( + FollowerReadInvocationHandler.class.getClassLoader(), + new Class<?>[] {protocol}, new FollowerReadInvocationHandler()); + combinedProxy = new ProxyInfo<>(wrappedProxy, combinedInfo.toString()); + + if (wrappedProxy instanceof OzoneManagerProtocolPB) { + this.followerReadEnabled = true; + } else { + LOG.debug("Disabling follower reads for {} because the requested proxy " + + "class does not implement {}", omServiceId, OzoneManagerProtocolPB.class.getName()); + this.followerReadEnabled = false; + } + } + + @Override + public Class<T> getInterface() { + return protocolClass; + } + + @Override + public ProxyInfo<T> getProxy() { + return combinedProxy; + } + + @Override + public void performFailover(T currProxy) { + // Since FollowerReadInvocationHandler might user or fallback to leader-based failover logic, + // we should delegate the failover logic to the leader's failover. + failoverProxy.performFailover(currProxy); + } + + public RetryPolicy getRetryPolicy(int maxFailovers) { + // We use the OMFailoverProxyProviderBase's RetryPolicy instead of using our own retry policy + // 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 + // OMFailoverProxyProviderbase's RetryPolicy + return failoverProxy.getRetryPolicy(maxFailovers); + } + + /** + * Parse the OM request from the request args. + * + * @return parsed OM request. + */ + private static OMRequest parseOMRequest(Object[] args) throws Throwable { + if (args == null || args.length < 2 || !(args[1] instanceof Message)) { + LOG.error("Request failed since OM request is null and cannot be parsed"); + // Throws a non-retriable exception to prevent retry and failover + // See the HddsUtils#shouldNotFailoverOnRpcException used in + // OMFailoverProxyProviderBase#shouldFailover + throw wrapInServiceException( + new RpcNoSuchProtocolException("OM request is null and cannot be parsed")); + } + final Message theRequest = (Message) args[1]; + return (OMRequest) theRequest; + } Review Comment: Let's provide a more specific error message: ```java private static OMRequest parseOMRequest(Object[] args) throws ServiceException { final String error = args == null ? "args == null" : args.length < 2 ? "args.length == " + args.length + " < 2" : !(args[1] instanceof OMRequest) ? "Non-OMRequest: " + args[1].getClass() : null; if (error != null) { // Throws a non-retriable exception to prevent retry and failover // See the HddsUtils#shouldNotFailoverOnRpcException used in // OMFailoverProxyProviderBase#shouldFailover throw wrapInServiceException(new RpcNoSuchProtocolException("Failed to parseOMRequest: " + error)); } return (OMRequest) args[1]; } ``` ########## hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java: ########## @@ -0,0 +1,414 @@ +/* + * 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.ozone.om.ha; + +import static org.apache.hadoop.ozone.om.ha.OMFailoverProxyProviderBase.getLeaderNotReadyException; +import static org.apache.hadoop.ozone.om.ha.OMFailoverProxyProviderBase.getNotLeaderException; + +import com.google.common.annotations.VisibleForTesting; +import com.google.protobuf.Message; +import com.google.protobuf.RpcController; +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.hdds.conf.ConfigurationSource; +import org.apache.hadoop.io.retry.FailoverProxyProvider; +import org.apache.hadoop.io.retry.RetryPolicy; +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.ozone.OmUtils; +import org.apache.hadoop.ozone.om.exceptions.OMLeaderNotReadyException; +import org.apache.hadoop.ozone.om.exceptions.OMNotLeaderException; +import org.apache.hadoop.ozone.om.protocolPB.OzoneManagerProtocolPB; +import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest; +import org.apache.hadoop.security.UserGroupInformation; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A {@link org.apache.hadoop.io.retry.FailoverProxyProvider} implementation + * that supports reading from follower OM(s) (i.e. non-leader OMs also includes + * OM listeners). + * <p> + * This constructs a wrapper proxy might send the read request to follower + * OM(s), if follower read is enabled. It will try to send read requests + * to the first OM node. If RPC failed, it will try to failover to the next OM node. + * It will fail back to the leader OM after it has exhausted all the OMs. + * TODO: Currently the logic does not prioritize forwarding to followers since + * it requires an extra RPC latency to check the OM role info. + * In the future, we can try to try to pick the followers before forwarding + * the request to the leader (similar to ObserverReadProxyProvider). + * <p> + * Read and write requests will still be sent to leader OM if reading from + * follower is disabled. + */ +public class HadoopRpcOMFollowerReadFailoverProxyProvider<T> implements FailoverProxyProvider<T> { + @VisibleForTesting + public static final Logger LOG = LoggerFactory.getLogger(HadoopRpcOMFollowerReadFailoverProxyProvider.class); + + private final Class<T> protocolClass; + + /** The inner proxy provider used for leader-based failover. */ + private final HadoopRpcOMFailoverProxyProvider<T> failoverProxy; + + /** The combined proxy which redirects to other proxies as necessary. */ + private final ProxyInfo<T> combinedProxy; + + /** + * Whether reading from follower is enabled. If this is false, all read + * requests will still go to OM leader. + */ + private volatile boolean followerReadEnabled; + + /** + * The current index of the underlying leader-based proxy provider's omNodesInOrder currently being used. + * Should only be accessed in synchronized methods. + */ + private int currentIndex = -1; + + /** + * The proxy currently being used to send the read request. + * Should only be accessed in synchronized methods. + */ + private OMProxyInfo<T> currentProxy; + + /** The last proxy that has been used. Only used for testing. */ + private volatile OMProxyInfo<T> lastProxy = null; + + public HadoopRpcOMFollowerReadFailoverProxyProvider( + ConfigurationSource configuration, UserGroupInformation ugi, String omServiceId, Class<T> protocol) + throws IOException { + this(omServiceId, protocol, + new HadoopRpcOMFailoverProxyProvider<>(configuration, ugi, omServiceId, protocol)); + } + + @SuppressWarnings("unchecked") + public HadoopRpcOMFollowerReadFailoverProxyProvider(String omServiceId, Class<T> protocol, + HadoopRpcOMFailoverProxyProvider<T> failoverProxy) throws IOException { + this.protocolClass = protocol; + this.failoverProxy = failoverProxy; + + // Create a wrapped proxy containing all the proxies. Since this combined + // proxy is just redirecting to other proxies, all invocations can share it. + StringBuilder combinedInfo = new StringBuilder("["); + for (int i = 0; i < failoverProxy.getOMProxies().size(); i++) { + if (i > 0) { + combinedInfo.append(','); + } + combinedInfo.append(failoverProxy.getOMProxies().get(i).proxyInfo); + } + combinedInfo.append(']'); + T wrappedProxy = (T) Proxy.newProxyInstance( + FollowerReadInvocationHandler.class.getClassLoader(), + new Class<?>[] {protocol}, new FollowerReadInvocationHandler()); + combinedProxy = new ProxyInfo<>(wrappedProxy, combinedInfo.toString()); + + if (wrappedProxy instanceof OzoneManagerProtocolPB) { + this.followerReadEnabled = true; + } else { + LOG.debug("Disabling follower reads for {} because the requested proxy " + + "class does not implement {}", omServiceId, OzoneManagerProtocolPB.class.getName()); + this.followerReadEnabled = false; + } + } + + @Override + public Class<T> getInterface() { + return protocolClass; + } + + @Override + public ProxyInfo<T> getProxy() { + return combinedProxy; + } + + @Override + public void performFailover(T currProxy) { + // Since FollowerReadInvocationHandler might user or fallback to leader-based failover logic, + // we should delegate the failover logic to the leader's failover. + failoverProxy.performFailover(currProxy); + } + + public RetryPolicy getRetryPolicy(int maxFailovers) { + // We use the OMFailoverProxyProviderBase's RetryPolicy instead of using our own retry policy + // 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 + // OMFailoverProxyProviderbase's RetryPolicy + return failoverProxy.getRetryPolicy(maxFailovers); + } + + /** + * Parse the OM request from the request args. + * + * @return parsed OM request. + */ + private static OMRequest parseOMRequest(Object[] args) throws Throwable { + if (args == null || args.length < 2 || !(args[1] instanceof Message)) { + LOG.error("Request failed since OM request is null and cannot be parsed"); + // Throws a non-retriable exception to prevent retry and failover + // See the HddsUtils#shouldNotFailoverOnRpcException used in + // OMFailoverProxyProviderBase#shouldFailover + throw wrapInServiceException( + new RpcNoSuchProtocolException("OM request is null and cannot be parsed")); + } + final Message theRequest = (Message) args[1]; + return (OMRequest) theRequest; + } + + @VisibleForTesting + void setFollowerReadEnabled(boolean flag) { + this.followerReadEnabled = flag; + } + + @VisibleForTesting + public ProxyInfo<T> getLastProxy() { + return lastProxy; + } + + /** + * Return the currently used proxy. If there is none, first calls + * {@link #changeProxy(OMProxyInfo)} to initialize one. + */ + @VisibleForTesting + public OMProxyInfo<T> getCurrentProxy() { + return changeProxy(null); + } + + /** + * Move to the next proxy in the proxy list. If the OMProxyInfo supplied by + * the caller does not match the current proxy, the call is ignored; this is + * to handle concurrent calls (to avoid changing the proxy multiple times). + * The service state of the newly selected proxy will be updated before + * returning. + * + * @param initial The expected current proxy + * @return The new proxy that should be used. + */ + private synchronized OMProxyInfo<T> changeProxy(OMProxyInfo<T> initial) { + if (currentProxy != initial) { + // Must have been a concurrent modification; ignore the move request + return currentProxy; + } + currentIndex = (currentIndex + 1) % failoverProxy.getOmNodesInOrder().size(); + String currentOmNodeId = failoverProxy.getOmNodesInOrder().get(currentIndex); + currentProxy = (OMProxyInfo<T>) failoverProxy.createOMProxyIfNeeded(currentOmNodeId); + LOG.debug("Changed current proxy from {} to {}", + initial == null ? "none" : initial.proxyInfo, + currentProxy.proxyInfo); + return currentProxy; + } + + /** + * An InvocationHandler to handle incoming requests. This class's invoke + * method contains the primary logic for redirecting 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 + * proxy's OM node fails, adjust the current proxy and return on the next one. + * <p> + * Write requests are always forwarded to the leader. + */ + private class FollowerReadInvocationHandler implements RpcInvocationHandler { + + @Override + public Object invoke(Object proxy, final Method method, final Object[] args) + throws Throwable { + 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); + } + Object retVal; + OMRequest omRequest = parseOMRequest(args); + if (followerReadEnabled && OmUtils.shouldSendToFollower(omRequest)) { + int failedCount = 0; + for (int i = 0; i < failoverProxy.getOmNodesInOrder().size(); i++) { Review Comment: Check followerReadEnabled: `followerReadEnabled && i < failoverProxy.getOmNodesInOrder().size()` ########## hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java: ########## @@ -0,0 +1,414 @@ +/* + * 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.ozone.om.ha; + +import static org.apache.hadoop.ozone.om.ha.OMFailoverProxyProviderBase.getLeaderNotReadyException; +import static org.apache.hadoop.ozone.om.ha.OMFailoverProxyProviderBase.getNotLeaderException; + +import com.google.common.annotations.VisibleForTesting; +import com.google.protobuf.Message; +import com.google.protobuf.RpcController; +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.hdds.conf.ConfigurationSource; +import org.apache.hadoop.io.retry.FailoverProxyProvider; +import org.apache.hadoop.io.retry.RetryPolicy; +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.ozone.OmUtils; +import org.apache.hadoop.ozone.om.exceptions.OMLeaderNotReadyException; +import org.apache.hadoop.ozone.om.exceptions.OMNotLeaderException; +import org.apache.hadoop.ozone.om.protocolPB.OzoneManagerProtocolPB; +import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest; +import org.apache.hadoop.security.UserGroupInformation; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A {@link org.apache.hadoop.io.retry.FailoverProxyProvider} implementation + * that supports reading from follower OM(s) (i.e. non-leader OMs also includes + * OM listeners). + * <p> + * This constructs a wrapper proxy might send the read request to follower + * OM(s), if follower read is enabled. It will try to send read requests + * to the first OM node. If RPC failed, it will try to failover to the next OM node. + * It will fail back to the leader OM after it has exhausted all the OMs. + * TODO: Currently the logic does not prioritize forwarding to followers since + * it requires an extra RPC latency to check the OM role info. + * In the future, we can try to try to pick the followers before forwarding + * the request to the leader (similar to ObserverReadProxyProvider). + * <p> + * Read and write requests will still be sent to leader OM if reading from + * follower is disabled. + */ +public class HadoopRpcOMFollowerReadFailoverProxyProvider<T> implements FailoverProxyProvider<T> { + @VisibleForTesting + public static final Logger LOG = LoggerFactory.getLogger(HadoopRpcOMFollowerReadFailoverProxyProvider.class); + + private final Class<T> protocolClass; + + /** The inner proxy provider used for leader-based failover. */ + private final HadoopRpcOMFailoverProxyProvider<T> failoverProxy; + + /** The combined proxy which redirects to other proxies as necessary. */ + private final ProxyInfo<T> combinedProxy; + + /** + * Whether reading from follower is enabled. If this is false, all read + * requests will still go to OM leader. + */ + private volatile boolean followerReadEnabled; + + /** + * The current index of the underlying leader-based proxy provider's omNodesInOrder currently being used. + * Should only be accessed in synchronized methods. + */ + private int currentIndex = -1; + + /** + * The proxy currently being used to send the read request. + * Should only be accessed in synchronized methods. + */ + private OMProxyInfo<T> currentProxy; + + /** The last proxy that has been used. Only used for testing. */ + private volatile OMProxyInfo<T> lastProxy = null; + + public HadoopRpcOMFollowerReadFailoverProxyProvider( + ConfigurationSource configuration, UserGroupInformation ugi, String omServiceId, Class<T> protocol) + throws IOException { + this(omServiceId, protocol, + new HadoopRpcOMFailoverProxyProvider<>(configuration, ugi, omServiceId, protocol)); + } + + @SuppressWarnings("unchecked") + public HadoopRpcOMFollowerReadFailoverProxyProvider(String omServiceId, Class<T> protocol, Review Comment: For simplicity, let support only `OzoneManagerProtocolPB` for now? Then, we can remove the `followerReadEnabled` and simplify the code. ########## hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFailoverProxyProvider.java: ########## @@ -123,7 +123,7 @@ public synchronized ProxyInfo<T> getProxy() { /** * Creates proxy object. */ - protected ProxyInfo<T> createOMProxyIfNeeded(OMProxyInfo<T> omProxyInfo) { + protected synchronized ProxyInfo<T> createOMProxyIfNeeded(OMProxyInfo<T> omProxyInfo) { if (omProxyInfo.proxy == null) { Review Comment: The `synchronized` is for `omProxyInfo.proxy`. We probably should do it in `OMProxyInfo`. Let me fix it in HDDS-14470. ########## hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java: ########## @@ -0,0 +1,414 @@ +/* + * 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.ozone.om.ha; + +import static org.apache.hadoop.ozone.om.ha.OMFailoverProxyProviderBase.getLeaderNotReadyException; +import static org.apache.hadoop.ozone.om.ha.OMFailoverProxyProviderBase.getNotLeaderException; + +import com.google.common.annotations.VisibleForTesting; +import com.google.protobuf.Message; +import com.google.protobuf.RpcController; +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.hdds.conf.ConfigurationSource; +import org.apache.hadoop.io.retry.FailoverProxyProvider; +import org.apache.hadoop.io.retry.RetryPolicy; +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.ozone.OmUtils; +import org.apache.hadoop.ozone.om.exceptions.OMLeaderNotReadyException; +import org.apache.hadoop.ozone.om.exceptions.OMNotLeaderException; +import org.apache.hadoop.ozone.om.protocolPB.OzoneManagerProtocolPB; +import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest; +import org.apache.hadoop.security.UserGroupInformation; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A {@link org.apache.hadoop.io.retry.FailoverProxyProvider} implementation + * that supports reading from follower OM(s) (i.e. non-leader OMs also includes + * OM listeners). + * <p> + * This constructs a wrapper proxy might send the read request to follower + * OM(s), if follower read is enabled. It will try to send read requests + * to the first OM node. If RPC failed, it will try to failover to the next OM node. + * It will fail back to the leader OM after it has exhausted all the OMs. + * TODO: Currently the logic does not prioritize forwarding to followers since + * it requires an extra RPC latency to check the OM role info. + * In the future, we can try to try to pick the followers before forwarding + * the request to the leader (similar to ObserverReadProxyProvider). + * <p> + * Read and write requests will still be sent to leader OM if reading from + * follower is disabled. + */ +public class HadoopRpcOMFollowerReadFailoverProxyProvider<T> implements FailoverProxyProvider<T> { + @VisibleForTesting + public static final Logger LOG = LoggerFactory.getLogger(HadoopRpcOMFollowerReadFailoverProxyProvider.class); + + private final Class<T> protocolClass; + + /** The inner proxy provider used for leader-based failover. */ + private final HadoopRpcOMFailoverProxyProvider<T> failoverProxy; + + /** The combined proxy which redirects to other proxies as necessary. */ + private final ProxyInfo<T> combinedProxy; + + /** + * Whether reading from follower is enabled. If this is false, all read + * requests will still go to OM leader. + */ + private volatile boolean followerReadEnabled; Review Comment: Let's rename it to `useFollowerRead` to avoid confusion with the conf. ########## hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java: ########## @@ -0,0 +1,414 @@ +/* + * 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.ozone.om.ha; + +import static org.apache.hadoop.ozone.om.ha.OMFailoverProxyProviderBase.getLeaderNotReadyException; +import static org.apache.hadoop.ozone.om.ha.OMFailoverProxyProviderBase.getNotLeaderException; + +import com.google.common.annotations.VisibleForTesting; +import com.google.protobuf.Message; +import com.google.protobuf.RpcController; +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.hdds.conf.ConfigurationSource; +import org.apache.hadoop.io.retry.FailoverProxyProvider; +import org.apache.hadoop.io.retry.RetryPolicy; +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.ozone.OmUtils; +import org.apache.hadoop.ozone.om.exceptions.OMLeaderNotReadyException; +import org.apache.hadoop.ozone.om.exceptions.OMNotLeaderException; +import org.apache.hadoop.ozone.om.protocolPB.OzoneManagerProtocolPB; +import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest; +import org.apache.hadoop.security.UserGroupInformation; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A {@link org.apache.hadoop.io.retry.FailoverProxyProvider} implementation + * that supports reading from follower OM(s) (i.e. non-leader OMs also includes + * OM listeners). + * <p> + * This constructs a wrapper proxy might send the read request to follower + * OM(s), if follower read is enabled. It will try to send read requests + * to the first OM node. If RPC failed, it will try to failover to the next OM node. + * It will fail back to the leader OM after it has exhausted all the OMs. + * TODO: Currently the logic does not prioritize forwarding to followers since + * it requires an extra RPC latency to check the OM role info. + * In the future, we can try to try to pick the followers before forwarding + * the request to the leader (similar to ObserverReadProxyProvider). + * <p> + * Read and write requests will still be sent to leader OM if reading from + * follower is disabled. + */ +public class HadoopRpcOMFollowerReadFailoverProxyProvider<T> implements FailoverProxyProvider<T> { + @VisibleForTesting + public static final Logger LOG = LoggerFactory.getLogger(HadoopRpcOMFollowerReadFailoverProxyProvider.class); + + private final Class<T> protocolClass; + + /** The inner proxy provider used for leader-based failover. */ + private final HadoopRpcOMFailoverProxyProvider<T> failoverProxy; + + /** The combined proxy which redirects to other proxies as necessary. */ + private final ProxyInfo<T> combinedProxy; + + /** + * Whether reading from follower is enabled. If this is false, all read + * requests will still go to OM leader. + */ + private volatile boolean followerReadEnabled; + + /** + * The current index of the underlying leader-based proxy provider's omNodesInOrder currently being used. + * Should only be accessed in synchronized methods. + */ + private int currentIndex = -1; + + /** + * The proxy currently being used to send the read request. + * Should only be accessed in synchronized methods. + */ + private OMProxyInfo<T> currentProxy; + + /** The last proxy that has been used. Only used for testing. */ + private volatile OMProxyInfo<T> lastProxy = null; + + public HadoopRpcOMFollowerReadFailoverProxyProvider( + ConfigurationSource configuration, UserGroupInformation ugi, String omServiceId, Class<T> protocol) + throws IOException { + this(omServiceId, protocol, + new HadoopRpcOMFailoverProxyProvider<>(configuration, ugi, omServiceId, protocol)); + } + + @SuppressWarnings("unchecked") + public HadoopRpcOMFollowerReadFailoverProxyProvider(String omServiceId, Class<T> protocol, + HadoopRpcOMFailoverProxyProvider<T> failoverProxy) throws IOException { + this.protocolClass = protocol; + this.failoverProxy = failoverProxy; + + // Create a wrapped proxy containing all the proxies. Since this combined + // proxy is just redirecting to other proxies, all invocations can share it. + StringBuilder combinedInfo = new StringBuilder("["); + for (int i = 0; i < failoverProxy.getOMProxies().size(); i++) { + if (i > 0) { + combinedInfo.append(','); + } + combinedInfo.append(failoverProxy.getOMProxies().get(i).proxyInfo); + } + combinedInfo.append(']'); + T wrappedProxy = (T) Proxy.newProxyInstance( + FollowerReadInvocationHandler.class.getClassLoader(), + new Class<?>[] {protocol}, new FollowerReadInvocationHandler()); + combinedProxy = new ProxyInfo<>(wrappedProxy, combinedInfo.toString()); + + if (wrappedProxy instanceof OzoneManagerProtocolPB) { + this.followerReadEnabled = true; + } else { + LOG.debug("Disabling follower reads for {} because the requested proxy " + + "class does not implement {}", omServiceId, OzoneManagerProtocolPB.class.getName()); + this.followerReadEnabled = false; + } + } + + @Override + public Class<T> getInterface() { + return protocolClass; + } + + @Override + public ProxyInfo<T> getProxy() { + return combinedProxy; + } + + @Override + public void performFailover(T currProxy) { + // Since FollowerReadInvocationHandler might user or fallback to leader-based failover logic, + // we should delegate the failover logic to the leader's failover. + failoverProxy.performFailover(currProxy); + } + + public RetryPolicy getRetryPolicy(int maxFailovers) { + // We use the OMFailoverProxyProviderBase's RetryPolicy instead of using our own retry policy + // 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 + // OMFailoverProxyProviderbase's RetryPolicy + return failoverProxy.getRetryPolicy(maxFailovers); + } + + /** + * Parse the OM request from the request args. + * + * @return parsed OM request. + */ + private static OMRequest parseOMRequest(Object[] args) throws Throwable { + if (args == null || args.length < 2 || !(args[1] instanceof Message)) { + LOG.error("Request failed since OM request is null and cannot be parsed"); + // Throws a non-retriable exception to prevent retry and failover + // See the HddsUtils#shouldNotFailoverOnRpcException used in + // OMFailoverProxyProviderBase#shouldFailover + throw wrapInServiceException( + new RpcNoSuchProtocolException("OM request is null and cannot be parsed")); + } + final Message theRequest = (Message) args[1]; + return (OMRequest) theRequest; + } + + @VisibleForTesting + void setFollowerReadEnabled(boolean flag) { + this.followerReadEnabled = flag; + } + + @VisibleForTesting + public ProxyInfo<T> getLastProxy() { + return lastProxy; + } + + /** + * Return the currently used proxy. If there is none, first calls + * {@link #changeProxy(OMProxyInfo)} to initialize one. + */ + @VisibleForTesting + public OMProxyInfo<T> getCurrentProxy() { + return changeProxy(null); + } + + /** + * Move to the next proxy in the proxy list. If the OMProxyInfo supplied by + * the caller does not match the current proxy, the call is ignored; this is + * to handle concurrent calls (to avoid changing the proxy multiple times). + * The service state of the newly selected proxy will be updated before + * returning. + * + * @param initial The expected current proxy + * @return The new proxy that should be used. + */ + private synchronized OMProxyInfo<T> changeProxy(OMProxyInfo<T> initial) { + if (currentProxy != initial) { + // Must have been a concurrent modification; ignore the move request + return currentProxy; + } + currentIndex = (currentIndex + 1) % failoverProxy.getOmNodesInOrder().size(); + String currentOmNodeId = failoverProxy.getOmNodesInOrder().get(currentIndex); + currentProxy = (OMProxyInfo<T>) failoverProxy.createOMProxyIfNeeded(currentOmNodeId); + LOG.debug("Changed current proxy from {} to {}", + initial == null ? "none" : initial.proxyInfo, + currentProxy.proxyInfo); + return currentProxy; + } + + /** + * An InvocationHandler to handle incoming requests. This class's invoke + * method contains the primary logic for redirecting 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 + * proxy's OM node fails, adjust the current proxy and return on the next one. + * <p> + * Write requests are always forwarded to the leader. + */ + private class FollowerReadInvocationHandler implements RpcInvocationHandler { + + @Override + public Object invoke(Object proxy, final Method method, final Object[] args) + throws Throwable { + lastProxy = null; + if (method.getDeclaringClass() == Object.class) { Review Comment: Check `(method.getDeclaringClass() != OzoneManagerProtocolPB.class)` instead? ########## hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java: ########## @@ -0,0 +1,414 @@ +/* + * 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.ozone.om.ha; + +import static org.apache.hadoop.ozone.om.ha.OMFailoverProxyProviderBase.getLeaderNotReadyException; +import static org.apache.hadoop.ozone.om.ha.OMFailoverProxyProviderBase.getNotLeaderException; + +import com.google.common.annotations.VisibleForTesting; +import com.google.protobuf.Message; +import com.google.protobuf.RpcController; +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.hdds.conf.ConfigurationSource; +import org.apache.hadoop.io.retry.FailoverProxyProvider; +import org.apache.hadoop.io.retry.RetryPolicy; +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.ozone.OmUtils; +import org.apache.hadoop.ozone.om.exceptions.OMLeaderNotReadyException; +import org.apache.hadoop.ozone.om.exceptions.OMNotLeaderException; +import org.apache.hadoop.ozone.om.protocolPB.OzoneManagerProtocolPB; +import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest; +import org.apache.hadoop.security.UserGroupInformation; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A {@link org.apache.hadoop.io.retry.FailoverProxyProvider} implementation + * that supports reading from follower OM(s) (i.e. non-leader OMs also includes + * OM listeners). + * <p> + * This constructs a wrapper proxy might send the read request to follower + * OM(s), if follower read is enabled. It will try to send read requests + * to the first OM node. If RPC failed, it will try to failover to the next OM node. + * It will fail back to the leader OM after it has exhausted all the OMs. + * TODO: Currently the logic does not prioritize forwarding to followers since + * it requires an extra RPC latency to check the OM role info. + * In the future, we can try to try to pick the followers before forwarding + * the request to the leader (similar to ObserverReadProxyProvider). + * <p> + * Read and write requests will still be sent to leader OM if reading from + * follower is disabled. + */ +public class HadoopRpcOMFollowerReadFailoverProxyProvider<T> implements FailoverProxyProvider<T> { + @VisibleForTesting + public static final Logger LOG = LoggerFactory.getLogger(HadoopRpcOMFollowerReadFailoverProxyProvider.class); + + private final Class<T> protocolClass; + + /** The inner proxy provider used for leader-based failover. */ + private final HadoopRpcOMFailoverProxyProvider<T> failoverProxy; + + /** The combined proxy which redirects to other proxies as necessary. */ + private final ProxyInfo<T> combinedProxy; + + /** + * Whether reading from follower is enabled. If this is false, all read + * requests will still go to OM leader. + */ + private volatile boolean followerReadEnabled; + + /** + * The current index of the underlying leader-based proxy provider's omNodesInOrder currently being used. + * Should only be accessed in synchronized methods. + */ + private int currentIndex = -1; + + /** + * The proxy currently being used to send the read request. + * Should only be accessed in synchronized methods. + */ + private OMProxyInfo<T> currentProxy; + + /** The last proxy that has been used. Only used for testing. */ + private volatile OMProxyInfo<T> lastProxy = null; + + public HadoopRpcOMFollowerReadFailoverProxyProvider( + ConfigurationSource configuration, UserGroupInformation ugi, String omServiceId, Class<T> protocol) + throws IOException { + this(omServiceId, protocol, + new HadoopRpcOMFailoverProxyProvider<>(configuration, ugi, omServiceId, protocol)); + } + + @SuppressWarnings("unchecked") + public HadoopRpcOMFollowerReadFailoverProxyProvider(String omServiceId, Class<T> protocol, + HadoopRpcOMFailoverProxyProvider<T> failoverProxy) throws IOException { + this.protocolClass = protocol; + this.failoverProxy = failoverProxy; + + // Create a wrapped proxy containing all the proxies. Since this combined + // proxy is just redirecting to other proxies, all invocations can share it. + StringBuilder combinedInfo = new StringBuilder("["); + for (int i = 0; i < failoverProxy.getOMProxies().size(); i++) { + if (i > 0) { + combinedInfo.append(','); + } + combinedInfo.append(failoverProxy.getOMProxies().get(i).proxyInfo); + } + combinedInfo.append(']'); + T wrappedProxy = (T) Proxy.newProxyInstance( + FollowerReadInvocationHandler.class.getClassLoader(), + new Class<?>[] {protocol}, new FollowerReadInvocationHandler()); + combinedProxy = new ProxyInfo<>(wrappedProxy, combinedInfo.toString()); + + if (wrappedProxy instanceof OzoneManagerProtocolPB) { + this.followerReadEnabled = true; + } else { + LOG.debug("Disabling follower reads for {} because the requested proxy " + + "class does not implement {}", omServiceId, OzoneManagerProtocolPB.class.getName()); + this.followerReadEnabled = false; + } + } + + @Override + public Class<T> getInterface() { + return protocolClass; + } + + @Override + public ProxyInfo<T> getProxy() { + return combinedProxy; + } + + @Override + public void performFailover(T currProxy) { + // Since FollowerReadInvocationHandler might user or fallback to leader-based failover logic, + // we should delegate the failover logic to the leader's failover. + failoverProxy.performFailover(currProxy); + } + + public RetryPolicy getRetryPolicy(int maxFailovers) { + // We use the OMFailoverProxyProviderBase's RetryPolicy instead of using our own retry policy + // 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 + // OMFailoverProxyProviderbase's RetryPolicy + return failoverProxy.getRetryPolicy(maxFailovers); + } + + /** + * Parse the OM request from the request args. + * + * @return parsed OM request. + */ + private static OMRequest parseOMRequest(Object[] args) throws Throwable { Review Comment: throws `ServiceException` instead of `Throwable`. ########## hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java: ########## @@ -0,0 +1,414 @@ +/* + * 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.ozone.om.ha; + +import static org.apache.hadoop.ozone.om.ha.OMFailoverProxyProviderBase.getLeaderNotReadyException; +import static org.apache.hadoop.ozone.om.ha.OMFailoverProxyProviderBase.getNotLeaderException; + +import com.google.common.annotations.VisibleForTesting; +import com.google.protobuf.Message; +import com.google.protobuf.RpcController; +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.hdds.conf.ConfigurationSource; +import org.apache.hadoop.io.retry.FailoverProxyProvider; +import org.apache.hadoop.io.retry.RetryPolicy; +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.ozone.OmUtils; +import org.apache.hadoop.ozone.om.exceptions.OMLeaderNotReadyException; +import org.apache.hadoop.ozone.om.exceptions.OMNotLeaderException; +import org.apache.hadoop.ozone.om.protocolPB.OzoneManagerProtocolPB; +import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest; +import org.apache.hadoop.security.UserGroupInformation; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A {@link org.apache.hadoop.io.retry.FailoverProxyProvider} implementation + * that supports reading from follower OM(s) (i.e. non-leader OMs also includes + * OM listeners). + * <p> + * This constructs a wrapper proxy might send the read request to follower + * OM(s), if follower read is enabled. It will try to send read requests + * to the first OM node. If RPC failed, it will try to failover to the next OM node. + * It will fail back to the leader OM after it has exhausted all the OMs. + * TODO: Currently the logic does not prioritize forwarding to followers since + * it requires an extra RPC latency to check the OM role info. + * In the future, we can try to try to pick the followers before forwarding + * the request to the leader (similar to ObserverReadProxyProvider). + * <p> + * Read and write requests will still be sent to leader OM if reading from + * follower is disabled. + */ +public class HadoopRpcOMFollowerReadFailoverProxyProvider<T> implements FailoverProxyProvider<T> { + @VisibleForTesting + public static final Logger LOG = LoggerFactory.getLogger(HadoopRpcOMFollowerReadFailoverProxyProvider.class); + + private final Class<T> protocolClass; + + /** The inner proxy provider used for leader-based failover. */ + private final HadoopRpcOMFailoverProxyProvider<T> failoverProxy; + + /** The combined proxy which redirects to other proxies as necessary. */ + private final ProxyInfo<T> combinedProxy; + + /** + * Whether reading from follower is enabled. If this is false, all read + * requests will still go to OM leader. + */ + private volatile boolean followerReadEnabled; + + /** + * The current index of the underlying leader-based proxy provider's omNodesInOrder currently being used. + * Should only be accessed in synchronized methods. + */ + private int currentIndex = -1; + + /** + * The proxy currently being used to send the read request. + * Should only be accessed in synchronized methods. + */ + private OMProxyInfo<T> currentProxy; + + /** The last proxy that has been used. Only used for testing. */ + private volatile OMProxyInfo<T> lastProxy = null; + + public HadoopRpcOMFollowerReadFailoverProxyProvider( + ConfigurationSource configuration, UserGroupInformation ugi, String omServiceId, Class<T> protocol) + throws IOException { + this(omServiceId, protocol, + new HadoopRpcOMFailoverProxyProvider<>(configuration, ugi, omServiceId, protocol)); + } + + @SuppressWarnings("unchecked") + public HadoopRpcOMFollowerReadFailoverProxyProvider(String omServiceId, Class<T> protocol, + HadoopRpcOMFailoverProxyProvider<T> failoverProxy) throws IOException { + this.protocolClass = protocol; + this.failoverProxy = failoverProxy; + + // Create a wrapped proxy containing all the proxies. Since this combined + // proxy is just redirecting to other proxies, all invocations can share it. + StringBuilder combinedInfo = new StringBuilder("["); + for (int i = 0; i < failoverProxy.getOMProxies().size(); i++) { + if (i > 0) { + combinedInfo.append(','); + } + combinedInfo.append(failoverProxy.getOMProxies().get(i).proxyInfo); + } + combinedInfo.append(']'); + T wrappedProxy = (T) Proxy.newProxyInstance( + FollowerReadInvocationHandler.class.getClassLoader(), + new Class<?>[] {protocol}, new FollowerReadInvocationHandler()); + combinedProxy = new ProxyInfo<>(wrappedProxy, combinedInfo.toString()); + + if (wrappedProxy instanceof OzoneManagerProtocolPB) { + this.followerReadEnabled = true; + } else { + LOG.debug("Disabling follower reads for {} because the requested proxy " + + "class does not implement {}", omServiceId, OzoneManagerProtocolPB.class.getName()); + this.followerReadEnabled = false; + } + } + + @Override + public Class<T> getInterface() { + return protocolClass; + } + + @Override + public ProxyInfo<T> getProxy() { + return combinedProxy; + } + + @Override + public void performFailover(T currProxy) { + // Since FollowerReadInvocationHandler might user or fallback to leader-based failover logic, + // we should delegate the failover logic to the leader's failover. + failoverProxy.performFailover(currProxy); + } + + public RetryPolicy getRetryPolicy(int maxFailovers) { + // We use the OMFailoverProxyProviderBase's RetryPolicy instead of using our own retry policy + // 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 + // OMFailoverProxyProviderbase's RetryPolicy + return failoverProxy.getRetryPolicy(maxFailovers); + } + + /** + * Parse the OM request from the request args. + * + * @return parsed OM request. + */ + private static OMRequest parseOMRequest(Object[] args) throws Throwable { + if (args == null || args.length < 2 || !(args[1] instanceof Message)) { + LOG.error("Request failed since OM request is null and cannot be parsed"); + // Throws a non-retriable exception to prevent retry and failover + // See the HddsUtils#shouldNotFailoverOnRpcException used in + // OMFailoverProxyProviderBase#shouldFailover + throw wrapInServiceException( + new RpcNoSuchProtocolException("OM request is null and cannot be parsed")); + } + final Message theRequest = (Message) args[1]; + return (OMRequest) theRequest; + } + + @VisibleForTesting + void setFollowerReadEnabled(boolean flag) { + this.followerReadEnabled = flag; + } + + @VisibleForTesting + public ProxyInfo<T> getLastProxy() { + return lastProxy; + } + + /** + * Return the currently used proxy. If there is none, first calls + * {@link #changeProxy(OMProxyInfo)} to initialize one. + */ + @VisibleForTesting + public OMProxyInfo<T> getCurrentProxy() { + return changeProxy(null); + } + + /** + * Move to the next proxy in the proxy list. If the OMProxyInfo supplied by + * the caller does not match the current proxy, the call is ignored; this is + * to handle concurrent calls (to avoid changing the proxy multiple times). + * The service state of the newly selected proxy will be updated before + * returning. + * + * @param initial The expected current proxy + * @return The new proxy that should be used. + */ + private synchronized OMProxyInfo<T> changeProxy(OMProxyInfo<T> initial) { + if (currentProxy != initial) { + // Must have been a concurrent modification; ignore the move request + return currentProxy; + } + currentIndex = (currentIndex + 1) % failoverProxy.getOmNodesInOrder().size(); + String currentOmNodeId = failoverProxy.getOmNodesInOrder().get(currentIndex); + currentProxy = (OMProxyInfo<T>) failoverProxy.createOMProxyIfNeeded(currentOmNodeId); + LOG.debug("Changed current proxy from {} to {}", + initial == null ? "none" : initial.proxyInfo, + currentProxy.proxyInfo); + return currentProxy; + } + + /** + * An InvocationHandler to handle incoming requests. This class's invoke + * method contains the primary logic for redirecting 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 + * proxy's OM node fails, adjust the current proxy and return on the next one. + * <p> + * Write requests are always forwarded to the leader. + */ + private class FollowerReadInvocationHandler implements RpcInvocationHandler { + + @Override + public Object invoke(Object proxy, final Method method, final Object[] args) + throws Throwable { + 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); + } + Object retVal; + OMRequest omRequest = parseOMRequest(args); + if (followerReadEnabled && OmUtils.shouldSendToFollower(omRequest)) { + int failedCount = 0; + for (int i = 0; i < failoverProxy.getOmNodesInOrder().size(); i++) { + OMProxyInfo<T> current = getCurrentProxy(); + LOG.debug("Attempting to service {} with cmdType {} using proxy {}", + method.getName(), omRequest.getCmdType(), current.proxyInfo); + try { + retVal = method.invoke(current.proxy, args); + 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)) { + throw wrapInServiceException(ite.getCause()); + } + Exception e = (Exception) ite.getCause(); + if (e instanceof InterruptedIOException || + e instanceof InterruptedException) { + // If interrupted, do not retry. + LOG.warn("Invocation returned interrupted exception on [{}];", + current.proxyInfo, e); + throw wrapInServiceException(e); + } + + if (e instanceof ServiceException) { + OMNotLeaderException notLeaderException = + getNotLeaderException(e); + if (notLeaderException != null) { + // We should disable follower read here since this means + // the OM follower does not support / disable follower read or something is misconfigured + LOG.debug("Encountered OMNotLeaderException from {}. " + + "Disable OM follower read and retry OM leader directly.", current.proxyInfo); + followerReadEnabled = false; + // Break here instead of throwing exception so that it is not counted + // as a failover + break; + } + + OMLeaderNotReadyException leaderNotReadyException = + getLeaderNotReadyException(e); + if (leaderNotReadyException != null) { + LOG.debug("Encountered OMLeaderNotReadyException from {}. " + + "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; + } + } + + if (!failoverProxy.shouldFailover(e)) { + // We reuse the leader proxy provider failover since we want to ensure + // if the follower read proxy decides that the exception should be failed, + // the leader proxy provider failover retry policy (i.e. OMFailoverProxyProviderBase#getRetryPolicy) + // should also fail the call. + // Otherwise, if the follower read proxy decides the exception should be failed, but + // the leader decides to failover to the its next proxy, the follower read proxy remains + // unchanged and the next read calls might query the same failing OM node and + // fail indefinitely. + LOG.debug("Invocation with cmdType {} returned exception on [{}] that cannot be retried; " + + "{} failure(s) so far", + omRequest.getCmdType(), current.proxyInfo, failedCount, e); + throw e; + } else { + failedCount++; + LOG.warn( + "Invocation with cmdType {} returned exception on [{}]; {} failure(s) so far", + omRequest.getCmdType(), current.proxyInfo, failedCount, e); + changeProxy(current); + } + } + } + + // Only log message if there are actual follower failures. + // Getting here with failedCount = 0 could + // 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 {}." + + " Falling back to leader.", failedCount, + omRequest.getCmdType(), method.getName()); + } else { + if (LOG.isDebugEnabled()) { + LOG.debug("Read falling back to leader without follower read " + + "fail, is there no follower node running?"); + } + } + } + + // 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()); + OMProxyInfo<T> leaderProxy = (OMProxyInfo<T>) failoverProxy.getProxy(); + try { + retVal = method.invoke(leaderProxy.proxy, args); + } catch (InvocationTargetException e) { + LOG.debug("Exception thrown from leader-based failoverProxy", e.getCause()); + // This exception will be handled by the OMFailoverProxyProviderBase#getRetryPolicy + // (see getRetryPolicy). This ensures that the leader-only failover should still work. + throw wrapInServiceException(e.getCause()); + } + lastProxy = leaderProxy; + return retVal; + } + + @Override + public void close() throws IOException { + + } + + @Override + public ConnectionId getConnectionId() { + return RPC.getConnectionIdForProxy(followerReadEnabled + ? getCurrentProxy().proxy : failoverProxy.getProxy().proxy); + } + } + + @Override + public synchronized void close() throws IOException { + // All the proxies are stored in the underlying failoverProxy + // so we invoke close on the underlying failoverProxy + failoverProxy.close(); + } + + @VisibleForTesting + public boolean isFollowerReadEnabled() { + return followerReadEnabled; + } + + @VisibleForTesting + public List<OMProxyInfo<T>> getOMProxies() { + return failoverProxy.getOMProxies(); + } + + public synchronized void changeInitialProxyForTest(String initialOmNodeId) { + if (currentProxy != null && currentProxy.getNodeId().equals(initialOmNodeId)) { + return; + } + + int indexOfTargetNodeId = failoverProxy.getOmNodesInOrder().indexOf(initialOmNodeId); + if (indexOfTargetNodeId == -1) { + return; + } + + currentIndex = indexOfTargetNodeId; + currentProxy = (OMProxyInfo<T>) failoverProxy.createOMProxyIfNeeded(initialOmNodeId); + } + + /** + * Wrap the throwable in {@link ServiceException} if necessary. + * 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}. + * @return if the throwable is already an instance {@link ServiceException} simply returns the exception itself. + * Otherwise, return the exception wrapped in {@link ServiceException} + */ + private static Throwable wrapInServiceException(Throwable e) { + if (e instanceof ServiceException) { + return e; + } + return new ServiceException(e); + } Review Comment: throws `ServiceException`. ```java private static void throwServiceException(Throwable e) throws ServiceException { throw e instanceof ServiceException ? (ServiceException) e : new ServiceException(e); } ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
