title: Dynamic Kafka Consumer using functional paradigm
body: <p>I have a particulier requirement in which I want to collect messages from a topic until a specified duration (for example for 40 seconds), but only when asked to (so start a consumer for 40 sec when asked for and then stop).</p>
  <p>I came across examples over creating consumers dynamically using</p>
  <ol>
  <li>DefaultKafkaConsumerFactory</li>
  <li>ConcurrentKafkaListenerContainerFactory</li>
  </ol>
  <ol>
  <li><p><strong>Firstly</strong>, I would like to know if its possible to do the same using <strong>KStream</strong> + <strong>functional paradigm (i.e. using (bi)function, consumer interfaces)?</strong></p>
  </li>
  <li><p><strong>Secondly, What should I do if another call comes in to start collecting messages while the first duration hasn't finished ?</strong> <em>Create a new container (unique group-id ofcoarse) ??</em></p>
  </li>
  <li><p><strong>Lastly, When the collect duration has expired, I have no use of this container anymore. I could stop the container but what then, can it be reused, i.e. when a new request comes in ?</strong></p>
  </li>
  </ol>