I think there's a misconception about `setAutowatermarkInterval`. It establishes the rate at which your periodic watermark generator is polled for the current watermark. Like most generators, `BoundedOutOfOrdernessTimestampExtractor` produces a watermark based solely on observed elements. Therefore, `setAutowatermarkInterval` does not compensate for idle sources (see FLINK-5479 and FLINK-5018).
Keep in mind that sources do not re-order emitted elements into event time order; depending on the source's internals, it might emit elements in a highly unordered fashion with respect to event time. For example, the Kafka consumer processes elements across numerous partitions simultaneously, and the resultant ordering is highly variable. When you use the generic `assignTimestampsAndWatermarks` facility, the assigner is challenged to make sense of this multiplexed stream of elements. For this reason, I would strongly suggest you make use of the Kafka consumer's support for per-partition assigners, to be able to reason about the progression of time in each partition independently. Here's a good diagram of the phenomemon that I'm describing. Observe how some elements seem to 'move upward' together, and imagine that they correspond to one partition. https://www.oreilly.com/ideas/the-world-beyond-batch-streaming-102#FIG12 Hope this helps! Eron On Mon, Jan 22, 2018 at 2:24 AM, Fabian Hueske <fhue...@gmail.com> wrote: > Hi William, > > The TsExtractor looks good. > This sounds like a strange behavior and should not (or only indirectly) be > related to the Kafka source since the WMs are generated by a separate > extractor. > > - Did you compare the first (and only) generated watermark to the > timestamps of the records that are produced by the sources? > It might be far ahead of the timestamps in the records and won't be > updated because the timestamps of the records are smaller. > > - What is the parallelism of the file sources / Kafka source and following > operators? > Watermarks can only advance if they advance in all parallel instance of > the timestamp extractor. > > Best, Fabian > > 2018-01-18 16:09 GMT+01:00 William Saar <will...@saar.se>: > >> Hi, >> The watermark does not seem to get updated at all after the first one is >> emitted. We used to get out-of-order warnings, but we changed to job to >> support a bounded timestamp extractor so we no longer get those warnings. >> >> Our timestamp extractor looks like this >> >> class TsExtractor[T](time : Time) extends >> BoundedOutOfOrdernessTimestampExtractor[Timestamped[T]](time : Time) { >> override def extractTimestamp(element: Timestamped[T]): Long = >> element.timestamp >> } >> >> Our stream topology starts with a single stream, then we do two separate >> flat map and filtering operations on the initial stream to transform data >> batches >> into streams of two different event types. We then >> assignTimestampsAndWatermarks(new TsExtractor[EventType](Time.seconds(20))) >> for each event type on both >> branches before unioning the two branches to a single stream again (the >> reason for the split is that the data used to come from two different >> topics). >> >> William >> >> >> >> >> ----- Original Message ----- >> From: >> "Gary Yao" <g...@data-artisans.com> >> >> To: >> "William Saar" <will...@saar.se> >> Cc: >> "user" <user@flink.apache.org> >> Sent: >> Thu, 18 Jan 2018 11:11:17 +0100 >> Subject: >> Re: Far too few watermarks getting generated with Kafka source >> >> >> >> Hi William, >> >> How often does the Watermark get updated? Can you share your code that >> generates >> the watermarks? Watermarks should be strictly ascending. If your code >> produces >> watermarks that are not ascending, smaller ones will be discarded. Could >> it be >> that the events in Kafka are more "out of order" with respect to event >> time than >> in your file? >> >> You can assign timestamps in the Kafka source or later. The Flink >> documentation >> has a section on why it could be beneficial to assign Watermarks in the >> Kafka >> source: >> >> https://ci.apache.org/projects/flink/flink-docs-release-1.4/ >> dev/event_timestamps_watermarks.html#timestamps-per-kafka-partition >> >> Best, >> Gary >> >> On Wed, Jan 17, 2018 at 5:15 PM, William Saar <will...@saar.se> wrote: >> >>> Hi, >>> I have a job where we read data from either Kafka or a file (for >>> testing), decode the entries and flat map them into events, and then add a >>> timestamp and watermark assigner to the events in a later operation. This >>> seems to generate periodic watermarks when running from a file, but when >>> Kafka is the source we barely get any watermark updates. What could be >>> causing this? (the environment has setAutowatermarkInterval(1000)) >>> >>> Do we need to do all the timestamp and watermark assignment in the Kafka >>> source? or should it work to do it in later operations? The events do seem >>> to get propagated through the pipeline, we're just not getting watermarks... >>> >>> Thanks, >>> William >>> >> >> >
