Hello, I have a question about the custom implementation of the window strategy (Using Kafka as a source). According to this doc, it looks like the Flink supports per-partition watermark with custom event time from POJO (after key-by operation), which looks fine. In my specific use case, I have Kafka as a source that gets events from two different regions (by an upstream service). In some cases, one region would process the events at a slower rate due to some issue or strange scenario. In that case, the per partition watermark is not handling my case, because if one of the regions inserts data, then the window would close. Is there a way to handle my case? One of the hacky solutions is to have X partitions and use the first X/2 partition for a region X and similarly, the second X/2 for another region Y, in this case, I am able to solve my use case. Is there any other decent way to extend and support this using a custom watermark generator with per-partition + custom field (region)?

What I think should work is a custom WatermarkStrategy that in the onEvent method independently tracks the largest timestamp seen so far for each region, and in the onPeriodicEmit method uses the smaller of those two max timestamp values as the basis for the watermark. You can use the implementation of BoundedOutOfOrdernessWatermarks as a starting point.

Thanks David. Looks like promising. Just to be clear from this code, I have to try something like below, am I right?

// Instance variable 
long maxTimestampInX=Long.MIN_VALUE;
long maxTimestampInY=Long.MIN_VALUE;

// Method
public void onEvent(MyEvent event, long eventTimestamp, WatermarkOutput output) {
        if(event.isRegionX()) {
            maxTimestampInX = Math.max(maxTimestampInX, eventTimestamp);
        } else if(event.isRegionY()) {
            maxTimestampInY = Math.max(maxTimestampInY, eventTimestamp);
        } else {
            throw new RuntimeException("Region is not X OR Y");
        }
    }


public void onPeriodicEmit(WatermarkOutput output) {
        output.emitWatermark(new Watermark(Math.min(maxTimestampInX, maxTimestampInY)));
    }

The watermark being emitted should be more like

new Watermark(Math.min(maxTimestampInX, maxTimestampInY) - this.outOfOrdernessMillis - 1)

Ok. Right! I think I get it. Final question, By this way, still the per-partition watermark are guaranteed in all cases, right? ie - before keyBy and after keyBy and possible whatever the watermark strategy work like?

You only get per-partition watermarking if you arrange for the KafkaSource to apply the watermark strategy in the source. Once the event records leave the source, a downstream watermark strategy can't do per-partition watermarking.

hmm. I think I have to dig more about the docs. One more question then, Lets say if I have the following topology, KafkaSource -> Map -> KeyBy -> Aggregation -> Sink In this case, assume we have 4 partition in the source Kafka and we have 2 parallelism for each task. Each task should take care of two partition. So after KeyBy, I want the aggregation task to kick in once all the partition (4 partition) crosses its watermark. How to ensure this?

That should just work.

I see. let me test this then. Thanks a lot David!

I think this should work, but let me know if it doesn’t.

Sure!