title: Kafka Stream: KTable materialization
body: <p>How to identify when the KTable materialization to a topic has completed?</p>
  
  <p>For e.g. assume KTable has few million rows.  Pseudo code below:</p>
  
  <pre><code>KTable&lt;String, String&gt; kt = kgroupedStream.groupByKey(..).reduce(..); //Assume this produces few million rows
  </code></pre>
  
  <p>At somepoint in time, I wanted to schedule a thread to invoke the following, that writes to the topic:
  kt.toStream().to("output_topic_name"); </p>
  
  <p>I wanted to ensure all the data is written as part of the above invoke.  Also, once the above "to" method is invoked, can it be invoked in the next schedule OR will the first invoke always stay active?</p>
  
  <p><strong>Follow-up Question:</strong></p>
  
  <p><em>Constraints</em><br>
  1) Ok, I see that the kstream and the ktable are unbounded/infinite once the kafkastream is kicked off.  However, wouldn't ktable materialization (to a compacted topic) send multiple entries for the same key within a specified period.  </p>
  
  <p>So, unless the compaction process attempts to clean these and retain only the latest one, the downstream application will consume all available entries for the same key querying from the topic, causing duplicates.  Even if the compaction process does some level of cleanup, it is always not possible that at a given point in time, there are some keys that have more than one entries as the compaction process is catching up.</p>
  
  <p>I assume KTable will only have one record for a given key in the RocksDB.  If we have a way to schedule the materialization, that will help to avoid the duplicates.  Also, reduce the amount of data being persisted in topic (increasing the storage), increase in the network traffic, additional overhead to the compaction process to clean it up.</p>
  
  <p>2) Perhaps a ReadOnlyKeyValueStore would allow a controlled retrieval from the store, but it still lacks the way to schedule the retrieval of key, value and write to a topic, which requires additional coding. </p>
  
  <p>Can the API be improved to allow a controlled materialization?</p>