Address comments

Author: xuanyuanking

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

@@ -70,6 +70,7 @@ public String toString() {
       .add("shuffleId", shuffleId)
       .add("mapIds", Arrays.toString(mapIds))
       .add("reduceIds", Arrays.deepToString(reduceIds))
+      .add("batchFetchEnabled", batchFetchEnabled)
       .toString();
   }
 

core/src/main/scala/org/apache/spark/network/netty/NettyBlockRpcServer.scala

@@ -73,8 +73,8 @@ class NettyBlockRpcServer(
           } else {
             val startAndEndId = fetchShuffleBlocks.reduceIds(index)
             if (startAndEndId.length != 2) {
-              throw new IllegalStateException(s"Invalid shuffle fetch request: " +
-                s"$fetchShuffleBlocks")
+              throw new IllegalStateException(s"Invalid shuffle fetch request when batch mode " +
+                s"is enabled: $fetchShuffleBlocks")
             }
             Array(blockManager.getBlockData(
               ShuffleBlockBatchId(

core/src/main/scala/org/apache/spark/shuffle/BlockStoreShuffleReader.scala

@@ -21,7 +21,6 @@ import org.apache.spark._
 import org.apache.spark.internal.{config, Logging}
 import org.apache.spark.io.CompressionCodec
 import org.apache.spark.serializer.SerializerManager
-import org.apache.spark.shuffle.sort.BypassMergeSortShuffleHandle
 import org.apache.spark.storage.{BlockManager, ShuffleBlockFetcherIterator}
 import org.apache.spark.util.CompletionIterator
 import org.apache.spark.util.collection.ExternalSorter
@@ -48,10 +47,21 @@ private[spark] class BlockStoreShuffleReader[K, C](
     val compressed = conf.get(config.SHUFFLE_COMPRESS)
     val featureEnabled = conf.get(config.SHUFFLE_FETCH_CONTINUOUS_BLOCKS_IN_BATCH)
     val serializerRelocatable = dep.serializer.supportsRelocationOfSerializedObjects
+    val codecConcatenation = if (compressed) {
+      CompressionCodec.supportsConcatenationOfSerializedStreams(CompressionCodec.createCodec(conf))
+    } else {
+      true
+    }
 
-    featureEnabled && endPartition - startPartition > 1 &&
-      serializerRelocatable && (!compressed || CompressionCodec
-        .supportsConcatenationOfSerializedStreams(CompressionCodec.createCodec(conf)))
+    val res = featureEnabled && endPartition - startPartition > 1 &&
+      serializerRelocatable && (!compressed || codecConcatenation)
+    if (featureEnabled && !res) {
+      logWarning("The feature tag of continuous shuffle block fetching is set to true, but " +
+        "we can not enable the feature because other conditions are not satisfied. " +
+        s"Shuffle compress: $compressed, serializer relocatable: $serializerRelocatable, " +
+        s"codec concatenation: $codecConcatenation")
+    }
+    res
   }
 
   /** Read the combined key-values for this reduce task */