Usually de/serialize the response too

Author: DaveCTurner

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

@@ -26,13 +26,15 @@
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.component.AbstractComponent;
+import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.lease.Releasable;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.threadpool.ThreadPool.Names;
 import org.elasticsearch.transport.TransportException;
 import org.elasticsearch.transport.TransportResponseHandler;
 import org.elasticsearch.transport.TransportService;
 
+import java.io.IOException;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.LongConsumer;
@@ -127,6 +129,11 @@ void start(final Iterable<DiscoveryNode> broadcastNodes) {
             logger.debug("{} requesting pre-votes from {}", this, broadcastNodes);
             broadcastNodes.forEach(n -> transportService.sendRequest(n, REQUEST_PRE_VOTE_ACTION_NAME, preVoteRequest,
                 new TransportResponseHandler<PreVoteResponse>() {
+                    @Override
+                    public PreVoteResponse read(StreamInput in) throws IOException {
+                        return new PreVoteResponse(in);
+                    }
+
                     @Override
                     public void handleResponse(PreVoteResponse response) {
                         handlePreVoteResponse(response, n);

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

@@ -19,7 +19,6 @@
 
 package org.elasticsearch.test.transport;
 
-import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.Nullable;
@@ -91,7 +90,18 @@ public TransportService createTransportService(Settings settings, ThreadPool thr
     public void handleResponse(final long requestId, final TransportResponse response) {
         final TransportResponseHandler transportResponseHandler = responseHandlers.onResponseReceived(requestId, listener);
         if (transportResponseHandler != null) {
-            transportResponseHandler.handleResponse(response);
+            final TransportResponse deliveredResponse;
+            if (rarely()) {
+                deliveredResponse = response;
+            } else {
+                try (BytesStreamOutput output = new BytesStreamOutput()) {
+                    response.writeTo(output);
+                    deliveredResponse = transportResponseHandler.read(output.bytes().streamInput());
+                } catch (IOException | UnsupportedOperationException e) {
+                    throw new AssertionError("failed to serialize/deserialize response " + response, e);
+                }
+            }
+            transportResponseHandler.handleResponse(deliveredResponse);
         }
     }
 
@@ -126,7 +136,7 @@ public void handleRemoteError(final long requestId, final Throwable t) {
                 output.writeException(t);
                 remoteException = new RemoteTransportException("remote failure", output.bytes().streamInput().readException());
             } catch (IOException ioException) {
-                throw new ElasticsearchException("failed to serialize/deserialize supplied exception " + t, ioException);
+                throw new AssertionError("failed to serialize/deserialize supplied exception " + t, ioException);
             }
         }
         this.handleError(requestId, remoteException);