Addressed one more comment

Author: tdas

sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/CheckpointFileManagerSuite.scala

@@ -101,10 +101,10 @@ class CheckpointFileManagerSuite extends SparkFunSuite with SharedSparkSession {
   test("CheckpointFileManager.create() should pick up user-specified class from conf") {
     withSQLConf(
       SQLConf.STREAMING_CHECKPOINT_FILE_MANAGER_CLASS.parent.key ->
-        classOf[TestCheckpointFileManager].getName) {
+        classOf[CreateAtomicTestManager].getName) {
       val fileManager =
         CheckpointFileManager.create(new Path("/"), spark.sessionState.newHadoopConf)
-      assert(fileManager.isInstanceOf[TestCheckpointFileManager])
+      assert(fileManager.isInstanceOf[CreateAtomicTestManager])
     }
   }
 
@@ -142,34 +142,34 @@ class FileSystemBasedCheckpointFileManagerSuite extends CheckpointFileManagerTes
 
 
 /** A fake implementation to test different characteristics of CheckpointFileManager interface */
-class TestCheckpointFileManager(path: Path, hadoopConf: Configuration)
+class CreateAtomicTestManager(path: Path, hadoopConf: Configuration)
   extends FileSystemBasedCheckpointFileManager(path, hadoopConf) {
 
   import CheckpointFileManager._
 
   override def createAtomic(path: Path, overwrite: Boolean): CancellableFSDataOutputStream = {
-    if (TestCheckpointFileManager.shouldFailInCreateAtomic) {
-      TestCheckpointFileManager.cancelCalledInCreateAtomic = false
+    if (CreateAtomicTestManager.shouldFailInCreateAtomic) {
+      CreateAtomicTestManager.cancelCalledInCreateAtomic = false
     }
     val originalOut = super.createAtomic(path, overwrite)
 
     new CancellableFSDataOutputStream(originalOut) {
       override def close(): Unit = {
-        if (TestCheckpointFileManager.shouldFailInCreateAtomic) {
+        if (CreateAtomicTestManager.shouldFailInCreateAtomic) {
           throw new IOException("Copy failed intentionally")
         }
         super.close()
       }
 
       override def cancel(): Unit = {
-        TestCheckpointFileManager.cancelCalledInCreateAtomic = true
+        CreateAtomicTestManager.cancelCalledInCreateAtomic = true
         originalOut.cancel()
       }
     }
   }
 }
 
-object TestCheckpointFileManager {
+object CreateAtomicTestManager {
   @volatile var shouldFailInCreateAtomic = false
   @volatile var cancelCalledInCreateAtomic = false
 }

sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/CompactibleFileStreamLogSuite.scala

@@ -22,7 +22,6 @@ import java.nio.charset.StandardCharsets._
 
 import org.apache.spark.{SparkConf, SparkFunSuite}
 import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.execution.streaming.FakeFileSystem._
 import org.apache.spark.sql.test.SharedSQLContext
 
 class CompactibleFileStreamLogSuite extends SparkFunSuite with SharedSQLContext {

sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StateStoreSuite.scala

@@ -475,34 +475,36 @@ class StateStoreSuite extends StateStoreSuiteBase[HDFSBackedStateStoreProvider]
     val hadoopConf = new Configuration()
     hadoopConf.set(
       SQLConf.STREAMING_CHECKPOINT_FILE_MANAGER_CLASS.parent.key,
-      classOf[TestCheckpointFileManager].getName)
+      classOf[CreateAtomicTestManager].getName)
     val remoteDir = Utils.createTempDir().getAbsolutePath
 
     val provider = newStoreProvider(
       opId = Random.nextInt, partition = 0, dir = remoteDir, hadoopConf = hadoopConf)
 
     // Disable failure of output stream and generate versions
-    TestCheckpointFileManager.shouldFailInCreateAtomic = false
+    CreateAtomicTestManager.shouldFailInCreateAtomic = false
     for (version <- 1 to 10) {
       val store = provider.getStore(version - 1)
       put(store, version.toString, version) // update "1" -> 1, "2" -> 2, ...
       store.commit()
     }
     val version10Data = (1L to 10).map(_.toString).map(x => x -> x).toSet
 
+    CreateAtomicTestManager.cancelCalledInCreateAtomic = false
     val store = provider.getStore(10)
     // Fail commit for next version and verify that reloading resets the files
-    TestCheckpointFileManager.shouldFailInCreateAtomic = true
+    CreateAtomicTestManager.shouldFailInCreateAtomic = true
     put(store, "11", 11)
     val e = intercept[IllegalStateException] { quietly { store.commit() } }
-    assert(e.getCause.isInstanceOf[IOException], "Was waiting the IOException to be thrown")
-    TestCheckpointFileManager.shouldFailInCreateAtomic = false
+    assert(e.getCause.isInstanceOf[IOException])
+    CreateAtomicTestManager.shouldFailInCreateAtomic = false
 
     // Abort commit for next version and verify that reloading resets the files
+    CreateAtomicTestManager.cancelCalledInCreateAtomic = false
     val store2 = provider.getStore(10)
     put(store2, "11", 11)
     store2.abort()
-    assert(TestCheckpointFileManager.cancelCalledInCreateAtomic)
+    assert(CreateAtomicTestManager.cancelCalledInCreateAtomic)
   }
 
   override def newStoreProvider(): HDFSBackedStateStoreProvider = {