Hi all, I’m trying to troubleshoot an integration test for my Flink pipeline, which is using the

MiniClusterWithClientResource

. At a high level, my pipeline does a map operation, then a keyBy, and then a process function immediately after the keyBy. The process function contains an aggregator that puts records into a state store. In my integration test, I’m feeding in 2 records using

env.fromElements()

. With parallelism set to 1, the test passes every time. With parallelism set to 2 (or really anything greater than 1), the test will intermittently fail. After debugging, what I’m seeing is that occasionally my process function (which extends KeyedProcessFunction) will get the 2nd test record first, in other words - the input records occasionally arrive out-of-order. (I have logic in the process function that expects these records to arrive in the correct order, which is why the test fails) After arriving at the keyBy function, both test records get re-keyed to the same key. My assumption was that they should get processed in timestamp order by the subsequent process function, even if the record with the higher timestamp arrives first. In my test, I’m using a watermark strategy of boundedOutOfOrderness with a duration of 20 seconds. I’ve checked the elements in the process function by debugging, and they do have timestamps set on each record, and the timestamps are correct. I’m not sure if I’m doing something wrong, or misunderstanding how ordering works. When I deploy this same application “for real” in a Flink cluster (where the source data is actually Kafka), I don’t see this issue at all. I only see the problem in the integration test. Thanks for any ideas or advice!