CODEWITHSUNDEEP
apache-flink
Tom
Tom

Reputation: 6342

No checkpoint files are created in my simple application

I have the following simple flink application running within IDE, and I do a checkpoint every 5 seconds, and would like to write the checkpoint data into directory file:///d:/applog/out/mycheckpoint/, but after running for a while, i stop the application,but I didn't find anything under the directory file:///d:/applog/out/mycheckpoint/

The code is:

import java.util.Date

import io.github.streamingwithflink.util.DateUtil
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.api.scala._
import org.apache.flink.runtime.state.filesystem.FsStateBackend
import org.apache.flink.runtime.state.{FunctionInitializationContext, FunctionSnapshotContext}
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction
import org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
    object SourceFunctionExample {

      def main(args: Array[String]): Unit = {

        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(4)
        env.getCheckpointConfig.setCheckpointInterval(5 * 1000)
        env.getCheckpointConfig.enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
        env.setStateBackend(new FsStateBackend("file:///d:/applog/out/mycheckpoint/"))
        val numbers: DataStream[Long] = env.addSource(new ReplayableCountSource)
        numbers.print()

        env.execute()
      }

    }


class ReplayableCountSource extends SourceFunction[Long] with CheckpointedFunction {

  var isRunning: Boolean = true
  var cnt: Long = _
  var offsetState: ListState[Long] = _

  override def run(ctx: SourceFunction.SourceContext[Long]): Unit = {

    while (isRunning && cnt < Long.MaxValue) {
      ctx.getCheckpointLock.synchronized {
        // increment cnt
        cnt += 1
        ctx.collect(cnt)
      }
      Thread.sleep(200)
    }

  }

  override def cancel(): Unit = isRunning = false

  override def snapshotState(snapshotCtx: FunctionSnapshotContext): Unit = {
    println("snapshotState is called at " + DateUtil.format(new Date) + s", cnt is ${cnt}")
    // remove previous cnt
    offsetState.clear()
    // add current cnt
    offsetState.add(cnt)
  }

  override def initializeState(initCtx: FunctionInitializationContext): Unit = {
    // obtain operator list state to store the current cnt

    val desc = new ListStateDescriptor[Long]("offset", classOf[Long])
    offsetState = initCtx.getOperatorStateStore.getListState(desc)

    // initialize cnt variable from the checkpoint
    val it = offsetState.get()
    cnt = if (null == it || !it.iterator().hasNext) {
      -1L
    } else {
      it.iterator().next()
    }

    println("initializeState is called at " + DateUtil.format(new Date) + s", cnt is ${cnt}")
  }
}

Upvotes: 0

Views: 202

Answers (1)

Fabian Hueske
Fabian Hueske

Reputation: 18987

I tested the application on Windows and Linux and in both cases the checkpoint files were created as expected.

Note that the program keeps running if a checkpoint fails, for example due to some permission errors or invalid path.

Flink logs a WARN message with the exception that caused the checkpoint to fail.

Upvotes: 2

Related Questions