No need to fix up ClusterState, it should be done by the transport layer

Author: DaveCTurner

server/src/main/java/org/elasticsearch/cluster/coordination/Coordinator.java

@@ -122,9 +122,8 @@ private void handleApplyCommit(ApplyCommitRequest applyCommitRequest) {
     }
 
     PublishWithJoinResponse handlePublishRequest(PublishRequest publishRequest) {
-        // overwrite local node
-        publishRequest = new PublishRequest(ClusterState.builder(publishRequest.getAcceptedState()).nodes(
-            DiscoveryNodes.builder(publishRequest.getAcceptedState().nodes()).localNodeId(getLocalNode().getId()).build()).build());
+        assert publishRequest.getAcceptedState().nodes().getLocalNode().equals(getLocalNode()) :
+            publishRequest.getAcceptedState().nodes().getLocalNode() + " != " + getLocalNode();
 
         synchronized (mutex) {
             final DiscoveryNode sourceNode = publishRequest.getAcceptedState().nodes().getMasterNode();

test/framework/src/main/java/org/elasticsearch/test/disruption/DisruptableMockTransport.java

@@ -20,7 +20,9 @@
 
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedMessage;
+import org.elasticsearch.cluster.ClusterModule;
 import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.test.transport.MockTransport;
 import org.elasticsearch.transport.ConnectTransportException;
 import org.elasticsearch.transport.RequestHandlerRegistry;
@@ -29,8 +31,11 @@
 import org.elasticsearch.transport.TransportResponse;
 import org.elasticsearch.transport.TransportResponseOptions;
 
+import java.io.IOException;
 import java.util.Optional;
 
+import static org.elasticsearch.test.ESTestCase.copyWriteable;
+
 public abstract class DisruptableMockTransport extends MockTransport {
     private final Logger logger;
 
@@ -68,7 +73,6 @@ public String toString() {
     protected void onSendRequest(long requestId, String action, TransportRequest request, DiscoveryNode destination) {
 
         assert destination.equals(getLocalNode()) == false : "non-local message from " + getLocalNode() + " to itself";
-        super.onSendRequest(requestId, action, request, destination);
 
         final String requestDescription = new ParameterizedMessage("[{}][{}] from {} to {}",
             action, requestId, getLocalNode(), destination).getFormattedMessage();
@@ -162,8 +166,15 @@ public String toString() {
                             }
                         };
 
+                        final TransportRequest copiedRequest;
+                        try {
+                            copiedRequest = copyWriteable(request, writeableRegistry(), requestHandler::newRequest);
+                        } catch (IOException e) {
+                            throw new AssertionError("exception de/serializing request", e);
+                        }
+
                         try {
-                            requestHandler.processMessageReceived(request, transportChannel);
+                            requestHandler.processMessageReceived(copiedRequest, transportChannel);
                         } catch (Exception e) {
                             try {
                                 transportChannel.sendResponse(e);
@@ -181,6 +192,10 @@ public String toString() {
         });
     }
 
+    private NamedWriteableRegistry writeableRegistry() {
+        return new NamedWriteableRegistry(ClusterModule.getNamedWriteables());
+    }
+
     public enum ConnectionStatus {
         CONNECTED,
         DISCONNECTED, // network requests to or from this node throw a ConnectTransportException

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

@@ -111,7 +111,6 @@ public void clear() {
 
     @Override
     protected void onSendRequest(long requestId, String action, TransportRequest request, DiscoveryNode node) {
-        super.onSendRequest(requestId, action, request, node);
         capturedRequests.add(new CapturingTransport.CapturedRequest(node, requestId, action, request));
     }
 

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

@@ -181,7 +181,6 @@ public void close() {
     }
 
     protected void onSendRequest(long requestId, String action, TransportRequest request, DiscoveryNode node) {
-
     }
 
     protected boolean nodeConnected(DiscoveryNode discoveryNode) {