Mailing-List: contact user-help@flink.apache.org; run by ezmlm
Precedence: bulk
Reply-To: user@flink.apache.org
MIME-Version: 1.0
In-Reply-To: <D29583AD.554A%andrew.griess@sap.com>
References: <D29583AD.554A%andrew.griess@sap.com>
Date: Wed, 16 Dec 2015 11:06:07 +0100
Message-ID: 
 <CAC27z=Ms9QPzp3ffFSWDvfZBQ9sTfdVA5K_hsqm1Ebany0Zt4g@mail.gmail.com>
Subject: Re: global watermark across multiple kafka consumers
From: Till Rohrmann <trohrmann@apache.org>
To: user@flink.apache.org
Content-Type: multipart/alternative; boundary=001a11435c0404115c0527010e7b

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

Hi Andrew,

as far as I know, there is nothing such as a prescribed way of handling
this kind of situation. If you want to synchronize the watermark generation
given a set of KafkaConsumers you need some kind of ground truth.

This could be, for example, a central registry such as ZooKeeper in which
you collect the current watermarks of the different consumers. You could
access ZooKeeper from inside the TimestampExtractor.

Alternatively, however a bit more hacky, you could exploit that the
consumer tasks are usually colocated with consumer tasks from different
topics. This means that you'll have multiple subtasks reading from the
different Kafka topics running in the same JVM. You could then use class
variables to synchronize the watermarks. But this assumes that each subtask
reading the topic t from Kafka is colocated with at least one other subtask
reading the topic t' from Kafka with t' in T \ {t} and T being the set of
Kafka topics. Per default this should be the case.

I'm wondering why you need a global watermark for you Kafka topics. Isn't
it enough that you have individual watermarks for each topic?

Cheers,
Till

On Tue, Dec 15, 2015 at 4:45 PM, Griess, Andrew <andrew.griess@sap.com>
wrote:

> Hi guys,
>
> I have a question related to utilizing watermarks with multiple
> FlinkKakfkaConsumer082 instances. The aim is to have a global watermark
> across multiple kafka consumers where any message from any kafka partition
> would update the same watermark. When testing a simple TimeStampExtractor
> implementation it seems each consumer results in a separate watermark. Is
> there a prescribed way of handling such a thing that anyone has any
> experience with?
>
> Thanks for your help,
>
> Andrew Griess
>
>

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

<div dir=3D"ltr">Hi Andrew,<div><br></div><div>as far as I know, there is n=
othing such as a prescribed way of handling this kind of situation. If you =
want to synchronize the watermark generation given a set of KafkaConsumers =
you need some kind of ground truth.=C2=A0</div><div><br></div><div>This cou=
ld be, for example, a central registry such as ZooKeeper in which you colle=
ct the current watermarks of the different consumers. You could access ZooK=
eeper from inside the TimestampExtractor.</div><div><br></div><div>Alternat=
ively, however a bit more hacky, you could exploit that the consumer tasks =
are usually colocated with consumer tasks from different topics. This means=
 that you&#39;ll have multiple subtasks reading from the different Kafka to=
pics running in the same JVM. You could then use class variables to synchro=
nize the watermarks. But this assumes that each subtask reading the topic t=
 from Kafka is colocated with at least one other subtask reading the topic =
t&#39; from Kafka with t&#39; in T \ {t} and T being the set of Kafka topic=
s. Per default this should be the case.</div><div><br></div><div>I&#39;m wo=
ndering why you need a global watermark for you Kafka topics. Isn&#39;t it =
enough that you have individual watermarks for each topic?</div><div><br></=
div><div>Cheers,</div><div>Till</div></div><div class=3D"gmail_extra"><br><=
div class=3D"gmail_quote">On Tue, Dec 15, 2015 at 4:45 PM, Griess, Andrew <=
span dir=3D"ltr">&lt;<a href=3D"mailto:andrew.griess@sap.com" target=3D"_bl=
ank">andrew.griess@sap.com</a>&gt;</span> wrote:<br><blockquote class=3D"gm=
ail_quote" style=3D"margin:0 0 0 .8ex;border-left:1px #ccc solid;padding-le=
ft:1ex">


<div style=3D"word-wrap:break-word;color:rgb(0,0,0);font-size:14px;font-fam=
ily:Calibri,sans-serif">
<div>
<div>Hi guys,</div>
<div><br>
</div>
<div>I have a question related to utilizing watermarks with multiple FlinkK=
akfkaConsumer082 instances. The aim is to have a global watermark across mu=
ltiple kafka consumers where any message from any kafka partition would upd=
ate the same watermark. When testing
 a simple TimeStampExtractor implementation it seems each consumer results =
in a separate watermark. Is there a prescribed way of handling such a thing=
 that anyone has any experience with?</div>
<div><br>
</div>
<div>Thanks for your help,</div>
<div><br>
</div>
<div>Andrew Griess</div>
</div>
<div><br>
</div>
</div>

</blockquote></div><br></div>

--001a11435c0404115c0527010e7b--