Repository: apache/flink
File: flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/utils/StreamingWithStateTestBase.scala (Scala)

Code before:
import java.util

import scala.collection.JavaConversions._
import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer

class StreamingWithStateTestBase(state: StateBackendMode) extends StreamingTestBase {

  enableObjectReuse = state match {
    case HEAP_BACKEND => false // TODO heap statebackend not support obj reuse now.
    case ROCKSDB_BACKEND => true
  }

  private val log = LoggerFactory.getLogger(classOf[StreamingWithStateTestBase])

  private val classLoader = Thread.currentThread.getContextClassLoader

  var baseCheckpointPath: File = _

  @BeforeEach
  override def before(): Unit = {
    super.before()
    // set state backend

    val baseCheckpointPath = Files.createTempDirectory(getClass.getCanonicalName)
    Files.deleteIfExists(baseCheckpointPath);
    state match {
      case HEAP_BACKEND =>
        val conf = new Configuration()
        env.setStateBackend(
          new MemoryStateBackend("file://" + baseCheckpointPath, null).configure(conf, classLoader))
      case ROCKSDB_BACKEND =>
        val conf = new Configuration()
        conf.setBoolean(CheckpointingOptions.INCREMENTAL_CHECKPOINTS, true)
        env.setStateBackend(
          new RocksDBStateBackend("file://" + baseCheckpointPath).configure(conf, classLoader))
    }
    this.tEnv = StreamTableEnvironment.create(env, TableTestUtil.STREAM_SETTING)
    FailingCollectionSource.failedBefore = true
  }

