Would I be correct in presuming that Flink lacks the semantics to attempt to minimize hash exchanges by trying to take advantage of the partition assignments in a Kafka source? If it’s possible, I’d look into changing the upsert-kafka connector to support it, but it doesn’t seem likely.