https://flink.apache.org/ logo
Join Slack
Powered by
Hi All- having an issue where the watermark never ...
# troubleshooting
c
Hi All- having an issue where the watermark never seems to move forward past 
-9223372036854775808
, meaning non of my timers in my process function are working. I have 5 events coming exactly every 16 to 17 ms (the 5 events have the same timestamp, but this should be fine with a monotonous timestamp strategy?) and the auto watermark interval to be 50ms, so as I understand it, the watermark should be updated by the 4th ish event, but it's not. The timestmaps seem fine though. I've seen other people have this issue, but none of the aforementioned fixes seem to be working (ex. set parallelism to 1). This is my watermark generator:
Copy code
class MyTimestampAssigner(TimestampAssigner):
    def extract_timestamp(self, value, record_timestamp):
        time = int(dateparser.parse(value[0]).timestamp() * 1000)
        return time

watermark_strategy = WatermarkStrategy.for_monotonous_timestamps().with_timestamp_assigner(MyTimestampAssigner())
j
Few things to look into. 1. Did you set the auto watermark interval on the stream? 
env.getConfig.setAutoWatermarkInterval(TimeUnit.SECONDS.toMillis(10))
2. Are all sub tasks receiving data? If some of the partitions aren't receiving data, the event timestamp won't advance. This can be resolved by calling 
withIdlenss
on the watermark strategy. i.e. 
WatermarkStrategy.for_monotonous_timestamps().with_idleness(100)
c
yep have tried both of those things
j
Can you link a snippet of the stream where the watermark strategy is used?
c
Copy code
self.env.set_parallelism(1)

        loc_stream = input_streams.get("com_feed_topic")
        watermark_strategy = WatermarkStrategy.for_monotonous_timestamps().with_timestamp_assigner(MyTimestampAssigner())
        processed_loc_stream = loc_stream.flat_map(LocationMapper(), 
                                                   output_type=Types.TUPLE([Types.STRING(), 
                                                                            Types.BASIC_ARRAY(Types.FLOAT()), 
                                                                            <http://Types.INT|Types.INT>(), 
                                                                            Types.BOOLEAN()])) \
                                        .assign_timestamps_and_watermarks(watermark_strategy)
The first string element of the stream used for timestamp extraction appears like this is the stream, with many of the same times in a row, then a jump of 16-17ms, then many of the next timestamp. This is converted to epoch time in milliseconds the timestamp extractor (code for that is above):
Copy code
"2023-06-13T00:39:06.073Z"
"2023-06-13T00:39:06.073Z"
"2023-06-13T00:39:06.073Z"
"2023-06-13T00:39:06.073Z"
"2023-06-13T00:39:06.073Z"
"2023-06-13T00:39:06.089Z"
"2023-06-13T00:39:06.089Z"
"2023-06-13T00:39:06.089Z"
"2023-06-13T00:39:06.089Z"
"2023-06-13T00:39:06.089Z"
"2023-06-13T00:39:06.106Z"
"2023-06-13T00:39:06.106Z"
"2023-06-13T00:39:06.106Z"
"2023-06-13T00:39:06.106Z"
"2023-06-13T00:39:06.106Z" ...
2 Views
Open in Slack
PreviousNext
Powered by