Hello, I'm playing with aggregating a value by a key on a tumbling window over 1 minute. Each record has an attribute named marker. There is a single late record and it is marked as late. As the window size is one minute, I expect the late value should be used for aggregation together if it is late by less than 60 seconds. Below I made it to be late for 50 seconds (50000 milliseconds) but it is aggregated in the next window. If I change it to 30 seconds, for example, it is aggregated in the same window. Is there something I misunderstand?

...

class CustomTimestampAssigner(TimestampAssigner):
    def extract_timestamp(self, value, record_timestamp):
        if value.marker == "late":
            return int(time.time_ns() / 1000000) + 50000
        else:
            return int(time.time_ns() / 1000000)


custom_watermark_strategy = WatermarkStrategy.for_monotonous_timestamps().with_timestamp_assigner(
    CustomTimestampAssigner()
)

env = StreamExecutionEnvironment.get_execution_environment()

data_stream = env.from_collection(
    collection=[(1, "a", "on_time"), (2, "a", "on_time"), (3, "a", "late")],
    type_info=MyData.get_type_info(),
).assign_timestamps_and_watermarks(custom_watermark_strategy)

data_stream.map(MyData.from_row).map(Stats.from_mydata).key_by(lambda s: s.key).window(
    TumblingEventTimeWindows.of(Time.minutes(1))
).reduce(Stats.merge).print()

env.execute()

# expected

3> Stats(value=6, key='a')

# actual

3> Stats(value=3, key='a')
3> Stats(value=3, key='a')

It’s late for me, so I might be completely off, but isn’t that message early rather than late? If it were late, it would arrive at a later time, but carry a smaller timestamp, right? In other words, a late arriving message would be a message from the past, not the future 🙂 Try a minus sign instead of +

The late data is just coming late. Here the issue is, for a window size of 1 minute, • a record is aggregated in the next window if it is late by 50 seconds, on the other hand • it is aggregated in the same window if it is late by 30 seconds To my understanding, both of them should be aggregated in the same window as they are late by less than 60 seconds.

Yeah, but it isn’t actually late, it’s early. If you look at it as time stamps (example): 12:00 12:01 12:02 11:59 < this one arrived late 12:03 This means that to make something late, it’s time should be small compared to the other messages, not large

Actually it's translated from a unit test case from the building apache flink applications in java course by confluent. https://github.com/confluentinc/learn-building-flink-applications-in-java-exercise[…]windows/src/test/java/userstatistics/UserStatisticsJobTest.java The test case indicates if a record is late (in terms of proc_time) for 1 minute, it should be aggregated in the next window. I works as expected. However, if I make it smaller to 50 seconds, it keeps being aggregated in the next window. This is what I don't understand as the tumbling window size is 1 minute.