review feedback

common/network-common/src/main/java/org/apache/spark/network/client/StreamInterceptor.java

@@ -64,6 +64,8 @@ public void channelInactive() throws Exception {
 
   private void deactivateStream() {
     if (handler instanceof TransportResponseHandler) {
+      // we only have to do this for TransportResponseHandler as it exposes numOutstandingFetches
+      // (there is no extra cleanup that needs to happen)
       ((TransportResponseHandler) handler).deactivateStream();
     }
   }

common/network-common/src/main/java/org/apache/spark/network/protocol/StreamResponse.java

@@ -67,7 +67,7 @@ public static StreamResponse decode(ByteBuf buf) {
 
   @Override
   public int hashCode() {
-    return Objects.hashCode(byteCount, streamId, body());
+    return Objects.hashCode(byteCount, streamId);
   }
 
   @Override

common/network-common/src/main/java/org/apache/spark/network/protocol/UploadStream.java

@@ -85,7 +85,7 @@ public static UploadStream decode(ByteBuf buf) {
 
   @Override
   public int hashCode() {
-    return Objects.hashCode(requestId, body());
+    return Long.hashCode(requestId);
   }
 
   @Override

common/network-common/src/main/java/org/apache/spark/network/server/StreamData.java

@@ -33,20 +33,17 @@ public class StreamData {
   private final TransportRequestHandler handler;
   private final TransportFrameDecoder frameDecoder;
   private final RpcResponseCallback rpcCallback;
-  private final ByteBuffer meta;
   private final long streamByteCount;
   private boolean hasCallback = false;
 
   public StreamData(
       TransportRequestHandler handler,
       TransportFrameDecoder frameDecoder,
       RpcResponseCallback rpcCallback,
-      ByteBuffer meta,
       long streamByteCount) {
     this.handler = handler;
     this.frameDecoder = frameDecoder;
     this.rpcCallback = rpcCallback;
-    this.meta = meta;
     this.streamByteCount = streamByteCount;
   }
 

common/network-common/src/main/java/org/apache/spark/network/server/TransportRequestHandler.java

@@ -218,7 +218,7 @@ public void onFailure(Throwable e) {
           channel.pipeline().get(TransportFrameDecoder.HANDLER_NAME);
       ByteBuffer meta = req.meta.nioByteBuffer();
       StreamData streamData = new StreamData(TransportRequestHandler.this, frameDecoder,
-          callback, meta, req.bodyByteCount);
+          callback, req.bodyByteCount);
       rpcHandler.receive(reverseClient, meta, streamData, callback);
       if (!streamData.hasCallback()) {
         throw new RuntimeException("Destination did not register stream handler");