(DEPRECATED) Apache Flink User Mailing List archive.

Checkpoint failed because of TimeWindow cannot be cast to VoidNamespace

Classic

List

Threaded

13 messages Options

Si-li Liu

Checkpoint failed because of TimeWindow cannot be cast to VoidNamespace

I'm using flink 1.9 on Mesos and I try to use my own trigger and evictor. The state is stored to memory.

input.setParallelism(processParallelism)
.assignTimestampsAndWatermarks(new UETimeAssigner)
.keyBy(_.key)
.window(TumblingEventTimeWindows.of(Time.minutes(20)))
.trigger(new MyTrigger)
.evictor(new MyEvictor)
.process(new MyFunction).setParallelism(aggregateParallelism)
.addSink(kafkaSink).setParallelism(sinkParallelism)
.name("kafka-record-sink")

And the exception stack is here, could anyone help with this? Thanks!

java.lang.Exception: Could not materialize checkpoint 1 for operator Window(TumblingEventTimeWindows(1200000), JoinTrigger, JoinEvictor, ScalaProcessWindowFunctionWrapper) -> Sink: kafka-record-sink (2/5).
    at org.apache.flink.streaming.runtime.tasks.StreamTask$AsyncCheckpointRunnable.handleExecutionException(StreamTask.java:1100)
    at org.apache.flink.streaming.runtime.tasks.StreamTask$AsyncCheckpointRunnable.run(StreamTask.java:1042)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)
Caused by: java.util.concurrent.ExecutionException: java.lang.ClassCastException: org.apache.flink.streaming.api.windowing.windows.TimeWindow cannot be cast to org.apache.flink.runtime.state.VoidNamespace
    at java.util.concurrent.FutureTask.report(FutureTask.java:122)
    at java.util.concurrent.FutureTask.get(FutureTask.java:192)
    at org.apache.flink.runtime.concurrent.FutureUtils.runIfNotDoneAndGet(FutureUtils.java:450)
    at org.apache.flink.streaming.api.operators.OperatorSnapshotFinalizer.<init>(OperatorSnapshotFinalizer.java:47)
    at org.apache.flink.streaming.runtime.tasks.StreamTask$AsyncCheckpointRunnable.run(StreamTask.java:1011)
    ... 3 more
Caused by: java.lang.ClassCastException: org.apache.flink.streaming.api.windowing.windows.TimeWindow cannot be cast to org.apache.flink.runtime.state.VoidNamespace
    at org.apache.flink.runtime.state.VoidNamespaceSerializer.serialize(VoidNamespaceSerializer.java:32)
    at org.apache.flink.runtime.state.heap.CopyOnWriteStateMapSnapshot.writeState(CopyOnWriteStateMapSnapshot.java:114)
    at org.apache.flink.runtime.state.heap.AbstractStateTableSnapshot.writeStateInKeyGroup(AbstractStateTableSnapshot.java:121)
    at org.apache.flink.runtime.state.heap.CopyOnWriteStateTableSnapshot.writeStateInKeyGroup(CopyOnWriteStateTableSnapshot.java:37)
    at org.apache.flink.runtime.state.heap.HeapSnapshotStrategy$1.callInternal(HeapSnapshotStrategy.java:191)
    at org.apache.flink.runtime.state.heap.HeapSnapshotStrategy$1.callInternal(HeapSnapshotStrategy.java:158)
    at org.apache.flink.runtime.state.AsyncSnapshotCallable.call(AsyncSnapshotCallable.java:75)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at org.apache.flink.runtime.concurrent.FutureUtils.runIfNotDoneAndGet(FutureUtils.java:447)
    ... 5 more

Best regards

Sili Liu

r_khachatryan

Re: Checkpoint failed because of TimeWindow cannot be cast to VoidNamespace

Thanks for reporting this.

Looks like the window namespace was replaced by VoidNamespace in state entry.

I've created https://issues.apache.org/jira/browse/FLINK-18464 to further investigate it.

Regards,
Roman

On Thu, Jul 2, 2020 at 10:52 AM Si-li Liu <[hidden email]> wrote:

Si-li Liu

Re: Checkpoint failed because of TimeWindow cannot be cast to VoidNamespace

Rocksdb backend has the same problem

Khachatryan Roman <[hidden email]> 于2020年7月2日周四下午6:11写道：

Best regards

Sili Liu

r_khachatryan

Re: Checkpoint failed because of TimeWindow cannot be cast to VoidNamespace

Thanks for the clarification.

Can you also share the code of other parts, particularly MyFunction?

Regards,
Roman

On Thu, Jul 2, 2020 at 12:49 PM Si-li Liu <[hidden email]> wrote:

Si-li Liu

Re: Checkpoint failed because of TimeWindow cannot be cast to VoidNamespace

Hi, this is our production code so I have to modify it a little bit, such as variable name and function name. I think 3 classes I provide here is enough.

I try to join two streams, but I don't want to use the default join function, because I want to send the joined log immediately and remove it from window state immediately. And my window gap time is very long( 20 minutes), so it maybe evaluate it multiple times.

