Do not deserialize back BytesTransportRequest to clone a request in MockTransportService (elastic#89926) (elastic#89936)

Some handlers, such as JOIN_VALIDATE_ACTION_NAME, deserialize a
BytesTransportRequest into a different class that cannot be serialized.
This commit changes that behaviour and directly clones the BytesTransportRequest
instead of relying on the transport handler.

Closes elastic#88120

Author: fcofdez

test/framework/src/main/java/org/elasticsearch/test/transport/MockTransportService.java

@@ -37,6 +37,7 @@
 import org.elasticsearch.test.tasks.MockTaskManager;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.tracing.Tracer;
+import org.elasticsearch.transport.BytesTransportRequest;
 import org.elasticsearch.transport.ClusterConnectionManager;
 import org.elasticsearch.transport.ConnectTransportException;
 import org.elasticsearch.transport.ConnectionProfile;
@@ -502,10 +503,21 @@ public void sendRequest(
                 }
 
                 // poor mans request cloning...
-                RequestHandlerRegistry<?> reg = MockTransportService.this.getRequestHandler(action);
                 BytesStreamOutput bStream = new BytesStreamOutput();
                 request.writeTo(bStream);
-                final TransportRequest clonedRequest = reg.newRequest(bStream.bytes().streamInput());
+                final TransportRequest clonedRequest;
+                if (request instanceof BytesTransportRequest) {
+                    // Some request handlers read back a BytesTransportRequest
+                    // into a different class that cannot be re-serialized (i.e. JOIN_VALIDATE_ACTION_NAME),
+                    // in those cases we just copy the raw bytes back to a BytesTransportRequest.
+                    // This is only needed for the BwC for JOIN_VALIDATE_ACTION_NAME and can be removed in the next major
+                    assert Version.CURRENT.major == Version.V_7_17_0.major + 1;
+                    clonedRequest = new BytesTransportRequest(bStream.bytes().streamInput());
+                } else {
+                    RequestHandlerRegistry<?> reg = MockTransportService.this.getRequestHandler(action);
+                    clonedRequest = reg.newRequest(bStream.bytes().streamInput());
+                }
+                assert clonedRequest.getClass().equals(request.getClass()) : clonedRequest + " vs " + request;
 
                 final RunOnce runnable = new RunOnce(new AbstractRunnable() {
                     @Override