title: Flink KeyedCoProcessFunction working with state
body: <p>I use <code>KeyedCoProcessFunction</code> function to enrich main datastream with data comes from another stream</p>
  <p>Code:</p>
  <pre class="lang-scala prettyprint-override"><code>class AssetDataEnrichment extends KeyedCoProcessFunction[String, PacketData, AssetCommandState, AssetData] with LazyLogging {
  
    case class AssetStateDoc(assetId: Option[String])
    private var associatedDevices: ValueState[AssetStateDoc] = _
  
    override def open(parameters: Configuration): Unit = {
      val associatedDevicesDescriptor =
        new ValueStateDescriptor[AssetStateDoc](&quot;associatedDevices&quot;, classOf[AssetStateDoc])
      associatedDevices = getRuntimeContext.getState[AssetStateDoc](associatedDevicesDescriptor)
    }
  
    override def processElement1(
                                  packet: PacketData,
                                  ctx: KeyedCoProcessFunction[String, PacketData, AssetCommandState, AssetData]#Context,
                                  out: Collector[AssetData]): Unit = {
      
      val tmpState = associatedDevices.value
      val state = if (tmpState == null) AssetStateDoc(None) else tmpState
      
      state.assetId match {
        case Some(assetId) =&gt;
          logger.debug(s&quot;There are state for ${packet.tag.externalId} = $assetId&quot;)
          out.collect(AssetData(assetId, packet.tag.externalId.get, packet.toString))
        case None =&gt; logger.debug(s&quot;No state for a packet ${packet.tag.externalId}&quot;)
        case _ =&gt; logger.debug(&quot;Smth went wrong&quot;)
      }
    }
  
    override def processElement2(
                                  value: AssetCommandState,
                                  ctx: KeyedCoProcessFunction[String, PacketData, AssetCommandState, AssetData]#Context,
                                  out: Collector[AssetData]): Unit = {
      
      value.command match {
        case CREATE =&gt;
          logger.debug(s&quot;Got command to CREATE state for tag: ${value.id} with value: ${value.assetId}&quot;)
          logger.debug(s&quot;current state is ${associatedDevices.value()}&quot;)
          associatedDevices.update(AssetStateDoc(Some(value.assetId)))
          logger.debug(s&quot;new state is ${associatedDevices.value()}&quot;)
        case _ =&gt;
          logger.error(&quot;Got unknown AssetCommandState command&quot;)
      }
    }
  }
  
  </code></pre>
  <p><code>processElement2()</code> works good, it's accept data and update a state.<br />
  but in a <code>processElement1()</code> I am always hitting <code>case None =&gt; logger.debug(s&quot;No state for a packet ${packet.tag.externalId}&quot;)</code></p>
  <p>although I expect that there will be a value that was set in <code>processElement2</code> function</p>
  <p>as an example I used this guide - <code>https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/fault-tolerance/state/</code></p>