class JoinFunction extends
  ProcessWindowFunction[RawLog, OutputLog, String, TimeWindow]{

  var ueState: ValueState[RawLog] = _
  @transient var gZipThriftSerializer: GZipThriftSerializer[MyType] = _
  val invalidCounter = new LongCounter()
  val processCounter = new LongCounter()
  val sendToKafkaCounter = new LongCounter()

  override def open(parameters: Configuration): Unit = {
    ueState = getRuntimeContext.getState(
      new ValueStateDescriptor[RawLog](bidState, classOf[RawLog])
    )
    gZipThriftSerializer = new GZipThriftSerializer[MyType]()
    getRuntimeContext.addAccumulator("processCounter", this.processCounter)
    getRuntimeContext.addAccumulator("invalidCounter", this.invalidCounter)
    getRuntimeContext.addAccumulator("sendToKafkaCounter", this.sendToKafkaCounter)
  }

  override def process(key: String,
                       ctx: Context,
                       logs: Iterable[RawLog],
                       out: Collector[OutputLog]): Unit = {
    if (ueState.value() != null) {
      processCounter.add(1L)
      val bid = ueState.value()
      val bidLog = gZipThriftSerializer.decompressAndDeserialize(bid.payload, classOf[MyType])
      logs.foreach( log => {
        if (log.eventType == SHOW) {
          val showLog = gZipThriftSerializer.decompressAndDeserialize(log.payload, classOf[MyType])
          sendToKafkaCounter.add(1L)
          out.collect(new OutputLog(ThriftUtils.serialize(showLog), Utils.getOutputTopic(showLog)))
        }
      })
    } else {
      invalidCounter.add(1L)
    }
  }
}

class JoinTrigger extends Trigger[RawLog, TimeWindow] {

  override def onElement(log: RawLog,
                         timestamp: Long,
                         window: TimeWindow,
                         ctx: Trigger.TriggerContext): TriggerResult = {
    val ueState: ValueState[RawLog] = ctx.getPartitionedState(
      new ValueStateDescriptor[RawLog](bidState, classOf[RawLog])
    )
    val firstSeen: ValueState[Boolean] = ctx.getPartitionedState(
      new ValueStateDescriptor[Boolean](initState, classOf[Boolean]))

    if (!firstSeen.value()) {
      ctx.registerEventTimeTimer(window.getEnd)
      firstSeen.update(true)
    }
    val eventType = log.eventType
    if (eventType == BID) {
      ueState.update(log)
      TriggerResult.CONTINUE
    } else {
      if (ueState.value() == null) {
        TriggerResult.CONTINUE
      } else {
        TriggerResult.FIRE
      }
    }
  }

  override def onEventTime(timestamp: Long,
                           window: TimeWindow,
                           ctx: Trigger.TriggerContext): TriggerResult = {
    if (timestamp == window.getEnd) {
      TriggerResult.PURGE
    }
    TriggerResult.CONTINUE
  }

  override def onProcessingTime(timestamp: Long,
                                window: TimeWindow,
                                ctx: Trigger.TriggerContext): TriggerResult = {
    TriggerResult.CONTINUE
  }

  override def clear(window: TimeWindow,
                     ctx: Trigger.TriggerContext): Unit = {
    val ueState: ValueState[RawLog] = ctx.getPartitionedState(
      new ValueStateDescriptor[RawLog](bidState, classOf[RawLog])
    )
    ueState.clear()

    val firstSeen: ValueState[Boolean] = ctx.getPartitionedState(
      new ValueStateDescriptor[Boolean](initState, classOf[Boolean]))
    firstSeen.clear()

    ctx.deleteEventTimeTimer(window.getEnd)
  }
}

class JoinEvictor extends Evictor[RawLog, TimeWindow] {

  override def evictBefore(elements: JIterable[TimestampedValue[RawLog]],
                           size: Int,
                           window: TimeWindow,
                           evictorContext: Evictor.EvictorContext): Unit = {}

  override def evictAfter(elements: JIterable[TimestampedValue[RawLog]],
                           size: Int,
                           window: TimeWindow,
                           evictorContext: Evictor.EvictorContext): Unit = {
    val iter = elements.iterator()
    while (iter.hasNext) {
      iter.next()
      iter.remove()
    }
  }
}

Khachatryan Roman <[hidden email]> 于2020年7月2日周四下午7:18写道：

Best regards

Sili Liu

r_khachatryan

Re: Checkpoint failed because of TimeWindow cannot be cast to VoidNamespace

Hi,

Thanks for the details.

However, I was not able to reproduce the issue. I used parallelism levels 4, file system backend and tried different timings for checkpointing, windowing and source.

Do you encounter this problem deterministically, is it always 1st checkpoint?

What checkpointing interval do you use?

Regards,
Roman

On Thu, Jul 2, 2020 at 1:57 PM Si-li Liu <[hidden email]> wrote:

Hi, this is our production code so I have to modify it a little bit, such as variable name and function name. I think 3 classes I provide here is enough.

class JoinFunction extends
  ProcessWindowFunction[RawLog, OutputLog, String, TimeWindow]{

  var ueState: ValueState[RawLog] = _
  @transient var gZipThriftSerializer: GZipThriftSerializer[MyType] = _
  val invalidCounter = new LongCounter()
  val processCounter = new LongCounter()
  val sendToKafkaCounter = new LongCounter()

  override def open(parameters: Configuration): Unit = {
    ueState = getRuntimeContext.getState(
      new ValueStateDescriptor[RawLog](bidState, classOf[RawLog])
    )
    gZipThriftSerializer = new GZipThriftSerializer[MyType]()
    getRuntimeContext.addAccumulator("processCounter", this.processCounter)
    getRuntimeContext.addAccumulator("invalidCounter", this.invalidCounter)
    getRuntimeContext.addAccumulator("sendToKafkaCounter", this.sendToKafkaCounter)
  }

  override def process(key: String,
                       ctx: Context,
                       logs: Iterable[RawLog],
                       out: Collector[OutputLog]): Unit = {
    if (ueState.value() != null) {
      processCounter.add(1L)
      val bid = ueState.value()
      val bidLog = gZipThriftSerializer.decompressAndDeserialize(bid.payload, classOf[MyType])
      logs.foreach( log => {
        if (log.eventType == SHOW) {
          val showLog = gZipThriftSerializer.decompressAndDeserialize(log.payload, classOf[MyType])
          sendToKafkaCounter.add(1L)
          out.collect(new OutputLog(ThriftUtils.serialize(showLog), Utils.getOutputTopic(showLog)))
        }
      })
    } else {
      invalidCounter.add(1L)
    }
  }
}

