HDDS-16529. Replace reflective OM follower routing with typed dispatch

hadoop-hdds/common/src/main/java/org/apache/hadoop/io_/retry/RetryInvocationHandler.java

@@ -423,6 +423,9 @@ static boolean isRpcInvocation(Object proxy) {
     if (proxy instanceof ProtocolTranslator) {
       proxy = ((ProtocolTranslator) proxy).getUnderlyingProxyObject();
     }
+    if (proxy instanceof RpcProxy) {
+      return true;
+    }
     if (!Proxy.isProxyClass(proxy.getClass())) {
       return false;
     }

hadoop-hdds/common/src/main/java/org/apache/hadoop/ipc_/RPC.java

@@ -352,6 +352,9 @@ public static ConnectionId getConnectionIdForProxy(Object proxy) {
     if (proxy instanceof ProtocolTranslator) {
       proxy = ((ProtocolTranslator)proxy).getUnderlyingProxyObject();
     }
+    if (proxy instanceof RpcProxy) {
+      return ((RpcProxy) proxy).getConnectionId();
+    }
     RpcInvocationHandler inv = (RpcInvocationHandler) Proxy
         .getInvocationHandler(proxy);
     return inv.getConnectionId();

hadoop-hdds/common/src/main/java/org/apache/hadoop/ipc_/RpcInvocationHandler.java

@@ -17,20 +17,11 @@
  */
 package org.apache.hadoop.ipc_;
 
-import java.io.Closeable;
 import java.lang.reflect.InvocationHandler;
 
-import org.apache.hadoop.ipc_.Client.ConnectionId;
-
 /**
  * This interface must be implemented by all InvocationHandler
  * implementations.
  */
-public interface RpcInvocationHandler extends InvocationHandler, Closeable {
-
-  /**
-   * Returns the connection id associated with the InvocationHandler instance.
-   * @return ConnectionId
-   */
-  ConnectionId getConnectionId();
+public interface RpcInvocationHandler extends InvocationHandler, RpcProxy {
 }

hadoop-hdds/common/src/main/java/org/apache/hadoop/ipc_/RpcProxy.java

@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ipc_;
+
+import java.io.Closeable;
+import org.apache.hadoop.ipc_.Client.ConnectionId;
+
+/**
+ * Connection and lifecycle access for RPC proxies, including concrete protocol implementations.
+ */
+public interface RpcProxy extends Closeable {
+  /**
+   * @return the connection ID associated with this RPC proxy.
+   */
+  ConnectionId getConnectionId();
+}

hadoop-hdds/common/src/test/java/org/apache/hadoop/io_/retry/TestRetryProxy.java

@@ -25,6 +25,7 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 import static org.mockito.Mockito.any;
@@ -33,6 +34,7 @@
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
 import static org.mockito.Mockito.when;
 
 import java.io.IOException;
@@ -51,7 +53,10 @@
 import org.apache.hadoop.io.retry.RetryPolicy.RetryAction;
 import org.apache.hadoop.io.retry.RetryPolicy.RetryAction.RetryDecision;
 import org.apache.hadoop.io_.retry.UnreliableInterface.UnreliableException;
+import org.apache.hadoop.ipc_.Client.ConnectionId;
 import org.apache.hadoop.ipc_.ProtocolTranslator;
+import org.apache.hadoop.ipc_.RPC;
+import org.apache.hadoop.ipc_.RpcProxy;
 import org.apache.hadoop.security.AccessControlException;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -142,6 +147,24 @@ public Object getUnderlyingProxyObject() {
     // For non-proxy the method must return false
     assertFalse(RetryInvocationHandler.isRpcInvocation(new Object()));
   }
+
+  @Test
+  public void testConcreteRpcProxy() throws Exception {
+    RpcProxy proxy = mock(RpcProxy.class);
+    assertTrue(RetryInvocationHandler.isRpcInvocation(proxy));
+    verifyNoInteractions(proxy);
+
+    ConnectionId connectionId = mock(ConnectionId.class);
+    when(proxy.getConnectionId()).thenReturn(connectionId);
+    assertSame(connectionId, RPC.getConnectionIdForProxy(proxy));
+
+    ProtocolTranslator translator = () -> proxy;
+    assertTrue(RetryInvocationHandler.isRpcInvocation(translator));
+    assertSame(connectionId, RPC.getConnectionIdForProxy(translator));
+
+    RPC.stopProxy(proxy);
+    verify(proxy).close();
+  }
 
   @Test
   public void testRetryForever() throws UnreliableException {