confirmed that moving back to FlinkKafkaConsumer fixes things. is there some notification channel/medium that highlights critical bugs/issues on the intended features like this pretty readily?
On Fri, Apr 8, 2022 at 2:18 AM Jin Yi <j...@promoted.ai> wrote: > based on symptoms/observations on the first operator (LogRequestFilter) > watermark and event timestamps, it does seem like it's the bug. things > track fine (timestamp > watermark) for the first batch of events, then the > event timestamps go back into the past and are "late". > > looks like the 1.14 backport just got in 11 days ago ( > https://github.com/apache/flink/pull/19128). is there a way to easily > test this fix locally? based on the threads, should i just move back to > FlinkKafkaConsumer until 1.14.5? > > On Fri, Apr 8, 2022 at 1:34 AM Qingsheng Ren <renqs...@gmail.com> wrote: > >> Hi Jin, >> >> If you are using new FLIP-27 sources like KafkaSource, per-partition >> watermark (or per-split watermark) is a default feature integrated in >> SourceOperator. You might hit the bug described in FLINK-26018 [1], which >> happens during the first fetch of the source that records in the first >> split pushes the watermark far away, then records from other splits will be >> treated as late events. >> >> [1] https://issues.apache.org/jira/browse/FLINK-26018 >> >> Best regards, >> >> Qingsheng >> >> >> > On Apr 8, 2022, at 15:54, Jin Yi <j...@promoted.ai> wrote: >> > >> > how should the code look like to verify we're using per-partition >> watermarks if we moved away from FlinkKafkaConsumer to KafkaSource in >> 1.14.4? >> > >> > we currently have it looking like: >> > >> > streamExecutionEnvironment.fromSource( >> > KafkaSource.<T>builder().....build(), >> > watermarkStrategy, >> > "whatever", >> > typeInfo); >> > >> > when running this job with the streamExecutionEnviornment parallelism >> set to 1, and the kafka source having 30 partitions, i'm seeing weird >> behaviors where the first operator after this source consumes events out of >> order (and therefore, violates watermarks). the operator simply checks to >> see what "type" of event something is and uses side outputs to output the >> type-specific messages. here's a snippet of the event timestamp going back >> before the current watermark (first instance of going backwards in time in >> bold): >> > >> > 2022-04-08 05:47:06,315 WARN >> ai.promoted.metrics.logprocessor.common.functions.filter.LogRequestFilter >> [] - LogRequestFilter ts: 1649284267139 watermark: 1649284187140 >> > 2022-04-08 05:47:06,315 WARN >> ai.promoted.metrics.logprocessor.common.functions.filter.LogRequestFilter >> [] - LogRequestFilter ts: 1649284268138 watermark: 1649284187140 >> > 2022-04-08 05:47:06,315 WARN >> ai.promoted.metrics.logprocessor.common.functions.filter.LogRequestFilter >> [] - LogRequestFilter ts: 1649284269138 watermark: 1649284187140 >> > 2022-04-08 05:47:06,315 WARN >> ai.promoted.metrics.logprocessor.common.functions.filter.LogRequestFilter >> [] - LogRequestFilter ts: 1649284270139 watermark: 1649284187140 >> > 2022-04-08 05:47:06,315 WARN >> ai.promoted.metrics.logprocessor.common.functions.filter.LogRequestFilter >> [] - LogRequestFilter ts: 1649284271139 watermark: 1649284187140 >> > 2022-04-08 05:47:06,315 WARN >> ai.promoted.metrics.logprocessor.common.functions.filter.LogRequestFilter >> [] - LogRequestFilter ts: 1649284171037 watermark: 1649284187140 >> > 2022-04-08 05:47:06,316 WARN >> ai.promoted.metrics.logprocessor.common.functions.filter.LogRequestFilter >> [] - LogRequestFilter ts: 1649284172057 watermark: 1649284187140 >> > 2022-04-08 05:47:06,316 WARN >> ai.promoted.metrics.logprocessor.common.functions.filter.LogRequestFilter >> [] - LogRequestFilter ts: 1649284172067 watermark: 1649284187140 >> > 2022-04-08 05:47:06,316 WARN >> ai.promoted.metrics.logprocessor.common.functions.filter.LogRequestFilter >> [] - LogRequestFilter ts: 1649284172171 watermark: 1649284187140 >> > 2022-04-08 05:47:06,316 WARN >> ai.promoted.metrics.logprocessor.common.functions.filter.LogRequestFilter >> [] - LogRequestFilter ts: 1649284172174 watermark: 1649284187140 >> > 2022-04-08 05:47:06,317 WARN >> ai.promoted.metrics.logprocessor.common.functions.filter.LogRequestFilter >> [] - LogRequestFilter ts: 1649284172666 watermark: 1649284187140 >> > >> > >> > >> > On Sat, Mar 19, 2022 at 10:51 AM Dan Hill <quietgol...@gmail.com> >> wrote: >> > I dove deeper. I wasn't actually using per-partition watermarks. >> Thank you for the help! >> > >> > On Fri, Mar 18, 2022 at 12:11 PM Dan Hill <quietgol...@gmail.com> >> wrote: >> > Thanks, Thias and Dongwon. >> > >> > I'll keep debugging this with the idle watermark turned off. >> > >> > Next TODOs: >> > - Verify that we’re using per-partition watermarks. Our code matches >> the example but maybe something is disabling it. >> > - Enable logging of partition-consumer assignment, to see if that is >> the cause of the problem. >> > - Look at adding flags to set the source parallelism to see if that >> fixes the issue. >> > >> > Yes, I've seen Flink talks on creating our own watermarks through >> Kafka. Sounds like a good idea. >> > >> > On Fri, Mar 18, 2022 at 1:17 AM Dongwon Kim <eastcirc...@gmail.com> >> wrote: >> > I totally agree with Schwalbe that per-partition watermarking allows # >> source tasks < # kafka partitions. >> > >> > Otherwise, Dan, you should suspect other possibilities like what >> Schwalbe said. >> > >> > Best, >> > >> > Dongwon >> > >> > On Fri, Mar 18, 2022 at 5:01 PM Schwalbe Matthias < >> matthias.schwa...@viseca.ch> wrote: >> > Hi San, Dongwon, >> > >> > >> > >> > I share the opinion that when per-partition watermarking is enabled, >> you should observe correct behavior … would be interesting to see why it >> does not work for you. >> > >> > >> > >> > I’d like to clear one tiny misconception here when you write: >> > >> > >> > >> > >> - The same issue happens even if I use an idle watermark. >> > >> > >> > >> > You would expect to see glitches with watermarking when you enable >> idleness. >> > >> > Idleness sort of trades watermark correctness for reduces latency when >> processing timers (much simplified). >> > >> > With idleness enabled you have no guaranties whatsoever as to the >> quality of watermarks (which might be ok in some cases). >> > >> > BTW we dominantly use a mix of fast and slow sources (that only update >> once a day) which hand-pimped watermarking and late event processing, and >> enabling idleness would break everything. >> > >> > >> > >> > Oversight put aside things should work the way you implemented it. >> > >> > >> > >> > One thing I could imagine to be a cause is >> > >> > • that over time the kafka partitions get reassigned to >> different consumer subtasks which would probably stress correct >> recalculation of watermarks. Hence #partition == number subtask might >> reduce the problem >> > • can you enable logging of partition-consumer assignment, to see >> if that is the cause of the problem >> > • also involuntary restarts of the job can cause havoc as this >> resets watermarking >> > >> > >> > I’ll be off next week, unable to take part in the active discussion … >> > >> > >> > >> > Sincere greetings >> > >> > >> > >> > Thias >> > >> > >> > >> > >> > >> > >> > >> > >> > >> > From: Dan Hill <quietgol...@gmail.com> >> > Sent: Freitag, 18. März 2022 08:23 >> > To: Dongwon Kim <eastcirc...@gmail.com> >> > Cc: user <user@flink.apache.org> >> > Subject: Re: Weird Flink Kafka source watermark behavior >> > >> > >> > >> > ⚠EXTERNAL MESSAGE – CAUTION: Think Before You Click ⚠ >> > >> > >> > >> > I'll try forcing # source tasks = # partitions tomorrow. >> > >> > >> > >> > Thank you, Dongwon, for all of your help! >> > >> > >> > >> > On Fri, Mar 18, 2022 at 12:20 AM Dongwon Kim <eastcirc...@gmail.com> >> wrote: >> > >> > I believe your job with per-partition watermarking should be working >> okay even in a backfill scenario. >> > >> > >> > >> > BTW, is the problem still observed even with # sour tasks = # >> partitions? >> > >> > >> > >> > For committers: >> > >> > Is there a way to confirm that per-partition watermarking is used in TM >> log? >> > >> > >> > >> > On Fri, Mar 18, 2022 at 4:14 PM Dan Hill <quietgol...@gmail.com> wrote: >> > >> > I hit this using event processing and no idleness detection. The same >> issue happens if I enable idleness. >> > >> > >> > >> > My code matches the code example for per-partition watermarking. >> > >> > >> > >> > On Fri, Mar 18, 2022 at 12:07 AM Dongwon Kim <eastcirc...@gmail.com> >> wrote: >> > >> > Hi Dan, >> > >> > >> > >> > I'm quite confused as you already use per-partition watermarking. >> > >> > >> > >> > What I meant in the reply is >> > >> > - If you don't use per-partition watermarking, # tasks < # partitions >> can cause the problem for backfill jobs. >> > >> > - If you don't use per-partition watermarking, # tasks = # partitions >> is going to be okay even for backfill jobs. >> > >> > - If you use per-partition watermarking, # tasks < # partitions >> shouldn't cause any problems unless you turn on the idleness detection. >> > >> > >> > >> > Regarding the idleness detection which is based on processing time, >> what is your setting? If you set the value to 10 seconds for example, >> you'll face the same problem unless the watermark of your backfill job >> catches up real-time within 10 seconds. If you increase the value to 1 >> minute, your backfill job should catch up real-time within 1 minute. >> > >> > >> > >> > Best, >> > >> > >> > >> > Dongwon >> > >> > >> > >> > >> > >> > On Fri, Mar 18, 2022 at 3:51 PM Dan Hill <quietgol...@gmail.com> wrote: >> > >> > Thanks Dongwon! >> > >> > >> > >> > Wow. Yes, I'm using per-partition watermarking [1]. Yes, my # source >> tasks < # kafka partitions. This should be called out in the docs or the >> bug should be fixed. >> > >> > >> > >> > On Thu, Mar 17, 2022 at 10:54 PM Dongwon Kim <eastcirc...@gmail.com> >> wrote: >> > >> > Hi Dan, >> > >> > >> > >> > Do you use the per-partition watermarking explained in [1]? >> > >> > I've also experienced a similar problem when running backfill jobs >> specifically when # source tasks < # kafka partitions. >> > >> > - When # source tasks = # kafka partitions, the backfill job works as >> expected. >> > >> > - When # source tasks < # kafka partitions, a Kafka consumer consumes >> multiple partitions. This case can destroying the per-partition patterns as >> explained in [2]. >> > >> > >> > >> > Hope this helps. >> > >> > >> > >> > p.s. If you plan to use the per-partition watermarking, be aware that >> idleness detection [3] can cause another problem when you run a backfill >> job. Kafka source tasks in a backfill job seem to read a batch of records >> from Kafka and then wait for downstream tasks to catch up the progress, >> which can be counted as idleness. >> > >> > >> > >> > [1] >> https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/event-time/generating_watermarks/#using-watermark-strategie >> > >> > [2] >> https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/event-time/generating_watermarks/#watermark-strategies-and-the-kafka-connector >> > >> > [3] >> https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/event-time/generating_watermarks/#dealing-with-idle-sources >> > >> > >> > >> > Best, >> > >> > >> > >> > Dongwon >> > >> > >> > >> > On Fri, Mar 18, 2022 at 2:35 PM Dan Hill <quietgol...@gmail.com> wrote: >> > >> > I'm following the example from this section: >> > >> > >> https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/event-time/generating_watermarks/#watermark-strategies-and-the-kafka-connector >> > >> > >> > >> > On Thu, Mar 17, 2022 at 10:26 PM Dan Hill <quietgol...@gmail.com> >> wrote: >> > >> > Other points >> > >> > - I'm using the kafka timestamp as event time. >> > >> > - The same issue happens even if I use an idle watermark. >> > >> > >> > >> > On Thu, Mar 17, 2022 at 10:17 PM Dan Hill <quietgol...@gmail.com> >> wrote: >> > >> > There are 12 Kafka partitions (to keep the structure similar to other >> low traffic environments). >> > >> > >> > >> > On Thu, Mar 17, 2022 at 10:13 PM Dan Hill <quietgol...@gmail.com> >> wrote: >> > >> > Hi. >> > >> > >> > >> > I'm running a backfill from a kafka topic with very few records spread >> across a few days. I'm seeing a case where the records coming from a kafka >> source have a watermark that's more recent (by hours) than the event time. >> I haven't seen this before when running this. This violates what I'd >> assume the kafka source would do. >> > >> > >> > >> > Example problem: >> > >> > 1. I have kafka records at ts=1000, 2000, ... 500000. The actual times >> are separated by a longer time period. >> > >> > 2. My first operator after the FlinkKafkaConsumer sees: >> > >> > context.timestamp() = 1000 >> > >> > context.timerService().currentWatermark() = 500000 >> > >> > >> > >> > Details about how I'm running this: >> > >> > - I'm on Flink 1.12.3 that's running on EKS and using MSK as the source. >> > >> > - I'm using FlinkKafkaConsumer >> > >> > - I'm using WatermarkStrategy.forBoundedOutOfOrderness(5s). No >> idleness settings. >> > >> > - I'm running similar code in all the environments. The main >> difference is low traffic. I have not been able to reproduce this out of >> the environment. >> > >> > >> > >> > >> > >> > I put the following process function right after my kafka source. >> > >> > >> > >> > -------- >> > >> > >> > AfterSource >> > >> > ts=1647274892728 >> > watermark=1647575140007 >> > record=... >> > >> > >> > >> > >> > public static class TextLog extends ProcessFunction<Record, Record> { >> > private final String label; >> > public TextLogDeliveryLog(String label) { >> > this.label = label; >> > } >> > @Override >> > public void processElement(Record record, Context context, >> Collector<Record> collector) throws Exception { >> > LOGGER.info("{}\nts={}\nwatermark={}\nrecord={}", >> > label, context.timestamp(), >> context.timerService().currentWatermark(), record); >> > collector.collect(deliveryLog); >> > } >> > } >> > >> > Diese Nachricht ist ausschliesslich für den Adressaten bestimmt und >> beinhaltet unter Umständen vertrauliche Mitteilungen. Da die >> Vertraulichkeit von e-Mail-Nachrichten nicht gewährleistet werden kann, >> übernehmen wir keine Haftung für die Gewährung der Vertraulichkeit und >> Unversehrtheit dieser Mitteilung. Bei irrtümlicher Zustellung bitten wir >> Sie um Benachrichtigung per e-Mail und um Löschung dieser Nachricht sowie >> eventueller Anhänge. Jegliche unberechtigte Verwendung oder Verbreitung >> dieser Informationen ist streng verboten. >> > >> > This message is intended only for the named recipient and may contain >> confidential or privileged information. As the confidentiality of email >> communication cannot be guaranteed, we do not accept any responsibility for >> the confidentiality and the intactness of this message. If you have >> received it in error, please advise the sender by return e-mail and delete >> this message and any attachments. Any unauthorised use or dissemination of >> this information is strictly prohibited. >> >>