class JoinTrigger extends Trigger[RawLog, TimeWindow] {

  override def onElement(log: RawLog,
                         timestamp: Long,
                         window: TimeWindow,
                         ctx: Trigger.TriggerContext): TriggerResult = {
    val ueState: ValueState[RawLog] = ctx.getPartitionedState(
      new ValueStateDescriptor[RawLog](bidState, classOf[RawLog])
    )
    val firstSeen: ValueState[Boolean] = ctx.getPartitionedState(
      new ValueStateDescriptor[Boolean](initState, classOf[Boolean]))

    if (!firstSeen.value()) {
      ctx.registerEventTimeTimer(window.getEnd)
      firstSeen.update(true)
    }
    val eventType = log.eventType
    if (eventType == BID) {
      ueState.update(log)
      TriggerResult.CONTINUE
    } else {
      if (ueState.value() == null) {
        TriggerResult.CONTINUE
      } else {
        TriggerResult.FIRE
      }
    }
  }

  override def onEventTime(timestamp: Long,
                           window: TimeWindow,
                           ctx: Trigger.TriggerContext): TriggerResult = {
    if (timestamp == window.getEnd) {
      TriggerResult.PURGE
    }
    TriggerResult.CONTINUE
  }

  override def onProcessingTime(timestamp: Long,
                                window: TimeWindow,
                                ctx: Trigger.TriggerContext): TriggerResult = {
    TriggerResult.CONTINUE
  }

  override def clear(window: TimeWindow,
                     ctx: Trigger.TriggerContext): Unit = {
    val ueState: ValueState[RawLog] = ctx.getPartitionedState(
      new ValueStateDescriptor[RawLog](bidState, classOf[RawLog])
    )
    ueState.clear()

    val firstSeen: ValueState[Boolean] = ctx.getPartitionedState(
      new ValueStateDescriptor[Boolean](initState, classOf[Boolean]))
    firstSeen.clear()

    ctx.deleteEventTimeTimer(window.getEnd)
  }
}

class JoinEvictor extends Evictor[RawLog, TimeWindow] {

  override def evictBefore(elements: JIterable[TimestampedValue[RawLog]],
                           size: Int,
                           window: TimeWindow,
                           evictorContext: Evictor.EvictorContext): Unit = {}

  override def evictAfter(elements: JIterable[TimestampedValue[RawLog]],
                           size: Int,
                           window: TimeWindow,
                           evictorContext: Evictor.EvictorContext): Unit = {
    val iter = elements.iterator()
    while (iter.hasNext) {
      iter.next()
      iter.remove()
    }
  }
}

Khachatryan Roman <[hidden email]> 于2020年7月2日周四下午7:18写道：

Best regards

Sili Liu

Si-li Liu

Re: Checkpoint failed because of TimeWindow cannot be cast to VoidNamespace

Hi, Thanks for your help.

The checkpoint configuration is

checkpoint.intervalMS=300000
checkpoint.timeoutMS=300000

The error callstack is from JM's log, which happened in every cp. Currently I don't have a success cp yet.

Khachatryan Roman <[hidden email]> 于2020年7月3日周五上午3:50写道：

Hi,

Thanks for the details.

However, I was not able to reproduce the issue. I used parallelism levels 4, file system backend and tried different timings for checkpointing, windowing and source.

Do you encounter this problem deterministically, is it always 1st checkpoint?

What checkpointing interval do you use?

Regards,
Roman

On Thu, Jul 2, 2020 at 1:57 PM Si-li Liu <[hidden email]> wrote:

Hi, this is our production code so I have to modify it a little bit, such as variable name and function name. I think 3 classes I provide here is enough.

class JoinFunction extends
  ProcessWindowFunction[RawLog, OutputLog, String, TimeWindow]{

  var ueState: ValueState[RawLog] = _
  @transient var gZipThriftSerializer: GZipThriftSerializer[MyType] = _
  val invalidCounter = new LongCounter()
  val processCounter = new LongCounter()
  val sendToKafkaCounter = new LongCounter()

  override def open(parameters: Configuration): Unit = {
    ueState = getRuntimeContext.getState(
      new ValueStateDescriptor[RawLog](bidState, classOf[RawLog])
    )
    gZipThriftSerializer = new GZipThriftSerializer[MyType]()
    getRuntimeContext.addAccumulator("processCounter", this.processCounter)
    getRuntimeContext.addAccumulator("invalidCounter", this.invalidCounter)
    getRuntimeContext.addAccumulator("sendToKafkaCounter", this.sendToKafkaCounter)
  }

  override def process(key: String,
                       ctx: Context,
                       logs: Iterable[RawLog],
                       out: Collector[OutputLog]): Unit = {
    if (ueState.value() != null) {
      processCounter.add(1L)
      val bid = ueState.value()
      val bidLog = gZipThriftSerializer.decompressAndDeserialize(bid.payload, classOf[MyType])
      logs.foreach( log => {
        if (log.eventType == SHOW) {
          val showLog = gZipThriftSerializer.decompressAndDeserialize(log.payload, classOf[MyType])
          sendToKafkaCounter.add(1L)
          out.collect(new OutputLog(ThriftUtils.serialize(showLog), Utils.getOutputTopic(showLog)))
        }
      })
    } else {
      invalidCounter.add(1L)
    }
  }
}

