je-ik commented on pull request #12551: URL: https://github.com/apache/beam/pull/12551#issuecomment-673427301
@mxm I added this check: https://github.com/apache/beam/pull/12551/commits/8ea092afefe8e3602d73c591e92a7ff5250684f5 and it resulted in the following trace: ``` java.lang.RuntimeException: Failed to set timer at org.apache.beam.runners.flink.translation.wrappers.streaming.DoFnOperator$FlinkTimerInternals.setTimer(DoFnOperator.java:1398) at org.apache.beam.runners.flink.translation.wrappers.streaming.DoFnOperator$FlinkTimerInternals.setTimer(DoFnOperator.java:1373) at org.apache.beam.runners.core.StatefulDoFnRunner.setupFlushTimer(StatefulDoFnRunner.java:288) at org.apache.beam.runners.core.StatefulDoFnRunner.processElementOrdered(StatefulDoFnRunner.java:180) at org.apache.beam.runners.core.StatefulDoFnRunner.processElement(StatefulDoFnRunner.java:147) at org.apache.beam.runners.flink.metrics.DoFnRunnerWithMetricsUpdate.processElement(DoFnRunnerWithMetricsUpdate.java:62) at org.apache.beam.runners.flink.translation.wrappers.streaming.DoFnOperator.processElement(DoFnOperator.java:604) at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask$StreamTaskNetworkOutput.emitRecord(OneInputStreamTask.java:173) at org.apache.flink.streaming.runtime.io.StreamTaskNetworkInput.processElement(StreamTaskNetworkInput.java:151) at org.apache.flink.streaming.runtime.io.StreamTaskNetworkInput.emitNext(StreamTaskNetworkInput.java:128) at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:69) at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:310) at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:187) at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:485) at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:469) at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:708) at org.apache.flink.runtime.taskmanager.Task.run(Task.java:533) at java.lang.Thread.run(Thread.java:748) Caused by: java.lang.IllegalStateException at org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkState(Preconditions.java:491) at org.apache.beam.runners.flink.translation.wrappers.streaming.DoFnOperator$FlinkTimerInternals.lambda$onNewEventTimer$0(DoFnOperator.java:1287) at java.util.Map.compute(Map.java:1093) at org.apache.beam.runners.flink.translation.wrappers.streaming.DoFnOperator$FlinkTimerInternals.onNewEventTimer(DoFnOperator.java:1283) at org.apache.beam.runners.flink.translation.wrappers.streaming.DoFnOperator$FlinkTimerInternals.registerTimer(DoFnOperator.java:1408) at org.apache.beam.runners.flink.translation.wrappers.streaming.DoFnOperator$FlinkTimerInternals.setTimer(DoFnOperator.java:1396) ... 17 more ``` That was to proove that there really is already set output timestamp of one specific timer, that is added multiple times. From the code I still do not see how this can happen. Yes, the timer is set from StatefulDoFnRunner (due to @RequiresTimeSortedInput), but provided that the processing is singlethreaded, I'm still a little lost. And it seems that the same can happen to ordinary user-code timers. ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: [email protected]
