代码

class TimeoutStateFunction extends ProcessFunction[ObjectNode, (String, Long)] {    lazy val state : ListState[CountWithTimestamp] = getRuntimeContext      .getListState(new ListStateDescriptor[CountWithTimestamp]("myState", classOf[CountWithTimestamp]))     override def processElement(value: ObjectNode, ctx: Context, out: Collector[(String, Long)]): Unit = {         //Stuff here     }     override def onTimer(timestamp: Long, ctx: OnTimerContext, out: Collector[(String, Long)]): Unit = {         //More Stuff here     }}

错误：

Error:(8, 7) class TimeoutStateFunction needs to be abstract, since method processElement in class ProcessFunction of type (x$1: com.fasterxml.jackson.databind.node.ObjectNode, x$2: org.apache.flink.streaming.api.functions.ProcessFunction[com.fasterxml.jackson.databind.node.ObjectNode,(String, Long)]#Context, x$3: org.apache.flink.util.Collector[(String, Long)])Unit is not definedclass TimeoutStateFunction extends ProcessFunction[ObjectNode, (String, Long)] {Error:(17, 18) method processElement overrides nothing.Note: the super classes of class TimeoutStateFunction contain the following, non final members named processElement:def processElement(x$1: com.fasterxml.jackson.databind.node.ObjectNode,x$2: org.apache.flink.streaming.api.functions.ProcessFunction[com.fasterxml.jackson.databind.node.ObjectNode,(String, Long)]#Context,x$3: org.apache.flink.util.Collector[(String, Long)]): Unit  override def processElement(value: ObjectNode, ctx: Context, out: Collector[(String, Long)]): Unit = {Error:(36, 16) method onTimer overrides nothing.Note: the super classes of class TimeoutStateFunction contain the following, non final members named onTimer:def onTimer(x$1: Long,x$2: org.apache.flink.streaming.api.functions.ProcessFunction[com.fasterxml.jackson.databind.node.ObjectNode,(String, Long)]#OnTimerContext,x$3: org.apache.flink.util.Collector[(String, Long)]): Unit   override def onTimer(timestamp: Long, ctx: OnTimerContext, out: Collector[(String, Long)]): Unit = {

原因：

Context 和 OnTimerContext 要指明为ProcessFunction内部的

解决方法：

override def processElement(     value: ObjectNode,     ctx: ProcessFunction[ObjectNode, (String, Long)]#Context,     out: Collector[(String, Long)])   : Unit = {     //Stuff here } override def onTimer(     timestamp: Long,     ctx: ProcessFunction[ObjectNode, (String, Long)]#OnTimerContext,     out: Collector[(String, Long)])   : Unit = {     //More Stuff here }