class JoinTrigger extends Trigger[RawLog, TimeWindow] {

  override def onElement(log: RawLog,
                         timestamp: Long,
                         window: TimeWindow,
                         ctx: Trigger.TriggerContext): TriggerResult = {
    val ueState: ValueState[RawLog] = ctx.getPartitionedState(
      new ValueStateDescriptor[RawLog](bidState, classOf[RawLog])
    )
    val firstSeen: ValueState[Boolean] = ctx.getPartitionedState(
      new ValueStateDescriptor[Boolean](initState, classOf[Boolean]))

    if (!firstSeen.value()) {
      ctx.registerEventTimeTimer(window.getEnd)
      firstSeen.update(true)
    }
    val eventType = log.eventType
    if (eventType == BID) {
      ueState.update(log)
      TriggerResult.CONTINUE
    } else {
      if (ueState.value() == null) {
        TriggerResult.CONTINUE
      } else {
        TriggerResult.FIRE
      }
    }
  }

  override def onEventTime(timestamp: Long,
                           window: TimeWindow,
                           ctx: Trigger.TriggerContext): TriggerResult = {
    if (timestamp == window.getEnd) {
      TriggerResult.PURGE
    }
    TriggerResult.CONTINUE
  }

  override def onProcessingTime(timestamp: Long,
                                window: TimeWindow,
                                ctx: Trigger.TriggerContext): TriggerResult = {
    TriggerResult.CONTINUE
  }

  override def clear(window: TimeWindow,
                     ctx: Trigger.TriggerContext): Unit = {
    val ueState: ValueState[RawLog] = ctx.getPartitionedState(
      new ValueStateDescriptor[RawLog](bidState, classOf[RawLog])
    )
    ueState.clear()

    val firstSeen: ValueState[Boolean] = ctx.getPartitionedState(
      new ValueStateDescriptor[Boolean](initState, classOf[Boolean]))
    firstSeen.clear()

    ctx.deleteEventTimeTimer(window.getEnd)
  }
}

class JoinEvictor extends Evictor[RawLog, TimeWindow] {

  override def evictBefore(elements: JIterable[TimestampedValue[RawLog]],
                           size: Int,
                           window: TimeWindow,
                           evictorContext: Evictor.EvictorContext): Unit = {}

  override def evictAfter(elements: JIterable[TimestampedValue[RawLog]],
                           size: Int,
                           window: TimeWindow,
                           evictorContext: Evictor.EvictorContext): Unit = {
    val iter = elements.iterator()
    while (iter.hasNext) {
      iter.next()
      iter.remove()
    }
  }
}

Khachatryan Roman <[hidden email]> 于2020年7月2日周四下午7:18写道：

Best regards

Sili Liu

r_khachatryan

Re: Checkpoint failed because of TimeWindow cannot be cast to VoidNamespace

I still wasn't able to reproduce the issue.

Can you also clarify:

- Are you starting the job from a savepoint or externalized checkpoint?

- If yes, was the job graph changed?

- What StreamTimeCharacteristic is set, if any?

- What exact version of Flink do you use?

Regards,
Roman

On Fri, Jul 3, 2020 at 6:38 AM Si-li Liu <[hidden email]> wrote:

Hi, Thanks for your help.

The checkpoint configuration is

checkpoint.intervalMS=300000
checkpoint.timeoutMS=300000

The error callstack is from JM's log, which happened in every cp. Currently I don't have a success cp yet.

Khachatryan Roman <[hidden email]> 于2020年7月3日周五上午3:50写道：

Hi,

Thanks for the details.

However, I was not able to reproduce the issue. I used parallelism levels 4, file system backend and tried different timings for checkpointing, windowing and source.

Do you encounter this problem deterministically, is it always 1st checkpoint?

What checkpointing interval do you use?

Regards,
Roman

On Thu, Jul 2, 2020 at 1:57 PM Si-li Liu <[hidden email]> wrote:

Hi, this is our production code so I have to modify it a little bit, such as variable name and function name. I think 3 classes I provide here is enough.

class JoinFunction extends
  ProcessWindowFunction[RawLog, OutputLog, String, TimeWindow]{

  var ueState: ValueState[RawLog] = _
  @transient var gZipThriftSerializer: GZipThriftSerializer[MyType] = _
  val invalidCounter = new LongCounter()
  val processCounter = new LongCounter()
  val sendToKafkaCounter = new LongCounter()

  override def open(parameters: Configuration): Unit = {
    ueState = getRuntimeContext.getState(
      new ValueStateDescriptor[RawLog](bidState, classOf[RawLog])
    )
    gZipThriftSerializer = new GZipThriftSerializer[MyType]()
    getRuntimeContext.addAccumulator("processCounter", this.processCounter)
    getRuntimeContext.addAccumulator("invalidCounter", this.invalidCounter)
    getRuntimeContext.addAccumulator("sendToKafkaCounter", this.sendToKafkaCounter)
  }

  override def process(key: String,
                       ctx: Context,
                       logs: Iterable[RawLog],
                       out: Collector[OutputLog]): Unit = {
    if (ueState.value() != null) {
      processCounter.add(1L)
      val bid = ueState.value()
      val bidLog = gZipThriftSerializer.decompressAndDeserialize(bid.payload, classOf[MyType])
      logs.foreach( log => {
        if (log.eventType == SHOW) {
          val showLog = gZipThriftSerializer.decompressAndDeserialize(log.payload, classOf[MyType])
          sendToKafkaCounter.add(1L)
          out.collect(new OutputLog(ThriftUtils.serialize(showLog), Utils.getOutputTopic(showLog)))
        }
      })
    } else {
      invalidCounter.add(1L)
    }
  }
}

