flink 如何实现对watermark 的checkpoint,防止数据复写

importjava.util.{Collections, Date}
importjava.util

importscala.collection.JavaConverters._
import java.lang.{Long =>JavaLong}
importjava.text.SimpleDateFormat
importjava.util.concurrent.TimeUnit

importorg.apache.flink.api.common.restartstrategy.RestartStrategies
importorg.apache.flink.api.common.time.Time
importorg.apache.flink.streaming.api.scala.StreamExecutionEnvironment
importorg.apache.flink.api.scala._
importorg.apache.flink.contrib.streaming.state.RocksDBStateBackend
importorg.apache.flink.streaming.api.{CheckpointingMode, TimeCharacteristic}
importorg.apache.flink.streaming.api.checkpoint.ListCheckpointed
importorg.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup
importorg.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
importorg.apache.flink.streaming.api.watermark.Watermark

/*** CheckpointCount
 */object WatermarkCheckpoint {

  case class Goods(var id: Int = 0, var count: Int = 0, var time: Long = 0L) {
    override def toString: String = s"Goods(id=$id,count=$count,time=$time)"}

  def main(args: Array[String]): Unit ={
    val dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.sss")
    val env =StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    env.enableCheckpointing(1000 * 10)
    env.getCheckpointConfig.setCheckpointTimeout(1000 * 60) //checkpoint 超时时间
    env.getCheckpointConfig.setMinPauseBetweenCheckpoints(1000 * 5) //两次 checkpoint 的最小间隔
    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE) //checkpoint 模式
    env.getCheckpointConfig.setMaxConcurrentCheckpoints(2) //checkpoint 并发数
    env.getCheckpointConfig.enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION) //cancel job 时持久化checkopint
    env.getCheckpointConfig.setFailOnCheckpointingErrors(false) //当checkpoint 失败时不会导致任务失败终止
    //restart strategy
env.setRestartStrategy(
      RestartStrategies.fixedDelayRestart(2, Time.of(5, TimeUnit.SECONDS))
    )
    //state backend
    val file_rocksdb = "file:///tmp/state/rocksdb"  //需要提前建立路径
    env.setStateBackend(new RocksDBStateBackend(file_rocksdb, true))
    env.setParallelism(1)

    env.socketTextStream("localhost", 9999)
      .filter(_.nonEmpty)
      .map(x =>{
        val arr = x.split(",")
        val g = Goods(arr(0).toInt, arr(1).toInt, arr(2).toLong) //id,count,time
        println(s"input:$g")
        g
      })

      //watermark 没有 checkpoint
      /*.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks[Goods] {
        val maxOutOfOrderness = 2L // 最大无序数据到达的时间，用来生成水印2ms
        var currentMaxTimestamp: Long = _

        override def getCurrentWatermark: Watermark = {
          println(s"${dateFormat.format(new Date().getTime)} -------watermark: ${currentMaxTimestamp - maxOutOfOrderness}")
          new Watermark(currentMaxTimestamp - maxOutOfOrderness)
        }

        override def extractTimestamp(element: Goods, previousElementTimestamp: Long): Long = {
          currentMaxTimestamp = Math.max(element.time, currentMaxTimestamp)
          element.time
        }
      })*/

      //watermark  checkpoint
      .assignTimestampsAndWatermarks(newAssignerWithPeriodicWatermarks[Goods] with ListCheckpointed[JavaLong] {
        val maxOutOfOrderness = 2L //最大无序数据到达的时间，用来生成水印2ms
        var currentMaxTimestamp: Long =_

        override def getCurrentWatermark: Watermark ={
          println(s"${dateFormat.format(new Date().getTime)} -------watermark: ${currentMaxTimestamp - maxOutOfOrderness}")
          new Watermark(currentMaxTimestamp -maxOutOfOrderness)
        }

        override def extractTimestamp(element: Goods, previousElementTimestamp: Long): Long ={
          currentMaxTimestamp =Math.max(element.time, currentMaxTimestamp)
          element.time
        }

        override def snapshotState(checkpointId: Long, timestamp: Long): util.List[JavaLong] ={
          Collections.singletonList(currentMaxTimestamp)
        }

        override def restoreState(state: util.List[JavaLong]): Unit ={
          val stateMin =state.asScala.min
          if (stateMin > 0) currentMaxTimestamp =stateMin
        }
      })

      .map(x =>{
        if (x.id == 0) throw new RuntimeException("id is 0")
      })
      .print()

    env.execute(this.getClass.getSimpleName)
  }
}