Mailing-List: contact user-help@flink.apache.org; run by ezmlm
Precedence: bulk
Reply-To: user@flink.apache.org
MIME-Version: 1.0
Sender: niels@basj.es
Date: Mon, 30 Nov 2015 14:47:21 +0100
Message-ID: 
 <CADoiZqqT132XkvmLe1qH6XMZBRj1Zx7vMDyuqMYH3icaEJOZ5Q@mail.gmail.com>
Subject: Triggering events
From: Niels Basjes <Niels@basjes.nl>
To: user@flink.apache.org
Content-Type: multipart/alternative; boundary=001a114b2d40beeae90525c247b9

--001a114b2d40beeae90525c247b9
Content-Type: text/plain; charset=UTF-8

Hi,

I'm experimenting with a custom Windowing setup over clickstream data.
I want the timestamps of this clickstream data to be the timestamps 'when
the event occurred' and in the Windows I need to trigger on these times.

For testing I created a source roughly like this:

    public class ManualTimeEventSource extends
RichEventTimeSourceFunction<Long> {

                    ctx.collectWithTimestamp(event, event.timestamp);


But none of the triggers were called so I started digging through the code.

Then I figured I apparently needed to add the watermarks myself, so I
added a line:

                    ctx.emitWatermark(new Watermark(event.timestamp));


But now I get:


*Caused by: java.lang.ClassCastException:
org.apache.flink.streaming.api.watermark.Watermark cannot be cast to
org.apache.flink.streaming.runtime.streamrecord.StreamRecord*
* at
org.apache.flink.streaming.runtime.streamrecord.StreamRecordSerializer.serialize(StreamRecordSerializer.java:41)*
* at
org.apache.flink.runtime.plugable.SerializationDelegate.write(SerializationDelegate.java:56)*
* at
org.apache.flink.runtime.io.network.api.serialization.SpanningRecordSerializer.addRecord(SpanningRecordSerializer.java:79)*
* at
org.apache.flink.runtime.io.network.api.writer.RecordWriter.broadcastEmit(RecordWriter.java:109)*
* at
org.apache.flink.streaming.runtime.io.StreamRecordWriter.broadcastEmit(StreamRecordWriter.java:93)*
* at
org.apache.flink.streaming.runtime.io.RecordWriterOutput.emitWatermark(RecordWriterOutput.java:88)*
* ... 9 more*


This seems like a bug to me (StreamElement vs StreamRecord). Is it a bug in
Flink or in my code?

What is the right way to trigger the events in my Windowing setup?


P.S. I'm binding my Java application against Flink version 0.10.1

-- 
Best regards / Met vriendelijke groeten,

Niels Basjes

--001a114b2d40beeae90525c247b9
Content-Type: text/html; charset=UTF-8
Content-Transfer-Encoding: quoted-printable

<div dir=3D"ltr">Hi,<div><br></div><div>I&#39;m experimenting with a custom=
 Windowing setup over clickstream data.</div><div>I want the timestamps of =
this clickstream data to be the timestamps &#39;when the event occurred&#39=
; and in the Windows I need to trigger on these times.</div><div><br></div>=
<div>For testing I created a source roughly like this:</div><div><pre style=
=3D"color:rgb(0,0,0);font-family:&#39;DejaVu Sans Mono&#39;;font-size:9pt">=
    <span style=3D"color:rgb(0,0,128);font-weight:bold">public class </span=
>ManualTimeEventSource <span style=3D"color:rgb(0,0,128);font-weight:bold">=
extends </span>RichEventTimeSourceFunction&lt;Long&gt; {</pre><pre style=3D=
"color:rgb(0,0,0);font-family:&#39;DejaVu Sans Mono&#39;;font-size:9pt">   =
                 ctx.collectWithTimestamp(event,=C2=A0<span style=3D"font-s=
ize:9pt">event.</span><span style=3D"font-size:9pt">timestamp);</span></pre=
><pre style=3D"color:rgb(0,0,0);font-family:&#39;DejaVu Sans Mono&#39;;font=
-size:9pt"><br></pre><pre style=3D"color:rgb(0,0,0);font-family:&#39;DejaVu=
 Sans Mono&#39;;font-size:9pt">But none of the triggers were called so I st=
arted digging through the code.</pre><pre style=3D"color:rgb(0,0,0);font-fa=
mily:&#39;DejaVu Sans Mono&#39;;font-size:9pt">Then I figured I apparently =
needed to add the watermarks myself, so I added a line:</pre><pre style=3D"=
color:rgb(0,0,0);font-family:&#39;DejaVu Sans Mono&#39;;font-size:9pt"><spa=
n style=3D"font-size:9pt">                    ctx.emitWatermark(</span><spa=
n style=3D"font-size:9pt;color:rgb(0,0,128);font-weight:bold">new </span><s=
pan style=3D"font-size:9pt">Watermark(event.</span><span style=3D"font-size=
:9pt">timestamp</span><span style=3D"font-size:9pt">));</span><br></pre></d=
iv><div><div><br></div><div>But now I get:</div></div><blockquote style=3D"=
margin:0px 0px 0px 40px;border:none;padding:0px"><div><div><div><i><br></i>=
</div></div></div><div><div><div><i>Caused by: java.lang.<b>ClassCastExcept=
ion</b>: org.apache.flink.streaming.api.watermark.<b>Watermark cannot be ca=
st to </b>org.apache.flink.streaming.runtime.streamrecord.<b>StreamRecord</=
b></i></div></div></div><div><div><div><i><span class=3D"" style=3D"white-s=
pace:pre">	</span>at org.apache.flink.streaming.runtime.streamrecord.Stream=
RecordSerializer.serialize(StreamRecordSerializer.java:41)</i></div></div><=
/div><div><div><div><i><span class=3D"" style=3D"white-space:pre">	</span>a=
t org.apache.flink.runtime.plugable.SerializationDelegate.write(Serializati=
onDelegate.java:56)</i></div></div></div><div><div><div><i><span class=3D""=
 style=3D"white-space:pre">	</span>at org.apache.flink.runtime.io.network.a=
pi.serialization.SpanningRecordSerializer.addRecord(SpanningRecordSerialize=
r.java:79)</i></div></div></div><div><div><div><i><span class=3D"" style=3D=
"white-space:pre">	</span>at org.apache.flink.runtime.io.network.api.writer=
.RecordWriter.broadcastEmit(RecordWriter.java:109)</i></div></div></div><di=
v><div><div><i><span class=3D"" style=3D"white-space:pre">	</span>at org.ap=
ache.flink.streaming.runtime.io.StreamRecordWriter.broadcastEmit(StreamReco=
rdWriter.java:93)</i></div></div></div><div><div><div><i><span class=3D"" s=
tyle=3D"white-space:pre">	</span>at org.apache.flink.streaming.runtime.io.R=
ecordWriterOutput.emitWatermark(RecordWriterOutput.java:88)</i></div></div>=
</div><div><div><div><i><span class=3D"" style=3D"white-space:pre">	</span>=
... 9 more</i></div></div></div></blockquote><div><div><br></div><div>This =
seems like a bug to me (StreamElement vs StreamRecord). Is it a bug in Flin=
k or in my code?</div><div><br></div><div>What is the right way to trigger =
the events in my Windowing setup?<br></div><div><br></div><div><br></div><d=
iv><br></div><div>P.S. I&#39;m binding my Java application against Flink ve=
rsion 0.10.1<br></div><div><br></div>-- <br><div class=3D"gmail_signature">=
Best regards / Met vriendelijke groeten,<br><br>Niels Basjes</div>
</div></div>

--001a114b2d40beeae90525c247b9--