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 org.assertj.core.api.Assertions.assertThat
import org.junit.jupiter.api.{AfterEach, BeforeEach}
import org.slf4j.LoggerFactory

import java.io.{File, IOException}
import java.nio.file.Files
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()

Review comment (bug):
Thanks @snuyanzin for this hard work.

But I am a little confused, why is `baseCheckpointPath` defined as a stack variable, so does the object variable `baseCheckpointPath: File = _` have no effect anymore?

I suspect that there are still some problems about reuse the `tempFolder` by `TempDirUtils.newFolder(tempFolder)` instead of directly using `Files.createTempDirectory("junit")`.
I think the cleanup of `tempFolder` may still fail when `FileUtils.deleteDirectory(baseCheckpointPath)` met `DirectoryNotEmptyException` because it is possible that there are some threads still writing to the folder.



Code after:
import org.junit.jupiter.api.{AfterEach, BeforeEach}

import java.nio.file.Files
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 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)