Mailing-List: contact user-help@flink.apache.org; run by ezmlm
Precedence: bulk
Reply-To: user@flink.apache.org
MIME-Version: 1.0
Sender: ewenstephan@gmail.com
In-Reply-To: <561240F0.6040609@apache.org>
References: 
 <CACVCA=eH7jWuCkNj=Px5c_b+JNfOp5aGLDj4FnimEwPASzxL_A@mail.gmail.com>
	<561240F0.6040609@apache.org>
Date: Mon, 5 Oct 2015 11:32:37 +0200
Message-ID: 
 <CANC1h_vKxcfZM=0tQOg711ppCU=ju_LySNzCzM4Wyyrt_oBz7g@mail.gmail.com>
Subject: Re: Destroy StreamExecutionEnv
From: Stephan Ewen <sewen@apache.org>
To: user@flink.apache.org
Content-Type: multipart/alternative; boundary=001a1139b126a67d8a05215831fa

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

Matthias' solution should work in most cases.

In cases where you do not control the source (or the source can never be
finite, like the Kafka source), we often use a trick in the tests, which is
throwing a special type of exception (a SuccessException).

You can catch this exception on env.execute() (it is the nested cause) and
decide that this qualifies the test as successful...

Greetings,
Stephan


On Mon, Oct 5, 2015 at 11:20 AM, Matthias J. Sax <mjsax@apache.org> wrote:

> Hi,
>
> you just need to terminate your source (ie, return from run() method if
> you implement your own source function). This will finish the complete
> program. For already available sources, just make sure you read finite
> input.
>
> Hope this helps.
>
> -Matthias
>
> On 10/05/2015 12:15 AM, jay vyas wrote:
> > Hi folks.
> >
> > How do we end a stream execution environment?
> >
> > I have a unit test which runs a streaming job, and want the unit test to
> > die after the first round of output is processed...
> >
> >
> > DataStream<Tuple2<Map, Integer>> counts =
> >     dataStream.map(
> >         new MapFunction<String, Tuple2<Map, Integer>>() {
> >           @Override
> >           public Tuple2<Map, Integer> map(String s) throws Exception {
> >             Map transaction = MAPPER.readValue(s, Map.class);
> >             return new Tuple2<>(transaction, 1);
> >           }
> >         });
> > counts.print();
> >
> >
> >
> > --
> > jay vyas
>
>

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

<div dir=3D"ltr">Matthias&#39; solution should work in most cases.<div><br>=
</div><div>In cases where you do not control the source (or the source can =
never be finite, like the Kafka source), we often use a trick in the tests,=
 which is throwing a special type of exception (a SuccessException).</div><=
div><br></div><div>You can catch this exception on env.execute() (it is the=
 nested cause) and decide that this qualifies the test as successful...</di=
v><div><br></div><div>Greetings,<br>Stephan</div><div><br></div></div><div =
class=3D"gmail_extra"><br><div class=3D"gmail_quote">On Mon, Oct 5, 2015 at=
 11:20 AM, Matthias J. Sax <span dir=3D"ltr">&lt;<a href=3D"mailto:mjsax@ap=
ache.org" target=3D"_blank">mjsax@apache.org</a>&gt;</span> wrote:<br><bloc=
kquote class=3D"gmail_quote" style=3D"margin:0 0 0 .8ex;border-left:1px #cc=
c solid;padding-left:1ex">Hi,<br>
<br>
you just need to terminate your source (ie, return from run() method if<br>
you implement your own source function). This will finish the complete<br>
program. For already available sources, just make sure you read finite<br>
input.<br>
<br>
Hope this helps.<br>
<span class=3D"HOEnZb"><font color=3D"#888888"><br>
-Matthias<br>
</font></span><div class=3D"HOEnZb"><div class=3D"h5"><br>
On 10/05/2015 12:15 AM, jay vyas wrote:<br>
&gt; Hi folks.<br>
&gt;<br>
&gt; How do we end a stream execution environment?<br>
&gt;<br>
&gt; I have a unit test which runs a streaming job, and want the unit test =
to<br>
&gt; die after the first round of output is processed...<br>
&gt;<br>
&gt;<br>
&gt; DataStream&lt;Tuple2&lt;Map, Integer&gt;&gt; counts =3D<br>
&gt;=C2=A0 =C2=A0 =C2=A0dataStream.map(<br>
&gt;=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0new MapFunction&lt;String, Tuple2&lt;=
Map, Integer&gt;&gt;() {<br>
&gt;=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0@Override<br>
&gt;=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0public Tuple2&lt;Map, Integer&=
gt; map(String s) throws Exception {<br>
&gt;=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0Map transaction =3D MAP=
PER.readValue(s, Map.class);<br>
&gt;=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0return new Tuple2&lt;&g=
t;(transaction, 1);<br>
&gt;=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0}<br>
&gt;=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0});<br>
&gt; counts.print();<br>
&gt;<br>
&gt;<br>
&gt;<br>
&gt; --<br>
&gt; jay vyas<br>
<br>
</div></div></blockquote></div><br></div>

--001a1139b126a67d8a05215831fa--