title: Dynamically create and change Kafka topics with Flink
body: <p>I'm using Flink to read and write data from different Kafka topics.
  Specifically, I'm using the FlinkKafkaConsumer and FlinkKafkaProducer.</p>
  
  <p>I'd like to know if it is possible to change the Kafka topics I'm reading from and writing to 'on the fly' based on either logic within my program, or the contents of the records themselves.</p>
  
  <p>For example, if a record with a new field is read, I'd like to create a new topic and start diverting records with that field to the new topic.</p>
  
  <p>Thanks.</p>