HADOOP-19864. Cut WritableRPCEngine (#8433)

WritableRPCEngine is no more; MR TaskUmbilicalProtocol now uses protobuf.

RpcWritable still exists, and ClientCache uses it, but those are vestigal uses.

This is not backwards compatible as Tez uses it for their heartbeat protocol
in all versions without TEZ-4708.

Contains content generated by GitHub Copilot

Contributed by Steve Loughran

Author: steveloughran

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RPC.java

@@ -57,6 +57,7 @@
 import org.apache.hadoop.security.token.TokenIdentifier;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.util.Preconditions;
 import org.apache.hadoop.util.ReflectionUtils;
 import org.apache.hadoop.util.Time;
 
@@ -88,7 +89,7 @@ public class RPC {
   final static int RPC_SERVICE_CLASS_DEFAULT = 0;
   public enum RpcKind {
     RPC_BUILTIN ((short) 1),         // Used for built in calls by tests
-    RPC_WRITABLE ((short) 2),        // Use WritableRpcEngine 
+    RPC_WRITABLE((short) 2),         // WritableRpcEngine removed; kept for wire-level detection
     RPC_PROTOCOL_BUFFER ((short) 3); // Use ProtobufRpcEngine
     final static short MAX_SIZE = RPC_PROTOCOL_BUFFER.value; // used for array size
     private final short value;
@@ -221,9 +222,11 @@ static synchronized RpcEngine getProtocolEngine(Class<?> protocol,
       Configuration conf) {
     RpcEngine engine = PROTOCOL_ENGINES.get(protocol);
     if (engine == null) {
-      Class<?> impl = conf.getClass(ENGINE_PROP+"."+protocol.getName(),
-                                    WritableRpcEngine.class);
-      engine = (RpcEngine)ReflectionUtils.newInstance(impl, conf);
+      Class<?> impl = conf.getClass(ENGINE_PROP + "." + protocol.getName(),
+          null);
+      Preconditions.checkState(impl != null,
+          "No RPC engine configured for %s", protocol.getName());
+      engine = (RpcEngine) ReflectionUtils.newInstance(impl, conf);
       PROTOCOL_ENGINES.put(protocol, engine);
     }
     return engine;

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RpcWritable.java

@@ -33,7 +33,14 @@
 import org.apache.hadoop.thirdparty.protobuf.CodedOutputStream;
 import org.apache.hadoop.thirdparty.protobuf.Message;
 
-// note anything marked public is solely for access by SaslRpcClient
+/**
+ * Marshalling support, for hadoop shaded protobuf and legacy
+ * protobuf 2.5.
+ * It originally supported hadoop Writables for the WritableRPCEngine;
+ * that is no removed. All that is retained is the name.
+ * Anything marked public is solely for access by SaslRpcClient
+ */
+// note
 @InterfaceAudience.Private
 public abstract class RpcWritable implements Writable {
 
@@ -66,7 +73,7 @@ public final void write(DataOutput out) throws IOException {
   abstract void writeTo(ResponseBuffer out) throws IOException;
   abstract <T> T readFrom(ByteBuffer bb) throws IOException;
 
-  // adapter for Writables.
+  // adapter for Writables. Used for RPC_BUILTIN calls.
   static class WritableWrapper extends RpcWritable {
     private final Writable writable;
 

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java

@@ -2939,6 +2939,13 @@ private void checkRpcHeaders(RpcRequestHeaderProto header)
     private void processRpcRequest(RpcRequestHeaderProto header,
         RpcWritable.Buffer buffer) throws RpcServerException,
         InterruptedException {
+      if (header.getRpcKind() == RpcKindProto.RPC_WRITABLE) {
+        final String err = "WritableRpcEngine is no longer supported."
+            + " Upgrade your Hadoop client to use ProtobufRpcEngine.";
+        LOG.warn(err + " Client: " + getHostAddress());
+        throw new FatalRpcServerException(
+            RpcErrorCodeProto.FATAL_INVALID_RPC_HEADER, err);
+      }
       Class<? extends Writable> rpcRequestClass = 
           getRpcRequestWrapper(header.getRpcKind());
       if (rpcRequestClass == null) {
@@ -3642,6 +3649,8 @@ private void setupResponse(RpcCall call,
     call.setResponse(ByteBuffer.wrap(response));
   }
 
+
+
   private byte[] setupResponseForWritable(
       RpcResponseHeaderProto header, Writable rv) throws IOException {
     ResponseBuffer buf = responseBuffer.get().reset();
@@ -3660,7 +3669,6 @@ private byte[] setupResponseForWritable(
     }
   }
 
-
   // writing to a pre-allocated array is the most efficient way to construct
   // a protobuf response.
   private byte[] setupResponseForProtobuf(