HDDS-14379. Implement basic Hadoop OM client proxy provider to read from followers #9641

ivandika3 · 2026-01-17T05:32:11Z

What changes were proposed in this pull request?

This task is to come up with the basic implementation of follower read client proxy as a baseline before further performance improvements. The idea is to simply pick an OM node in random (which can be a leader or follower) and use it to submit read requests. The read requests need to keep sending to that OM node unless the OM is down which triggers failover. Write requests should be sent to the OM leader directly.

Further improvements such as followers affinity or picking OM based on latency, applied index, etc will be implemented in follow up tasks. The main focus of this patch is to ensure that long-lived client (e.g. S3G Ozone clients) will stick to the OM follower once it picks it as the current proxy. In the previous leader proxy provider implementation, the client only read from followers until a new write request triggers OMNotLeaderException and the failover causes proxy to always be pointing to the leader.

The implementation is to introduce HadoopRpcOMFollowerReadProxyProvider which wraps
HadoopRpcOMFailoverProxyProvider. FollowerReadProxyProvider tracks a different currentOmNodeId from HadoopRpcOMFailoverProxyProvider. FollowerReadInvocationHandler will check whether the request is a read request (using OmUtils#isReadOnly) and if so forwards it to its current proxy. If it's a write request, the request if forwarded to HadoopRpcOMFailoverProxyProvider to be sent to the leader.

So the proxy hierarchy (each with its own InvocationHandler) is

TracingUtil's proxy (InvocationHandler: TraceAllMethod)
- RetryProxy (InvocationHandler: RetryInvocationHandler)
  - HadoopRpcOMFollowerReadProxyProvider (InvocationHandler: FollowerReadInvocationHandler)
    - ProtocolProxy (which is created in OMFailoverProxyProviderBase#createOMProxy): ProtobufRpcEngine.Invoker

What is the link to the Apache JIRA

https://issues.apache.org/jira/browse/HDDS-14379

How was this patch tested?

UT and IT.

Clean CI: https://github.com/ivandika3/ozone/actions/runs/21102966224

…rom followers

ivandika3 · 2026-01-18T05:25:15Z

cc: @greenwich

szetszwo · 2026-01-19T18:56:08Z

@ivandika3 , thanks for working on this! I am reviewing this. The change is quite big. Need some time.

szetszwo

@ivandika3 , thanks a lot for working on this and adding a lot of tests! Please see the comments inlined.

BTW, filed HDDS-14455 and HDDS-14470 for improving the current code.

szetszwo · 2026-01-20T18:46:43Z

...one/common/src/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFailoverProxyProvider.java

+  protected synchronized ProxyInfo<T> createOMProxyIfNeeded(OMProxyInfo<T> omProxyInfo) {
    if (omProxyInfo.proxy == null) {


The synchronized is for omProxyInfo.proxy. We probably should do it in OMProxyInfo. Let me fix it in HDDS-14470.

Ok will wait until HDDS-14470 is merged.

Let merge this first if it is ready. I can update the change for HDDS-14470.

szetszwo · 2026-01-21T19:03:55Z

...rc/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java

+        new HadoopRpcOMFailoverProxyProvider<>(configuration, ugi, omServiceId, protocol));
+  }
+
+  @SuppressWarnings("unchecked")


Move it to wrappedProxy. Then it would only suppress the warning there but not the entire method.

+++ 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());

Thanks, updated.

szetszwo · 2026-01-21T19:15:32Z

...rc/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java

+    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(']');


We may use "map/reduce".

final String combinedInfo = "[" + failoverProxy.getOMProxies().stream() .map(a -> a.proxyInfo) .reduce((a, b) -> a + ", " + b).orElse("") + "]";

Thanks, updated.

szetszwo · 2026-01-21T19:18:26Z

...rc/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java

+  }
+
+  @SuppressWarnings("unchecked")
+  public HadoopRpcOMFollowerReadFailoverProxyProvider(String omServiceId, Class<T> protocol,


For simplicity, let support only OzoneManagerProtocolPB for now? Then, we can remove the followerReadEnabled and simplify the code.

Thanks for the review. Could you clarify on this?

Currently there are guards to ensure that follower read only supports OzoneManagerProtocolPB. One is during construction by disabling useFollowerRead if it's not OzoneManagerProtocolPB. Second is during FollowerReadInvocationHandler.

Do you mean to make HadoopRpcOMFollowerReadFailoverProxyProvider to be HadoopRpcOMFollowerReadFailoverProxyProvider<OzoneManagerProtocolPB> implements FailoverProxy<OzoneManagerProtocolPB>? I believe this might mean that we need to replace T generic type parameter with OzoneManagerProtocolPB. This adds verbosity and reduces the generic benefit, not sure if this is a good tradeoff for removing userFollowerRead.

Please let me know what you think.

Is it the case that it uses follower read only if protocol == OzoneManagerProtocolPB.class? If yes, we don't pass the protocol parameter.

szetszwo · 2026-01-21T19:27:21Z

...rc/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java

+  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;
+  }