class JoinTrigger extends Trigger[RawLog, TimeWindow] {

  override def onElement(log: RawLog,
                         timestamp: Long,
                         window: TimeWindow,
                         ctx: Trigger.TriggerContext): TriggerResult = {
    val ueState: ValueState[RawLog] = ctx.getPartitionedState(
      new ValueStateDescriptor[RawLog](bidState, classOf[RawLog])
    )
    val firstSeen: ValueState[Boolean] = ctx.getPartitionedState(
      new ValueStateDescriptor[Boolean](initState, classOf[Boolean]))

    if (!firstSeen.value()) {
      ctx.registerEventTimeTimer(window.getEnd)
      firstSeen.update(true)
    }
    val eventType = log.eventType
    if (eventType == BID) {
      ueState.update(log)
      TriggerResult.CONTINUE
    } else {
      if (ueState.value() == null) {
        TriggerResult.CONTINUE
      } else {
        TriggerResult.FIRE
      }
    }
  }

  override def onEventTime(timestamp: Long,
                           window: TimeWindow,
                           ctx: Trigger.TriggerContext): TriggerResult = {
    if (timestamp == window.getEnd) {
      TriggerResult.PURGE
    }
    TriggerResult.CONTINUE
  }

  override def onProcessingTime(timestamp: Long,
                                window: TimeWindow,
                                ctx: Trigger.TriggerContext): TriggerResult = {
    TriggerResult.CONTINUE
  }

  override def clear(window: TimeWindow,
                     ctx: Trigger.TriggerContext): Unit = {
    val ueState: ValueState[RawLog] = ctx.getPartitionedState(
      new ValueStateDescriptor[RawLog](bidState, classOf[RawLog])
    )
    ueState.clear()

    val firstSeen: ValueState[Boolean] = ctx.getPartitionedState(
      new ValueStateDescriptor[Boolean](initState, classOf[Boolean]))
    firstSeen.clear()

    ctx.deleteEventTimeTimer(window.getEnd)
  }
}

class JoinEvictor extends Evictor[RawLog, TimeWindow] {

  override def evictBefore(elements: JIterable[TimestampedValue[RawLog]],
                           size: Int,
                           window: TimeWindow,
                           evictorContext: Evictor.EvictorContext): Unit = {}

  override def evictAfter(elements: JIterable[TimestampedValue[RawLog]],
                           size: Int,
                           window: TimeWindow,
                           evictorContext: Evictor.EvictorContext): Unit = {
    val iter = elements.iterator()
    while (iter.hasNext) {
      iter.next()
      iter.remove()
    }
  }
}

Khachatryan Roman <[hidden email]> 于2020年7月2日周四下午7:18写道：

Best regards

Sili Liu

Si-li Liu

Re: Checkpoint failed because of TimeWindow cannot be cast to VoidNamespace

Thanks for your help

1. I started the job from scratch, not a savepoint or externalized checkpoint

2. No job graph change

3. env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

4. My Flink version is 1.9.1

Khachatryan Roman <[hidden email]> 于2020年7月3日周五下午4:49写道：

I still wasn't able to reproduce the issue.

Can you also clarify:

- Are you starting the job from a savepoint or externalized checkpoint?

- If yes, was the job graph changed?

- What StreamTimeCharacteristic is set, if any?

- What exact version of Flink do you use?

Regards,
Roman

On Fri, Jul 3, 2020 at 6:38 AM Si-li Liu <[hidden email]> wrote:

Hi, Thanks for your help.

The checkpoint configuration is

checkpoint.intervalMS=300000
checkpoint.timeoutMS=300000

The error callstack is from JM's log, which happened in every cp. Currently I don't have a success cp yet.

Khachatryan Roman <[hidden email]> 于2020年7月3日周五上午3:50写道：

Hi,

Thanks for the details.

However, I was not able to reproduce the issue. I used parallelism levels 4, file system backend and tried different timings for checkpointing, windowing and source.

Do you encounter this problem deterministically, is it always 1st checkpoint?

What checkpointing interval do you use?

Regards,
Roman

On Thu, Jul 2, 2020 at 1:57 PM Si-li Liu <[hidden email]> wrote:

Hi, this is our production code so I have to modify it a little bit, such as variable name and function name. I think 3 classes I provide here is enough.

class JoinFunction extends
  ProcessWindowFunction[RawLog, OutputLog, String, TimeWindow]{

  var ueState: ValueState[RawLog] = _
  @transient var gZipThriftSerializer: GZipThriftSerializer[MyType] = _
  val invalidCounter = new LongCounter()
  val processCounter = new LongCounter()
  val sendToKafkaCounter = new LongCounter()

  override def open(parameters: Configuration): Unit = {
    ueState = getRuntimeContext.getState(
      new ValueStateDescriptor[RawLog](bidState, classOf[RawLog])
    )
    gZipThriftSerializer = new GZipThriftSerializer[MyType]()
    getRuntimeContext.addAccumulator("processCounter", this.processCounter)
    getRuntimeContext.addAccumulator("invalidCounter", this.invalidCounter)
    getRuntimeContext.addAccumulator("sendToKafkaCounter", this.sendToKafkaCounter)
  }

  override def process(key: String,
                       ctx: Context,
                       logs: Iterable[RawLog],
                       out: Collector[OutputLog]): Unit = {
    if (ueState.value() != null) {
      processCounter.add(1L)
      val bid = ueState.value()
      val bidLog = gZipThriftSerializer.decompressAndDeserialize(bid.payload, classOf[MyType])
      logs.foreach( log => {
        if (log.eventType == SHOW) {
          val showLog = gZipThriftSerializer.decompressAndDeserialize(log.payload, classOf[MyType])
          sendToKafkaCounter.add(1L)
          out.collect(new OutputLog(ThriftUtils.serialize(showLog), Utils.getOutputTopic(showLog)))
        }
      })
    } else {
      invalidCounter.add(1L)
    }
  }
}

