More v3.5.3 backporting

Author: EnricoMi

core/src/main/scala/org/apache/spark/storage/FallbackStorage.scala

@@ -146,10 +146,6 @@ private[storage] class FileSystemSegmentManagedBuffer(
   override def convertToNetty(): AnyRef = {
     Unpooled.wrappedBuffer(nioByteBuffer());
   }
-
-  override def convertToNettyForSsl(): AnyRef = {
-    Unpooled.wrappedBuffer(nioByteBuffer());
-  }
 }
 
 private[spark] object FallbackStorage extends Logging {

core/src/test/scala/org/apache/spark/storage/FallbackStorageSuite.scala

@@ -167,7 +167,7 @@ class FallbackStorageSuite extends SparkFunSuite with LocalSparkContext {
         val clue = s"offset: $offset, length: $length"
 
         // creating the managed buffer does not open the file
-        val mfs = spy(fs)
+        val mfs = spy[FileSystem](fs)
         val buf = new FileSystemSegmentManagedBuffer(mfs, file, offset, length)
         verify(mfs, never()).open(mc.any[Path]())
         assert(buf.size() === length, clue)
@@ -186,11 +186,9 @@ class FallbackStorageSuite extends SparkFunSuite with LocalSparkContext {
           assert(bytes.mkString(",") === data.slice(offset, offset + length).mkString(","), clue)
         }
 
-        // getting a Netty ByteBufs opens the file again and again
+        // getting a Netty ByteBufs opens the file again
         assert(buf.convertToNetty().asInstanceOf[ByteBuf].release() === length > 0, clue)
         verify(mfs, times(3)).open(mc.any[Path]())
-        assert(buf.convertToNettyForSsl().asInstanceOf[ByteBuf].release() === length > 0, clue)
-        verify(mfs, times(4)).open(mc.any[Path]())
       }
     }
   }

core/src/test/scala/org/apache/spark/storage/ShuffleBlockFetcherIteratorSuite.scala

@@ -359,7 +359,6 @@ class ShuffleBlockFetcherIteratorSuite extends SparkFunSuite with PrivateMethodT
       verify(mockBuf, never()).nioByteBuffer()
       verify(mockBuf, never()).createInputStream()
       verify(mockBuf, never()).convertToNetty()
-      verify(mockBuf, never()).convertToNettyForSsl()
     }
 
     val allBlocks = localBlocks ++ remoteBlocks ++ hostLocalBlocks ++ fallbackBlocks
@@ -387,7 +386,6 @@ class ShuffleBlockFetcherIteratorSuite extends SparkFunSuite with PrivateMethodT
       verify(mockBuf, never()).nioByteBuffer()
       verify(mockBuf, times(1)).createInputStream()
       verify(mockBuf, never()).convertToNetty()
-      verify(mockBuf, never()).convertToNettyForSsl()
     }
   }
 
@@ -548,7 +546,6 @@ class ShuffleBlockFetcherIteratorSuite extends SparkFunSuite with PrivateMethodT
       verify(mockBuf, never()).nioByteBuffer()
       verify(mockBuf, never()).createInputStream()
       verify(mockBuf, never()).convertToNetty()
-      verify(mockBuf, never()).convertToNettyForSsl()
     }
 
     val allBlocks = mergedLocalBlocks ++ mergedFallbackBlocks ++ mergedRemoteBlocks ++
@@ -572,7 +569,6 @@ class ShuffleBlockFetcherIteratorSuite extends SparkFunSuite with PrivateMethodT
       verify(mockBuf, never()).nioByteBuffer()
       verify(mockBuf, times(1)).createInputStream()
       verify(mockBuf, never()).convertToNetty()
-      verify(mockBuf, never()).convertToNettyForSsl()
     }
 
     assert(blockManager.hostLocalDirManager.get.getCachedHostLocalDirs.size === 1)