Let's provide a more specific error message:

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]; }

Thanks, updated.

I updated this to handle null args[1]. I also changed this to use normal if else blocks since I personally find chained ternary operators hard to read.

szetszwo · 2026-01-21T19:30:04Z

...rc/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java

+    public Object invoke(Object proxy, final Method method, final Object[] args)
+        throws Throwable {
+      lastProxy = null;
+      if (method.getDeclaringClass() == Object.class) {


Check (method.getDeclaringClass() != OzoneManagerProtocolPB.class) instead?

Yes, better. Updated, thanks.

I reverted this due to failing test.

The actual Method#getDeclaringClass is org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos$OzoneManagerService$BlockingInterface so all submitRequest will go to this block.

Additionally, I tried to change the proxy to be pass through if it's not OzoneManagerProtocolPB.class

From

return method.invoke(this, args);

to

return method.invoke(proxy, args);

However, it seems because org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos$OzoneManagerService$BlockingInterface is not concrete class, it throws exception when usineg Object#toString.

Caused by: java.lang.NoSuchMethodException: org.apache.hadoop.ozone.om.protocolPB.OzoneManagerProtocolPB.toString()

Not sure if there is a way to handle all cases.

Thanks for trying it! Let's keep using Object.class.

szetszwo · 2026-01-21T19:38:51Z

...rc/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java

+  private static Throwable wrapInServiceException(Throwable e) {
+    if (e instanceof ServiceException) {
+      return e;
+    }
+    return new ServiceException(e);
+  }


throws ServiceException.

private static void throwServiceException(Throwable e) throws ServiceException { throw e instanceof ServiceException ? (ServiceException) e : new ServiceException(e); }

Thanks, updated. However, retVal needs to be initialized to null since the compiler cannot detect that throwServiceException always throws exception (since it might think that it simply returns). So I have initialized retVal to null.

Then, we may return ServiceException. Anyway, both are about the same.

private static ServiceException getServiceException(Throwable e) { return e instanceof ServiceException ? (ServiceException) e : new ServiceException(e); }

szetszwo · 2026-01-21T19:53:13Z

...rc/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java

+   *
+   * @return parsed OM request.
+   */
+  private static OMRequest parseOMRequest(Object[] args) throws Throwable {


throws ServiceException instead of Throwable.

Thanks, updated.

szetszwo · 2026-01-21T20:09:16Z

...rc/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java

+   * Whether reading from follower is enabled. If this is false, all read
+   * requests will still go to OM leader.
+   */
+  private volatile boolean followerReadEnabled;


Let's rename it to useFollowerRead to avoid confusion with the conf.

Thanks, updated.

szetszwo · 2026-01-21T20:10:51Z

...rc/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java

+      OMRequest omRequest = parseOMRequest(args);
+      if (followerReadEnabled && OmUtils.shouldSendToFollower(omRequest)) {
+        int failedCount = 0;
+        for (int i = 0; i < failoverProxy.getOmNodesInOrder().size(); i++) {


Check followerReadEnabled: followerReadEnabled && i < failoverProxy.getOmNodesInOrder().size()

Thanks, updated.

ivandika3

@szetszwo Thanks for the review.

ivandika3 · 2026-01-22T04:02:56Z

...one/common/src/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFailoverProxyProvider.java

+  protected synchronized ProxyInfo<T> createOMProxyIfNeeded(OMProxyInfo<T> omProxyInfo) {
    if (omProxyInfo.proxy == null) {


Ok will wait until HDDS-14470 is merged.

ivandika3 · 2026-01-22T04:03:49Z

...rc/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java

+   * Whether reading from follower is enabled. If this is false, all read
+   * requests will still go to OM leader.
+   */
+  private volatile boolean followerReadEnabled;


Thanks, updated.

ivandika3 · 2026-01-22T06:03:35Z

...rc/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java

+        new HadoopRpcOMFailoverProxyProvider<>(configuration, ugi, omServiceId, protocol));
+  }
+
+  @SuppressWarnings("unchecked")


Thanks, updated.

ivandika3 · 2026-01-22T06:04:54Z

...rc/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java

+    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(']');


Thanks, updated.

ivandika3 · 2026-01-22T06:06:09Z

...rc/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java

+   *
+   * @return parsed OM request.
+   */
+  private static OMRequest parseOMRequest(Object[] args) throws Throwable {


Thanks, updated.

ivandika3 · 2026-01-22T06:07:36Z

...rc/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java

+  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;
+  }


Thanks, updated.

ivandika3 · 2026-01-22T06:08:23Z

...rc/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java

+    public Object invoke(Object proxy, final Method method, final Object[] args)
+        throws Throwable {
+      lastProxy = null;
+      if (method.getDeclaringClass() == Object.class) {


Yes, better. Updated, thanks.

ivandika3 · 2026-01-22T06:09:59Z

...rc/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java

+      OMRequest omRequest = parseOMRequest(args);
+      if (followerReadEnabled && OmUtils.shouldSendToFollower(omRequest)) {
+        int failedCount = 0;
+        for (int i = 0; i < failoverProxy.getOmNodesInOrder().size(); i++) {


Thanks, updated.

ivandika3 · 2026-01-22T06:16:57Z

...rc/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java

+  private static Throwable wrapInServiceException(Throwable e) {
+    if (e instanceof ServiceException) {
+      return e;
+    }
+    return new ServiceException(e);
+  }


Thanks, updated. However, retVal needs to be initialized to null since the compiler cannot detect that throwServiceException always throws exception (since it might think that it simply returns). So I have initialized retVal to null.

ivandika3 · 2026-01-22T06:34:09Z

...rc/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java

+  }
+
+  @SuppressWarnings("unchecked")
+  public HadoopRpcOMFollowerReadFailoverProxyProvider(String omServiceId, Class<T> protocol,


Thanks for the review. Could you clarify on this?

Currently there are guards to ensure that follower read only supports OzoneManagerProtocolPB. One is during construction by disabling useFollowerRead if it's not OzoneManagerProtocolPB. Second is during FollowerReadInvocationHandler.

Do you mean to make HadoopRpcOMFollowerReadFailoverProxyProvider to be HadoopRpcOMFollowerReadFailoverProxyProvider<OzoneManagerProtocolPB> implements FailoverProxy<OzoneManagerProtocolPB>? I believe this might mean that we need to replace T generic type parameter with OzoneManagerProtocolPB. This adds verbosity and reduces the generic benefit, not sure if this is a good tradeoff for removing userFollowerRead.

Please let me know what you think.

# Conflicts: # hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/Hadoop3OmTransport.java # hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/OmTestUtil.java # hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHAWithAllRunning.java # hadoop-ozone/ozonefs-hadoop2/src/main/java/org/apache/hadoop/fs/ozone/Hadoop27RpcTransport.java

This reverts commit d804991.

szetszwo

+1 the change looks good.

szetszwo · 2026-01-22T18:45:36Z

...one/common/src/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFailoverProxyProvider.java

+  protected synchronized ProxyInfo<T> createOMProxyIfNeeded(OMProxyInfo<T> omProxyInfo) {
    if (omProxyInfo.proxy == null) {


Let merge this first if it is ready. I can update the change for HDDS-14470.

szetszwo · 2026-01-22T18:47:51Z

...rc/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java

+  }
+
+  @SuppressWarnings("unchecked")
+  public HadoopRpcOMFollowerReadFailoverProxyProvider(String omServiceId, Class<T> protocol,


Is it the case that it uses follower read only if protocol == OzoneManagerProtocolPB.class? If yes, we don't pass the protocol parameter.

szetszwo · 2026-01-22T18:50:05Z

...rc/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java

+    public Object invoke(Object proxy, final Method method, final Object[] args)
+        throws Throwable {
+      lastProxy = null;
+      if (method.getDeclaringClass() == Object.class) {


Thanks for trying it! Let's keep using Object.class.

ivandika3 added 3 commits January 17, 2026 13:28

HDDS-14379. Implement basic Hadoop OM client proxy provider to read f…

e05dfad

…rom followers

Fix leak in test

2fe5704

Try to use assertThat in slow node test

435ef92

ivandika3 marked this pull request as ready for review January 18, 2026 05:09

ivandika3 requested review from kerneltime, peterxcli, szetszwo and whbing January 18, 2026 05:09

ivandika3 self-assigned this Jan 18, 2026

chungen0126 self-requested a review January 20, 2026 05:40

szetszwo reviewed Jan 21, 2026

View reviewed changes

ivandika3 added 11 commits January 22, 2026 12:05

Change from isFollowerReadEnabled to useFollowerRead

f2db9cf

Update SuppressWarnings

0d663a4

Use stream API for combinedInfo

406e34d

Make parseOMRequest throw ServiceException

fdd6d6e

More detailed error message in parseOMRequest

c7e9a44

Better declaring class update

d804991

Check useFollowerRead while iterating in follower

3abd0f9

Change wrapInServiceException to throwServiceException

6a3f9b7

Remove redundant toString

fb6b942

Remove redundant throws clause

77939e5

Remove unused import

1a1a591

ivandika3 commented Jan 22, 2026

View reviewed changes

ivandika3 added 5 commits January 22, 2026 14:44

Adhere to HDDS-14455

d6ba3be

Revert "Better declaring class update"

0bffb04

This reverts commit d804991.

Handle null args[1] in parseOMRequest

02cbdf9

Fix checkstyle

8592a99

szetszwo approved these changes Jan 22, 2026

View reviewed changes

szetszwo merged commit a698f48 into apache:master Jan 22, 2026
44 checks passed

		protected synchronized ProxyInfo<T> createOMProxyIfNeeded(OMProxyInfo<T> omProxyInfo) {
		if (omProxyInfo.proxy == null) {

HDDS-14379. Implement basic Hadoop OM client proxy provider to read from followers #9641

HDDS-14379. Implement basic Hadoop OM client proxy provider to read from followers #9641

Conversation

ivandika3 commented Jan 17, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

What is the link to the Apache JIRA

How was this patch tested?

Uh oh!

ivandika3 commented Jan 18, 2026

Uh oh!

szetszwo commented Jan 19, 2026

Uh oh!

szetszwo left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ivandika3 Jan 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ivandika3 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

ivandika3 commented Jan 17, 2026 •

edited

Loading

ivandika3 Jan 22, 2026 •

edited

Loading