class JoinTrigger extends Trigger[RawLog, TimeWindow] {

  override def onElement(log: RawLog,
                         timestamp: Long,
                         window: TimeWindow,
                         ctx: Trigger.TriggerContext): TriggerResult = {
    val ueState: ValueState[RawLog] = ctx.getPartitionedState(
      new ValueStateDescriptor[RawLog](bidState, classOf[RawLog])
    )
    val firstSeen: ValueState[Boolean] = ctx.getPartitionedState(
      new ValueStateDescriptor[Boolean](initState, classOf[Boolean]))

    if (!firstSeen.value()) {
      ctx.registerEventTimeTimer(window.getEnd)
      firstSeen.update(true)
    }
    val eventType = log.eventType
    if (eventType == BID) {
      ueState.update(log)
      TriggerResult.CONTINUE
    } else {
      if (ueState.value() == null) {
        TriggerResult.CONTINUE
      } else {
        TriggerResult.FIRE
      }
    }
  }

  override def onEventTime(timestamp: Long,
                           window: TimeWindow,
                           ctx: Trigger.TriggerContext): TriggerResult = {
    if (timestamp == window.getEnd) {
      TriggerResult.PURGE
    }
    TriggerResult.CONTINUE
  }

  override def onProcessingTime(timestamp: Long,
                                window: TimeWindow,
                                ctx: Trigger.TriggerContext): TriggerResult = {
    TriggerResult.CONTINUE
  }

  override def clear(window: TimeWindow,
                     ctx: Trigger.TriggerContext): Unit = {
    val ueState: ValueState[RawLog] = ctx.getPartitionedState(
      new ValueStateDescriptor[RawLog](bidState, classOf[RawLog])
    )
    ueState.clear()

    val firstSeen: ValueState[Boolean] = ctx.getPartitionedState(
      new ValueStateDescriptor[Boolean](initState, classOf[Boolean]))
    firstSeen.clear()

    ctx.deleteEventTimeTimer(window.getEnd)
  }
}

class JoinEvictor extends Evictor[RawLog, TimeWindow] {

  override def evictBefore(elements: JIterable[TimestampedValue[RawLog]],
                           size: Int,
                           window: TimeWindow,
                           evictorContext: Evictor.EvictorContext): Unit = {}

  override def evictAfter(elements: JIterable[TimestampedValue[RawLog]],
                           size: Int,
                           window: TimeWindow,
                           evictorContext: Evictor.EvictorContext): Unit = {
    val iter = elements.iterator()
    while (iter.hasNext) {
      iter.next()
      iter.remove()
    }
  }
}

Khachatryan Roman <[hidden email]> 于2020年7月2日周四下午7:18写道：

Best regards

Sili Liu

Congxian Qiu

Re: Checkpoint failed because of TimeWindow cannot be cast to VoidNamespace

First, Could you please try this problem still there if use flink 1.10 or 1.11?

It seems strange, from the error message, here is an error when trying to convert a non-Window state(VoidNameSpace) to a Window State (serializer is the serializer of Window state, but the state is non-Window state).

Could you please try to replace the MyFuction with a reduce/aggregate/fold/apply() function to see what happens? -- this wants to narrow down the problem.

Best,

Congxian

Si-li Liu <[hidden email]> 于2020年7月3日周五下午6:44写道：

Thanks for your help

1. I started the job from scratch, not a savepoint or externalized checkpoint

2. No job graph change

3. env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

4. My Flink version is 1.9.1

Khachatryan Roman <[hidden email]> 于2020年7月3日周五下午4:49写道：

I still wasn't able to reproduce the issue.

Can you also clarify:

- Are you starting the job from a savepoint or externalized checkpoint?

- If yes, was the job graph changed?

- What StreamTimeCharacteristic is set, if any?

- What exact version of Flink do you use?

Regards,
Roman

On Fri, Jul 3, 2020 at 6:38 AM Si-li Liu <[hidden email]> wrote:

Hi, Thanks for your help.

The checkpoint configuration is

checkpoint.intervalMS=300000
checkpoint.timeoutMS=300000

The error callstack is from JM's log, which happened in every cp. Currently I don't have a success cp yet.

Khachatryan Roman <[hidden email]> 于2020年7月3日周五上午3:50写道：

Hi,

Thanks for the details.

However, I was not able to reproduce the issue. I used parallelism levels 4, file system backend and tried different timings for checkpointing, windowing and source.

Do you encounter this problem deterministically, is it always 1st checkpoint?

What checkpointing interval do you use?

Regards,
Roman

On Thu, Jul 2, 2020 at 1:57 PM Si-li Liu <[hidden email]> wrote:

Hi, this is our production code so I have to modify it a little bit, such as variable name and function name. I think 3 classes I provide here is enough.

