Revert "Extend FilterInputStream and use the stream directly (#613)"

This reverts commit 4fa14a3.

Author: rshkv

core/src/main/java/org/apache/spark/shuffle/api/ShuffleBlockInputStream.java

@@ -17,7 +17,6 @@
 
 package org.apache.spark.shuffle.api;
 
-import java.io.FilterInputStream;
 import org.apache.spark.storage.BlockId;
 
 import java.io.InputStream;
@@ -27,15 +26,20 @@
  * An object defining the shuffle block and length metadata associated with the block.
  * @since 3.0.0
  */
-public class ShuffleBlockInputStream extends FilterInputStream {
+public class ShuffleBlockInputStream {
   private final BlockId blockId;
+  private final InputStream inputStream;
 
   public ShuffleBlockInputStream(BlockId blockId, InputStream inputStream) {
-    super(inputStream);
     this.blockId = blockId;
+    this.inputStream = inputStream;
   }
 
   public BlockId getBlockId() {
     return this.blockId;
   }
+
+  public InputStream getInputStream() {
+    return this.inputStream;
+  }
 }

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

@@ -73,7 +73,8 @@ private[spark] class BlockStoreShuffleReader[K, C](
         }
       }.asJava).iterator()
 
-    val retryingWrappedStreams = streamsIterator.asScala.map { rawReaderStream =>
+    val retryingWrappedStreams = streamsIterator.asScala.map { shuffleBlock =>
+      val rawReaderStream = shuffleBlock.getInputStream
       if (shuffleExecutorComponents.shouldWrapPartitionReaderStream()) {
         if (compressShuffle) {
           compressionCodec.compressedInputStream(

project/plugins.sbt

@@ -1,4 +1,4 @@
-addSbtPlugin("com.typesafe.sbt" % "sbt-git" % "1.0.0")
+addSbtPlugin("com.typesafe.sbt" % "sbt-git" % "0.8.5")
 
 addSbtPlugin("com.etsy" % "sbt-checkstyle-plugin" % "3.1.1")