Fix implementation based on comment

Author: Russole

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMRatisServerImpl.java

@@ -41,6 +41,7 @@
 import org.apache.hadoop.hdds.scm.AddSCMRequest;
 import org.apache.hadoop.hdds.scm.RemoveSCMRequest;
 import org.apache.hadoop.hdds.scm.ScmConfigKeys;
+import org.apache.hadoop.hdds.scm.ha.invoker.ScmInvoker;
 import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
 import org.apache.hadoop.hdds.security.SecurityConfig;
 import org.apache.hadoop.ozone.OzoneConsts;
@@ -223,6 +224,9 @@ public SCMStateMachine getSCMStateMachine() {
   @Override
   public void registerStateMachineHandler(final RequestType handlerType,
                                           final Object handler) {
+    if (handler instanceof ScmInvoker) {
+      stateMachine.registerInvoker(handlerType, (ScmInvoker) handler);
+    }
     stateMachine.registerHandler(handlerType, handler);
   }
 

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMStateMachine.java

@@ -120,7 +120,7 @@ public void registerHandler(RequestType type, Object handler) {
     handlers.put(type, handler);
   }
 
-  public void registerStateMachineInvoker(RequestType type,
+  public void registerInvoker(RequestType type,
       ScmInvoker<?> invoker) {
     invokers.put(type, invoker);
   }
@@ -189,7 +189,7 @@ public CompletableFuture<Message> applyTransaction(
   private Message process(final SCMRatisRequest request) throws Exception {
     final ScmInvoker<?> invoker = invokers.get(request.getType());
     if (invoker != null) {
-      return process(request, invoker);
+      return invoker.invokeLocal(request.getOperation(), request.getArguments());
     }
     return process(request, handlers.get(request.getType()));
   }
@@ -212,24 +212,6 @@ public static Message process(final SCMRatisRequest request, Object handler) thr
     }
   }
 
-  public static Message process(final SCMRatisRequest request,
-      final ScmInvoker<?> invoker) throws Exception {
-
-    if (invoker == null) {
-      throw new IOException("No invoker found for request type "
-          + request.getType());
-    }
-
-    final Object result = invoker.invoke(
-        request.getOperation(),
-        request.getArguments());
-
-    final Class<?> returnType =
-        invoker.getReturnType(request.getOperation(), request.getParameterTypes());
-
-    return SCMRatisResponse.encode(result, returnType);
-  }
-
   @Override
   public void notifyLogFailed(Throwable ex,
       RaftProtos.LogEntryProto failedEntry) {

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/invoker/DeletedBlockLogStateManagerInvoker.java

@@ -23,8 +23,10 @@
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.DeletedBlocksTransactionSummary;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.DeletedBlocksTransaction;
 import org.apache.hadoop.hdds.scm.block.DeletedBlockLogStateManager;
+import org.apache.hadoop.hdds.scm.ha.SCMRatisResponse;
 import org.apache.hadoop.hdds.scm.ha.SCMRatisServer;
 import org.apache.hadoop.hdds.utils.db.Table;
+import org.apache.ratis.protocol.Message;
 
 /** Code generated for {@link DeletedBlockLogStateManager}.  Do not modify. */
 public class DeletedBlockLogStateManagerInvoker extends ScmInvoker<DeletedBlockLogStateManager> {
@@ -107,7 +109,9 @@ public void removeTransactionsFromDB(ArrayList arg0, DeletedBlocksTransactionSum
 
   @SuppressWarnings("unchecked")
   @Override
-  public Object invokeLocal(String methodName, Object[] p) throws Exception {
+  public Message invokeLocal(String methodName, Object[] p) throws Exception {
+    final Class<?> returnType;
+    final Object returnValue;
     switch (methodName) {
     case "addTransactionsToDB":
       final ArrayList arg0 = p.length > 0 ? (ArrayList) p[0] : null;
@@ -116,7 +120,9 @@ public Object invokeLocal(String methodName, Object[] p) throws Exception {
       return null;
 
     case "getReadOnlyIterator":
-      return getImpl().getReadOnlyIterator();
+      returnType = Table.KeyValueIterator.class;
+      returnValue = getImpl().getReadOnlyIterator();
+      break;
 
     case "onFlush":
       getImpl().onFlush();
@@ -137,6 +143,8 @@ public Object invokeLocal(String methodName, Object[] p) throws Exception {
     default:
       throw new IllegalArgumentException("Method not found: " + methodName + " in DeletedBlockLogStateManager");
     }
+
+    return SCMRatisResponse.encode(returnValue, returnType);
   }
 
   @Override

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/invoker/ScmInvoker.java

@@ -26,6 +26,7 @@
 import org.apache.hadoop.hdds.scm.ha.SCMRatisRequest;
 import org.apache.hadoop.hdds.scm.ha.SCMRatisResponse;
 import org.apache.hadoop.hdds.scm.ha.SCMRatisServer;
+import org.apache.ratis.protocol.Message;
 
 /**
  * Invokes methods without using reflection.
@@ -62,7 +63,7 @@ public final Object invoke(String methodName, Object[] args) throws Exception {
   public abstract Class<?> getReturnType(String methodName, Class<?>[] parameterTypes);
 
   /** For non-@Replicate methods. */
-  abstract Object invokeLocal(String methodName, Object[] args) throws Exception;
+  public abstract Message invokeLocal(String methodName, Object[] args) throws Exception;
 
   /** For @Replicate DIRECT methods. */
   final Object invokeReplicateDirect(NameAndParameterTypes method, Object[] args) throws SCMException {