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

import java.util.{Collections, Date}
import java.util

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

import org.apache.flink.api.common.restartstrategy.RestartStrategies
import org.apache.flink.api.common.time.Time
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.api.scala._
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend
import org.apache.flink.streaming.api.{CheckpointingMode, TimeCharacteristic}
import org.apache.flink.streaming.api.checkpoint.ListCheckpointed
import org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
import org.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(new AssignerWithPeriodicWatermarks[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)
  }
}