  @AfterEach
  override def after(): Unit = {
    super.after()
    try {
      FileUtils.deleteDirectory(baseCheckpointPath)
    } catch {
      case e: IOException =>
        if (baseCheckpointPath.exists) {
          log.error(
            s"The temporary files are not being deleted gracefully, remaining files " +

Review comment (bug):
```suggestion
    baseCheckpointPath = Files.createTempDirectory("junit")
```


I noticed that previous failed stack trace:
```
Suppressed: java.nio.file.NoSuchFileException: /tmp/junit8233404746490819295/junit2880192188533757139/71dc52714210ccdbd137bbcffa7955b6/chk-3
		at sun.nio.fs.UnixException.translateToIOException(UnixException.java:86)
		at sun.nio.fs.UnixException.rethrowAsIOException(UnixException.java:102)
		at sun.nio.fs.UnixException.rethrowAsIOException(UnixException.java:107)
		at sun.nio.fs.UnixFileAttributeViews$Basic.readAttributes(UnixFileAttributeViews.java:55)
		at sun.nio.fs.UnixFileSystemProvider.readAttributes(UnixFileSystemProvider.java:144)
		at sun.nio.fs.LinuxFileSystemProvider.readAttributes(LinuxFileSystemProvider.java:99)
		at java.nio.file.Files.readAttributes(Files.java:1737)
		at java.nio.file.FileTreeWalker.getAttributes(FileTreeWalker.java:219)
		at java.nio.file.FileTreeWalker.visit(FileTreeWalker.java:276)
		at java.nio.file.FileTreeWalker.next(FileTreeWalker.java:372)
		at java.nio.file.Files.walkFileTree(Files.java:2706)
		at java.nio.file.Files.walkFileTree(Files.java:2742)
		at org.junit.jupiter.engine.extension.TempDirectory$CloseablePath.deleteAllFilesAndDirectories(TempDirectory.java:329)
		at org.junit.jupiter.engine.extension.TempDirectory$CloseablePath.close(TempDirectory.java:310)
		... 96 more
		Suppressed: java.nio.file.NoSuchFileException: /tmp/junit8233404746490819295/junit2880192188533757139/71dc52714210ccdbd137bbcffa7955b6/chk-3
			at sun.nio.fs.UnixException.translateToIOException(UnixException.java:86)
			at sun.nio.fs.UnixException.rethrowAsIOException(UnixException.java:102)
			at sun.nio.fs.UnixException.rethrowAsIOException(UnixException.java:107)
			at sun.nio.fs.UnixFileSystemProvider.implDelete(UnixFileSystemProvider.java:244)
			at sun.nio.fs.AbstractFileSystemProvider.delete(AbstractFileSystemProvider.java:103)
			at java.nio.file.Files.delete(Files.java:1126)
			at org.junit.jupiter.engine.extension.TempDirectory$CloseablePath$1.resetPermissionsAndTryToDeleteAgain(TempDirectory.java:382)
			at org.junit.jupiter.engine.extension.TempDirectory$CloseablePath$1.visitFileFailed(TempDirectory.java:342)
			at org.junit.jupiter.engine.extension.TempDirectory$CloseablePath$1.visitFileFailed(TempDirectory.java:329)
			at java.nio.file.Files.walkFileTree(Files.java:2672)
			... 99 more
```

The problem seems to have some clues. The failure to junit clean up `TempDir` may be caused by the attempt to delete it concurrently with the `CheckpointCoordinator` cleaning up completed Checkpoints.

https://github.com/apache/flink/blob/45f966e8c3c5e903b3843391874f7d2478122d8c/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java#L1387-L1390

We can use `baseCheckpointPath = Files.createTempDirectory("junit")` to avoid using junit to clean up the `TempForlder`. This will be safe for upgrading junit to version 5.10.1.

By the way, the root cause of the previous deletion failed is likely to be caused by the automatic cleaning complete checkpoint  by `Checkpointcoordinator`. I believe we can solve it soon. 🤔️

Code after:
import scala.collection.mutable.ArrayBuffer

class StreamingWithStateTestBase(state: StateBackendMode) extends StreamingTestBase {

  enableObjectReuse = state match {
    case HEAP_BACKEND => false // TODO heap statebackend not support obj reuse now.
    case ROCKSDB_BACKEND => true
  }

  private val classLoader = Thread.currentThread.getContextClassLoader

  @BeforeEach
  override def before(): Unit = {
    super.before()
    // set state backend

    val baseCheckpointPath = Files.createTempDirectory(getClass.getCanonicalName)
    baseCheckpointPath.toFile.deleteOnExit();
    state match {
      case HEAP_BACKEND =>
        val conf = new Configuration()
        env.setStateBackend(
          new MemoryStateBackend("file://" + baseCheckpointPath, null).configure(conf, classLoader))
      case ROCKSDB_BACKEND =>
        val conf = new Configuration()
        conf.setBoolean(CheckpointingOptions.INCREMENTAL_CHECKPOINTS, true)
        env.setStateBackend(
          new RocksDBStateBackend("file://" + baseCheckpointPath).configure(conf, classLoader))
    }
    this.tEnv = StreamTableEnvironment.create(env, TableTestUtil.STREAM_SETTING)
    FailingCollectionSource.failedBefore = true
  }

  @AfterEach
  override def after(): Unit = {
    super.after()
    assertThat(FailingCollectionSource.failedBefore).isTrue
  }

  /** Creates a BinaryRowData DataStream from the given non-empty [[Seq]]. */
  def failingBinaryRowSource[T: TypeInformation](data: Seq[T]): DataStream[RowData] = {
    val typeInfo = implicitly[TypeInformation[_]].asInstanceOf[CompositeType[_]]
    val result = new mutable.MutableList[RowData]
    val reuse = new BinaryRowData(typeInfo.getArity)
    val writer = new BinaryRowWriter(reuse)
    data.foreach {
      case p: Product =>
        for (i <- 0 until typeInfo.getArity) {
          val fieldType = typeInfo.getTypeAt(i).asInstanceOf[TypeInformation[_]]
          fieldType match {
            case Types.INT => writer.writeInt(i, p.productElement(i).asInstanceOf[Int])