I was checking the CoProcessFunction example mentioned here -- https://nightlies.apache.org/flink/flink-docs-stable/docs/learn-flink/etl/#connected-streams For some reason it doesnt filter for me and prints everything out But If I set env.setParallelism(1); It works as expected. Can someone point out what am i missing here ? Why doesn't t work if parallelism is not 1.

You're not missing anything. Any flink operator with multiple inputs, like the one wrapped around a RichCoFlatMapFunction, consumes records from its various input channels non-deterministically. There's a race condition if this example is run with parallelism greater than 1. So sometimes this example will filter one or both of the relevant records, and sometimes it won't filter anything at all.

thank you for the reply..so should we always run with parallelism 1 ?

No, but if your use case depends on the order in which things are processed, then you have ensure that the ordering you require is made to happen. In a case like this, you could put timestamps on everything, and buffer (in Flink state) whichever side is ahead. As events arrive from the side that’s further behind, you can go ahead and process the older events from the side that is ahead (taking them from the buffer). (This is a somewhat simplified explanation — just trying to convey the flavor of what’s needed.) Temporal joins (from the SQL/Table API) implement this pattern for you, so in most cases there’s no need to implement this logic yourself.

Got it. Thank you