title: How to look up and update the state of a record from a database in Apache Flink?
body: <p>I am working on a data streaming application and I am investigating the possibility of using Apache Flink for this project. The main reason for this is that it supports nice high-level streaming constructs, very similar to Java 8's Stream API. </p>
  
  <p>I will be receiving events which correspond to a specific record in a database, and I want to be able to process these events (coming from a message broker such as RabbitMQ or Kafka) and eventually update the records in the database and push the processed/transformed events to another sink (probably another message broker).</p>
  
  <p>Events related to a specific record will ideally need to be processed in FIFO ordering (although there will be a timestamp which helps detect out of order events too), but events related to different records can be processed in parallel. I was planning to use the <code>keyBy()</code> construct to partition the stream by record. </p>
  
  <p>The processing that needs to be done depends on the current information in the database about the record. However, I am unable to find an example or recommended approach to query a database for such records to enrich the event that it is being processed with the additional information I need to process it. </p>
  
  <p>The pipeline I have in mind is as follows:</p>
  
  <p>-> keyBy() on the id received -> retrieve the record from the database corresponding to the id -> perform processing steps on the record -> push the processed event to an external queue and update the database record </p>
  
  <p>The database record will need to be updated because another application will be querying the data. </p>
  
  <p>There might be additional optimisations one could do after this pipeline is achieved. For example one could cache the (updated) record in a managed state so that the next event on the same record will not need another database query. However, if the application does not know about a specific record it will need to retrieve it from the database. </p>
  
  <p>What is the best approach to use for this kind of scenario in Apache Flink?</p>