class JoinFunction extends
  ProcessWindowFunction[RawLog, OutputLog, String, TimeWindow]{

  var ueState: ValueState[RawLog] = _
  @transient var gZipThriftSerializer: GZipThriftSerializer[MyType] = _
  val invalidCounter = new LongCounter()
  val processCounter = new LongCounter()
  val sendToKafkaCounter = new LongCounter()

  override def open(parameters: Configuration): Unit = {
    ueState = getRuntimeContext.getState(
      new ValueStateDescriptor[RawLog](bidState, classOf[RawLog])
    )
    gZipThriftSerializer = new GZipThriftSerializer[MyType]()
    getRuntimeContext.addAccumulator("processCounter", this.processCounter)
    getRuntimeContext.addAccumulator("invalidCounter", this.invalidCounter)
    getRuntimeContext.addAccumulator("sendToKafkaCounter", this.sendToKafkaCounter)
  }

  override def process(key: String,
                       ctx: Context,
                       logs: Iterable[RawLog],
                       out: Collector[OutputLog]): Unit = {
    if (ueState.value() != null) {
      processCounter.add(1L)
      val bid = ueState.value()
      val bidLog = gZipThriftSerializer.decompressAndDeserialize(bid.payload, classOf[MyType])
      logs.foreach( log => {
        if (log.eventType == SHOW) {
          val showLog = gZipThriftSerializer.decompressAndDeserialize(log.payload, classOf[MyType])
          sendToKafkaCounter.add(1L)
          out.collect(new OutputLog(ThriftUtils.serialize(showLog), Utils.getOutputTopic(showLog)))
        }
      })
    } else {
      invalidCounter.add(1L)
    }
  }
}

class JoinTrigger extends Trigger[RawLog, TimeWindow] {

  override def onElement(log: RawLog,
                         timestamp: Long,
                         window: TimeWindow,
                         ctx: Trigger.TriggerContext): TriggerResult = {
    val ueState: ValueState[RawLog] = ctx.getPartitionedState(
      new ValueStateDescriptor[RawLog](bidState, classOf[RawLog])
    )
    val firstSeen: ValueState[Boolean] = ctx.getPartitionedState(
      new ValueStateDescriptor[Boolean](initState, classOf[Boolean]))

    if (!firstSeen.value()) {
      ctx.registerEventTimeTimer(window.getEnd)
      firstSeen.update(true)
    }
    val eventType = log.eventType
    if (eventType == BID) {
      ueState.update(log)
      TriggerResult.CONTINUE
    } else {
      if (ueState.value() == null) {
        TriggerResult.CONTINUE
      } else {
        TriggerResult.FIRE
      }
    }
  }

  override def onEventTime(timestamp: Long,
                           window: TimeWindow,
                           ctx: Trigger.TriggerContext): TriggerResult = {
    if (timestamp == window.getEnd) {
      TriggerResult.PURGE
    }
    TriggerResult.CONTINUE
  }

  override def onProcessingTime(timestamp: Long,
                                window: TimeWindow,
                                ctx: Trigger.TriggerContext): TriggerResult = {
    TriggerResult.CONTINUE
  }

  override def clear(window: TimeWindow,
                     ctx: Trigger.TriggerContext): Unit = {
    val ueState: ValueState[RawLog] = ctx.getPartitionedState(
      new ValueStateDescriptor[RawLog](bidState, classOf[RawLog])
    )
    ueState.clear()

    val firstSeen: ValueState[Boolean] = ctx.getPartitionedState(
      new ValueStateDescriptor[Boolean](initState, classOf[Boolean]))
    firstSeen.clear()

    ctx.deleteEventTimeTimer(window.getEnd)
  }
}

class JoinEvictor extends Evictor[RawLog, TimeWindow] {

  override def evictBefore(elements: JIterable[TimestampedValue[RawLog]],
                           size: Int,
                           window: TimeWindow,
                           evictorContext: Evictor.EvictorContext): Unit = {}

  override def evictAfter(elements: JIterable[TimestampedValue[RawLog]],
                           size: Int,
                           window: TimeWindow,
                           evictorContext: Evictor.EvictorContext): Unit = {
    val iter = elements.iterator()
    while (iter.hasNext) {
      iter.next()
      iter.remove()
    }
  }
}

Khachatryan Roman <[hidden email]> 于2020年7月2日周四下午7:18写道：

Best regards

Sili Liu

Si-li Liu

Re: Checkpoint failed because of TimeWindow cannot be cast to VoidNamespace

Sorry

I can't reproduce it with reduce/aggregate/fold/apply and due to some limitations in my working environment, I can't use flink 1.10 or 1.11.

Congxian Qiu <[hidden email]> 于2020年7月5日周日下午6:21写道：

First, Could you please try this problem still there if use flink 1.10 or 1.11?

Could you please try to replace the MyFuction with a reduce/aggregate/fold/apply() function to see what happens? -- this wants to narrow down the problem.

Best,

Congxian

Si-li Liu <[hidden email]> 于2020年7月3日周五下午6:44写道：

Thanks for your help

1. I started the job from scratch, not a savepoint or externalized checkpoint

2. No job graph change

3. env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

4. My Flink version is 1.9.1

Khachatryan Roman <[hidden email]> 于2020年7月3日周五下午4:49写道：

I still wasn't able to reproduce the issue.

Can you also clarify:

- Are you starting the job from a savepoint or externalized checkpoint?

- If yes, was the job graph changed?

- What StreamTimeCharacteristic is set, if any?

- What exact version of Flink do you use?

Regards,
Roman

On Fri, Jul 3, 2020 at 6:38 AM Si-li Liu <[hidden email]> wrote:

Hi, Thanks for your help.

The checkpoint configuration is

checkpoint.intervalMS=300000
checkpoint.timeoutMS=300000

The error callstack is from JM's log, which happened in every cp. Currently I don't have a success cp yet.

Khachatryan Roman <[hidden email]> 于2020年7月3日周五上午3:50写道：

Hi,

Thanks for the details.

However, I was not able to reproduce the issue. I used parallelism levels 4, file system backend and tried different timings for checkpointing, windowing and source.

