[
https://issues.apache.org/jira/browse/BEAM-5909?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Maximilian Michels closed BEAM-5909.
------------------------------------
Resolution: Fixed
> QueryablePipeline fails to lookup UserState with Timers
> -------------------------------------------------------
>
> Key: BEAM-5909
> URL: https://issues.apache.org/jira/browse/BEAM-5909
> Project: Beam
> Issue Type: Bug
> Components: sdk-java-core
> Reporter: Maximilian Michels
> Assignee: Maximilian Michels
> Priority: Minor
> Fix For: 2.9.0
>
> Time Spent: 50m
> Remaining Estimate: 0h
>
> QueryablePipeline fails when given a ParDo with both state and timers,
> similar to the example in
> https://beam.apache.org/blog/2017/08/28/timely-processing.html#event-time-timers:
> {noformat}
> java.lang.IllegalArgumentException: expected one element but was:
> <org.apache.beam.sdk.values.PCollection.<init>:402#3d93cb799b3970be, expiry>
> at
> org.apache.beam.repackaged.beam_runners_core_construction_java.com.google.common.collect.Iterators.getOnlyElement(Iterators.java:322)
> at
> org.apache.beam.repackaged.beam_runners_core_construction_java.com.google.common.collect.Iterables.getOnlyElement(Iterables.java:294)
> at
> org.apache.beam.runners.core.construction.graph.QueryablePipeline.lambda$getUserStates$10(QueryablePipeline.java:362)
> at
> java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
> at java.util.Iterator.forEachRemaining(Iterator.java:116)
> at
> java.util.Spliterators$IteratorSpliterator.forEachRemaining(Spliterators.java:1801)
> at
> java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481)
> at
> java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471)
> at
> java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:708)
> at
> java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
> at
> java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:499)
> at
> org.apache.beam.runners.core.construction.graph.QueryablePipeline.getUserStates(QueryablePipeline.java:372)
> at
> org.apache.beam.runners.core.construction.graph.GreedyStageFuser.forGrpcPortRead(GreedyStageFuser.java:92)
> at
> org.apache.beam.runners.core.construction.graph.GreedyPipelineFuser.fuseSiblings(GreedyPipelineFuser.java:340)
> at
> org.apache.beam.runners.core.construction.graph.GreedyPipelineFuser.fusePipeline(GreedyPipelineFuser.java:142)
> at
> org.apache.beam.runners.core.construction.graph.GreedyPipelineFuser.<init>(GreedyPipelineFuser.java:75)
> at
> org.apache.beam.runners.core.construction.graph.GreedyPipelineFuser.fuse(GreedyPipelineFuser.java:88)
> at
> org.apache.beam.runners.flink.FlinkJobInvocation.runPipeline(FlinkJobInvocation.java:96)
> at
> org.apache.beam.repackaged.beam_runners_flink_2.11.com.google.common.util.concurrent.TrustedListenableFutureTask$TrustedFutureInterruptibleTask.runInterruptibly(TrustedListenableFutureTask.java:111)
> at
> org.apache.beam.repackaged.beam_runners_flink_2.11.com.google.common.util.concurrent.InterruptibleTask.run(InterruptibleTask.java:58)
> at
> org.apache.beam.repackaged.beam_runners_flink_2.11.com.google.common.util.concurrent.TrustedListenableFutureTask.run(TrustedListenableFutureTask.java:75)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
> at java.lang.Thread.run(Thread.java:745)
> {noformat}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
