Failed to restore checkpointed states using Flink Processor API

Question

The main program is consuming kafka events, then filter -> map -> keyBy -> CEP -> sink. I wrote another separate simple program to read checkpoint directory like the following:

object StateReader extends App {

  val path = "file://...."

  val env = ExecutionEnvironment.getExecutionEnvironment

  val chk = Savepoint.load(env.getJavaEnv, path, new FsStateBackend(path))

  val ds = chk.readKeyedState("cep", new CepOperatorReadFunction, TypeInformation.of(classOf[KEY]), TypeInformation.of(classOf[VALUE]))
  println(ds.count())

}

class CepOperatorReadFunction extends KeyedStateReaderFunction[KEY, VALUE] {
  override def open(parameters: Configuration): Unit = {

  }
  override def readKey(k: KEY, context: KeyedStateReaderFunction.Context, collector: Collector[VALUE]): Unit = {

  }//end readKey
}//end class CepOperatorReadFunction

However I got the following exception:

Caused by: java.lang.IllegalStateException: Unexpected state handle type, expected: class org.apache.flink.runtime.state.KeyGroupsStateHandle, but found: class org.apache.flink.runtime.state.IncrementalRemoteKeyedStateHandle
    at org.apache.flink.runtime.state.heap.HeapRestoreOperation.restore(HeapRestoreOperation.java:120)
    at org.apache.flink.runtime.state.heap.HeapKeyedStateBackendBuilder.build(HeapKeyedStateBackendBuilder.java:114)
    ... 13 more

Here are some configuraitons in flink-conf.yaml

state.backend: rocksdb
state.checkpoints.dir: hdfs:///.../checkpoints
state.savepoints.dir: hdfs:///.../savepoints
state.backend.incremental: true
state.backend.rocksdb.memory.write-buffer-ratio: 0.6
state.backend.rocksdb.localdir: /var/lib/.../rocksdb
execution.checkpointing.interval: 900000
execution.checkpointing.timeout: 600000
execution.checkpointing.unaligned: true
execution.checkpointing.mode: EXACTLY_ONCE
execution.checkpointing.max-concurrent-checkpoints: 1
execution.checkpointing.min-pause: 0

Any ideas why the exception happened and how to fix the problem?

Thanks

Answer 1

There's no out-of-the-box support to make reading the CEP operator's state easy. So to implement your KeyedStateReaderFunction, you'll have to dig into the CEP implementation, find the ValueStates and MapStates that are used, and implement a reader that uses those same state descriptors.