Remove unused thread pool in S3ShuffleReader and cleanup unused imports.

Signed-off-by: Pascal Spörri <psp@zurich.ibm.com>

Author: pspoerri

src/main/scala/org/apache/spark/shuffle/S3ShuffleMapOutputWriter.scala

@@ -8,7 +8,6 @@ package org.apache.spark.shuffle
 import org.apache.hadoop.fs.FSDataOutputStream
 import org.apache.spark.SparkConf
 import org.apache.spark.internal.Logging
-import org.apache.spark.internal.config.SHUFFLE_FILE_BUFFER_SIZE
 import org.apache.spark.shuffle.IndexShuffleBlockResolver.NOOP_REDUCE_ID
 import org.apache.spark.shuffle.api.metadata.MapOutputCommitMessage
 import org.apache.spark.shuffle.api.{ShuffleMapOutputWriter, ShufflePartitionWriter, WritableByteChannelWrapper}

src/main/scala/org/apache/spark/shuffle/S3SingleSpillShuffleMapOutputWriter.scala

@@ -10,7 +10,7 @@ import org.apache.spark.shuffle.helper.{S3ShuffleDispatcher, S3ShuffleHelper}
 import org.apache.spark.storage.ShuffleDataBlockId
 import org.apache.spark.util.Utils
 
-import java.io.{BufferedOutputStream, File, FileInputStream}
+import java.io.{File, FileInputStream}
 
 class S3SingleSpillShuffleMapOutputWriter(shuffleId: Int, mapId: Long) extends SingleSpillShuffleMapOutputWriter {
 

src/main/scala/org/apache/spark/shuffle/helper/S3ShuffleDispatcher.scala

@@ -6,17 +6,14 @@
 package org.apache.spark.shuffle.helper
 
 import org.apache.hadoop.fs._
-import org.apache.hadoop.fs.impl.OpenFileParameters
 import org.apache.spark.deploy.SparkHadoopUtil
-import org.apache.spark.internal.config.{MAX_REMOTE_BLOCK_SIZE_FETCH_TO_MEM, SHUFFLE_FILE_BUFFER_SIZE}
 import org.apache.spark.internal.{Logging, config}
 import org.apache.spark.shuffle.ConcurrentObjectMap
 import org.apache.spark.storage._
 import org.apache.spark.{SparkConf, SparkEnv}
 
 import java.io.IOException
 import java.net.URI
-import java.util.concurrent.CompletableFuture
 import scala.concurrent.ExecutionContext.Implicits.global
 import scala.concurrent.duration.Duration
 import scala.concurrent.{Await, Future}
@@ -32,8 +29,6 @@ class S3ShuffleDispatcher extends Logging {
 
   // Required
   val rootDir = conf.get("spark.shuffle.s3.rootDir", defaultValue = "sparkS3shuffle")
-  private val isCOS = rootDir.startsWith("cos://")
-  private val isS3A = rootDir.startsWith("s3a://")
 
   // Optional
   val bufferSize: Int = conf.getInt("spark.shuffle.s3.bufferSize", defaultValue = 8 * 1024 * 1024)

src/main/scala/org/apache/spark/storage/S3ShuffleBlockStream.scala

@@ -9,7 +9,7 @@ import org.apache.spark.internal.Logging
 import org.apache.spark.shuffle.IndexShuffleBlockResolver.NOOP_REDUCE_ID
 import org.apache.spark.shuffle.helper.S3ShuffleDispatcher
 
-import java.io.{EOFException, IOException, InputStream}
+import java.io.{IOException, InputStream}
 
 /**
  * InputStream that reads data from a shuffleBlock, mapId and exposes an InputStream from startReduceId to endReduceId.

src/main/scala/org/apache/spark/storage/S3ShuffleReader.scala

@@ -28,12 +28,10 @@ import org.apache.spark.serializer.SerializerManager
 import org.apache.spark.shuffle.helper.{S3ShuffleDispatcher, S3ShuffleHelper}
 import org.apache.spark.shuffle.{BaseShuffleHandle, ShuffleReadMetricsReporter, ShuffleReader}
 import org.apache.spark.storage.ShuffleBlockFetcherIterator.FetchBlockInfo
-import org.apache.spark.util.{CompletionIterator, ThreadUtils}
+import org.apache.spark.util.CompletionIterator
 import org.apache.spark.util.collection.ExternalSorter
 import org.apache.spark.{InterruptibleIterator, SparkConf, SparkEnv, TaskContext}
 
-import scala.concurrent.ExecutionContext
-
 /**
  * This class was adapted from Apache Spark: BlockStoreShuffleReader.
  */
@@ -176,8 +174,3 @@ class S3ShuffleReader[K, C](
     }
   }
 }
-
-object S3ShuffleReader {
-  private lazy val asyncThreadPool = ThreadUtils.newDaemonCachedThreadPool("s3-shuffle-reader-async-thread-pool", S3ShuffleDispatcher.get.prefetchThreadPoolSize)
-  private lazy implicit val asyncExecutionContext = ExecutionContext.fromExecutorService(asyncThreadPool)
-}