title: Apache Flink: How are events partitioned for a keyed CoFlatMapFunction?
body: <p>This is a pretty basic question about connected keyed stream.</p>
  
  <p>If I have two streams with related events that share same logical key, and these streams are being connected (logically joined using the key) and this is all running with parallelism > 1, then how does Flink guarantee that two events from different streams with same logic key end up in the same parallel operator instance?</p>
  
  <p>Here is a made-up example about hospital's patient streams - temperature stream and heartbeat stream. We want to join these two stream's by patient's id using <code>ConnectedStream</code> and <code>CoFlatMapFunction</code>.</p>
  
  <pre><code>DataStream&lt;PatientTemperature&gt; temperatureStream = ..
  DataStream&lt;HeartbeatStream&gt; heartbeatStream = ..
  
  temperatureStream
     .keyBy(pt -&gt; pt.getPatientId())
     .connect (heartBeatStream.keyBy(hbt -&gt; hbt.getPatientId() )
     .flatMap (new RichCoFlatMapFunction() {
  
           ValueState&lt;PatientTemperatureAndHeartBeat&gt; state = ...
  
           public void flatMap1(PatientTemperature value, Collector&lt;PatientTemperatureAndHeartBeat&gt; out) {
                  state.value().setTemperature(value);  
           }
  
        public void flatMap2(PatentHeartbeat value, Collector&lt;PatientTemperatureAndHeartBeat&gt; out) {
  
                 PatientTemperatureAndHeartBeat temperatureAndHeartBeat = state.value()
                 temperatureAndHeartBeat.setHeartBeat(value)
                 out.collect(temperatureAndHeartBeat);
  
        }
  
        });
  </code></pre>
  
  <p>Assume this is running with parallelism = 3, with operator tasks A, B, C, and they are all running in different physical machines.</p>
  
  <p>Flink will guarantee that all <code>Temperature</code> events for patient "JohnDoe" will end up in the same parallel operator instance. Say it ends up in Operator B.</p>
  
  <p>But when Flink receives <code>HeartBeat</code> events for "JohnDoe", how does it know to send them to Operator B where the patient's <code>Temperature</code> events were getting sent. Unless both <code>Temperature</code> and <code>HeartBeat</code> event are sent to the same parallel instance operator, the join would not work.</p>
  
  <p>The fact that both streams are using the same logical key ( i.e patient's id) is application-specific and Flink does not know about. These two connected streams could be using their own keys which are unrelated to each other.</p>