Mailing-List: contact user-help@flink.apache.org; run by ezmlm
Precedence: bulk
From: Fritz Budiyanto <fbudiyan@icloud.com>
Content-type: text/plain; charset=us-ascii
Content-transfer-encoding: quoted-printable
MIME-version: 1.0 (Mac OS X Mail 10.3 \(3273\))
Subject: Re: Parallelism, registerEventTimeTimer and watermark problem
Date: Tue, 17 Oct 2017 20:23:25 -0700
References: <6B7E471D-6AF3-4DAF-8F54-557BBA754E5B@icloud.com>
To: user@flink.apache.org
In-reply-to: <6B7E471D-6AF3-4DAF-8F54-557BBA754E5B@icloud.com>
Message-id: <67047C59-4D8A-4E39-9345-1D63AD924A34@icloud.com>
archived-at: Wed, 18 Oct 2017 03:23:37 -0000

Sorry, missing copy paste for the exception thrown:

10/17/2017 20:21:30	dropDetection -> (aggFlowDropDetectPrintln -> =
Sink: Unnamed, aggFlowDropDetectPrintln -> Sink: Unnamed, Sink: =
kafkaSink)(3/4) switched to CANCELED=20
20:21:30,244 INFO  =
org.apache.flink.runtime.executiongraph.ExecutionGraph        - Job =
Aggregate flows (313a46d5fd23e4c2d0d00d0033950b6d) switched from state =
FAILING to FAILED.
java.lang.NullPointerException: Keyed state can only be used on a 'keyed =
stream', i.e., after a 'keyBy()' operation.
	at =
org.apache.flink.util.Preconditions.checkNotNull(Preconditions.java:75)
	at =
org.apache.flink.streaming.api.operators.StreamingRuntimeContext.checkPrec=
onditionsAndGetKeyedStateStore(StreamingRuntimeContext.java:151)
	at =
org.apache.flink.streaming.api.operators.StreamingRuntimeContext.getState(=
StreamingRuntimeContext.java:115)
	at =
FlowContractStitcherProcess.endState$lzycompute(FlowContractResolver.scala=
:30)
	at =
FlowContractStitcherProcess.endState(FlowContractResolver.scala:30)
	at =
FlowContractStitcherProcess.processElement(FlowContractResolver.scala:96)
	at =
FlowContractStitcherProcess.processElement(FlowContractResolver.scala:17)
	at =
org.apache.flink.streaming.api.operators.ProcessOperator.processElement(Pr=
ocessOperator.java:66)
	at =
org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutp=
ut.pushToOperator(OperatorChain.java:528)
	at =
org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutp=
ut.collect(OperatorChain.java:503)
	at =
org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutp=
ut.collect(OperatorChain.java:483)
	at =
org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOu=
tput.collect(AbstractStreamOperator.java:891)
	at =
org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOu=
tput.collect(AbstractStreamOperator.java:869)
	at =
org.apache.flink.streaming.runtime.operators.TimestampsAndPeriodicWatermar=
ksOperator.processElement(TimestampsAndPeriodicWatermarksOperator.java:68)=

	at =
org.apache.flink.streaming.runtime.io.StreamInputProcessor.processInput(St=
reamInputProcessor.java:206)
	at =
org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.run(OneInputSt=
reamTask.java:69)
	at =
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java=
:263)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:702)
	at java.lang.Thread.run(Thread.java:745)


--
Fritz


> On Oct 17, 2017, at 7:55 PM, Fritz Budiyanto <fbudiyan@icloud.com> =
wrote:
>=20
> Hi All,
>=20
> If I have high parallelism and use processFunction to =
registerEventTimeTimer, the timer never gets fired.
> After debugging, I found out the watermark isn't updated because I =
have keyBy right after assignTimestampsAndWatermarks.
> And if I set assignTimestampsAndWatermarks right after the keyBy, an =
exception is thrown.
>=20
> val contractFlow =3D enrichedFlow
>      .keyBy(f =3D> f.fiveTupleKey)
>      .assignTimestampsAndWatermarks(new AggFlowTimestampAssigner) =
<<<<<
>      .process(new FlowContractStitcherProcess)
>      .name("contractStitcher")
>=20
> at FlowContractStitcherProcess.endState(FlowContractResolver.scala:30)
> 	at =
FlowContractStitcherProcess.processElement(FlowContractResolver.scala:96)
> 	at =
FlowContractStitcherProcess.processElement(FlowContractResolver.scala:17)
> 	at =
org.apache.flink.streaming.api.operators.ProcessOperator.processElement(Pr=
ocessOperator.java:66)
> 	at =
org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutp=
ut.pushToOperator(OperatorChain.java:528)
> 	at =
org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutp=
ut.collect(OperatorChain.java:503)
> 	at =
org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutp=
ut.collect(OperatorChain.java:483)
> 	at =
org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOu=
tput.collect(AbstractStreamOperator.java:891)
> 	at =
org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOu=
tput.collect(AbstractStreamOperator.java:869)
> 	at =
org.apache.flink.streaming.runtime.operators.TimestampsAndPeriodicWatermar=
ksOperator.processElement(TimestampsAndPeriodicWatermarksOperator.java:68)=

> 	at =
org.apache.flink.streaming.runtime.io.StreamInputProcessor.processInput(St=
reamInputProcessor.java:206)
> 	at =
org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.run(OneInputSt=
reamTask.java:69)
> 	at =
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java=
:263)
> 	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:702)
> 	at java.lang.Thread.run(Thread.java:745)
>=20
>=20
> Any idea how to solve my problem ? How do I update the watermark after =
keyBy ?
>=20
> Would I hit scaling issue if on large number of timer if I use =
registerProcessingTimeTimer instead ? I'm using event time throughout =
the pipeline, would mixing processing timer with event time might cause =
problem down the line ?
>=20
> --
> Fritz