Do you encounter this problem deterministically, is it always 1st checkpoint?

What checkpointing interval do you use?

Regards,
Roman

On Thu, Jul 2, 2020 at 1:57 PM Si-li Liu <[hidden email]> wrote:

Hi, this is our production code so I have to modify it a little bit, such as variable name and function name. I think 3 classes I provide here is enough.

class JoinFunction extends
  ProcessWindowFunction[RawLog, OutputLog, String, TimeWindow]{

  var ueState: ValueState[RawLog] = _
  @transient var gZipThriftSerializer: GZipThriftSerializer[MyType] = _
  val invalidCounter = new LongCounter()
  val processCounter = new LongCounter()
  val sendToKafkaCounter = new LongCounter()

  override def open(parameters: Configuration): Unit = {
    ueState = getRuntimeContext.getState(
      new ValueStateDescriptor[RawLog](bidState, classOf[RawLog])
    )
    gZipThriftSerializer = new GZipThriftSerializer[MyType]()
    getRuntimeContext.addAccumulator("processCounter", this.processCounter)
    getRuntimeContext.addAccumulator("invalidCounter", this.invalidCounter)
    getRuntimeContext.addAccumulator("sendToKafkaCounter", this.sendToKafkaCounter)
  }

  override def process(key: String,
                       ctx: Context,
                       logs: Iterable[RawLog],
                       out: Collector[OutputLog]): Unit = {
    if (ueState.value() != null) {
      processCounter.add(1L)
      val bid = ueState.value()
      val bidLog = gZipThriftSerializer.decompressAndDeserialize(bid.payload, classOf[MyType])
      logs.foreach( log => {
        if (log.eventType == SHOW) {
          val showLog = gZipThriftSerializer.decompressAndDeserialize(log.payload, classOf[MyType])
          sendToKafkaCounter.add(1L)
          out.collect(new OutputLog(ThriftUtils.serialize(showLog), Utils.getOutputTopic(showLog)))
        }
      })
    } else {
      invalidCounter.add(1L)
    }
  }
}

class JoinTrigger extends Trigger[RawLog, TimeWindow] {

  override def onElement(log: RawLog,
                         timestamp: Long,
                         window: TimeWindow,
                         ctx: Trigger.TriggerContext): TriggerResult = {
    val ueState: ValueState[RawLog] = ctx.getPartitionedState(
      new ValueStateDescriptor[RawLog](bidState, classOf[RawLog])
    )
    val firstSeen: ValueState[Boolean] = ctx.getPartitionedState(
      new ValueStateDescriptor[Boolean](initState, classOf[Boolean]))

    if (!firstSeen.value()) {
      ctx.registerEventTimeTimer(window.getEnd)
      firstSeen.update(true)
    }
    val eventType = log.eventType
    if (eventType == BID) {
      ueState.update(log)
      TriggerResult.CONTINUE
    } else {
      if (ueState.value() == null) {
        TriggerResult.CONTINUE
      } else {
        TriggerResult.FIRE
      }
    }
  }

  override def onEventTime(timestamp: Long,
                           window: TimeWindow,
                           ctx: Trigger.TriggerContext): TriggerResult = {
    if (timestamp == window.getEnd) {
      TriggerResult.PURGE
    }
    TriggerResult.CONTINUE
  }

  override def onProcessingTime(timestamp: Long,
                                window: TimeWindow,
                                ctx: Trigger.TriggerContext): TriggerResult = {
    TriggerResult.CONTINUE
  }

  override def clear(window: TimeWindow,
                     ctx: Trigger.TriggerContext): Unit = {
    val ueState: ValueState[RawLog] = ctx.getPartitionedState(
      new ValueStateDescriptor[RawLog](bidState, classOf[RawLog])
    )
    ueState.clear()

    val firstSeen: ValueState[Boolean] = ctx.getPartitionedState(
      new ValueStateDescriptor[Boolean](initState, classOf[Boolean]))
    firstSeen.clear()

    ctx.deleteEventTimeTimer(window.getEnd)
  }
}

class JoinEvictor extends Evictor[RawLog, TimeWindow] {

  override def evictBefore(elements: JIterable[TimestampedValue[RawLog]],
                           size: Int,
                           window: TimeWindow,
                           evictorContext: Evictor.EvictorContext): Unit = {}

  override def evictAfter(elements: JIterable[TimestampedValue[RawLog]],
                           size: Int,
                           window: TimeWindow,
                           evictorContext: Evictor.EvictorContext): Unit = {
    val iter = elements.iterator()
    while (iter.hasNext) {
      iter.next()
      iter.remove()
    }
  }
}

Khachatryan Roman <[hidden email]> 于2020年7月2日周四下午7:18写道：

Best regards

Sili Liu

Congxian Qiu

Re: Checkpoint failed because of TimeWindow cannot be cast to VoidNamespace

Hi Si-li

Thanks for the notice.

I just want to double-check is the original problem has been solved? As I found that the created issue FLINK-18464 has been closed with reason "can not reproduce". Am I missing something here?

Best,

Congxian

Si-li Liu <[hidden email]> 于2020年7月10日周五下午6:06写道：

Si-li Liu

Re: Checkpoint failed because of TimeWindow cannot be cast to VoidNamespace

Someone told me that maybe this issue is Mesos specific. I'm kind of a newbie in Flink, and I digged into the code but can not get a conclusion. Here I just wanna have a better JoinWindow that emits the result and delete it from the window state immediately when joined successfully, is there any other way? Thanks!

Congxian Qiu <[hidden email]> 于2020年7月11日周六下午3:20写道：

Best regards

Sili Liu