title: KTable state store infinite retention
body: <p>We have the following high level DSL processing topology:</p>
  
  <pre><code>TimeWindows timeWindow = TimeWindows.of(windowDurationMs).advanceBy(windowAdvanceMs).until(retensionTimeMs);
  
  KTable&lt;Windowed&lt;K&gt;, Long&gt; table1 = stream1.groupByKey().count(timeWindow, "Stream_1_Count_Store");
  KTable&lt;Windowed&lt;K&gt;, Long&gt; table2 = stream2.groupByKey().count(timeWindow, "Stream_2_Count_Store");
  
  
  KTable&lt;Windowed&lt;K&gt;, Pair&lt;Long,Long&gt; joined = table1.leftJoin(table2, someValueJoiner, joinSerde, "Join_Store");
  
  KTable&lt;Windowed&lt;SmallerKey&gt;, Tuple&lt;Long,Long,Long&gt;&gt; grouped = joined.groupBy(someSelector);
  
  KTable&lt;Windowed&lt;SmallerKey&gt;, Map&lt;Long, Pair&lt;Long,Long&gt;&gt;&gt; aggregated = grouped.aggregate(initializer, adder, subtractor, aggValueSerde, "Agg_Store_Name")
  </code></pre>
  
  <p>In short, what we're doing above is:</p>
  
  <ol>
  <li>Counting events using a hopping window</li>
  <li>Doing a left join between the resulting KTables (left because of business logic)</li>
  <li>Grouping and changing the key and value: taking a component (Long) of the key and moving to the value</li>
  <li>Aggregating the resulting KTable to a final KTable, the aggregation object is a map from T to the two joined counters from step 1. Note that the size of the map is no more than 600 and usually much less.</li>
  </ol>
  
  <p>The idea is to create windowed event counts and use these windowed keys for the join and aggregate operations (which in the KTable case don't have windows on such operations)</p>
  
  <p>The problem is this: 
  The state stores of the <em>join</em> and <em>aggregate</em> operations have no retention mechanism and cause a space explosion in the disk (RocksDB).</p>
  
  <p>More specifically:
  The (hopping) windows cause a cartesian product on the keys and there's no mechanism to delete old windows.</p>
  
  <p><strong>The same problem will also occur if the KTable key were not windowed, but just a large enough amount of unique keys</strong></p>
  
  <p>Note that the state stores backing table1 and table2 have no space issue, this is because they are provided a windowed store by the DSL which manages dropping old windows.
  In the join and aggregate we treat the windowed keys as "any old key" and the DSL does the same and uses a non-windowed KeyValueStore.</p>
  
  <p>This question is related to the following: <a href="https://issues.apache.org/jira/browse/KAFKA-4212" rel="noreferrer">KAFKA-4212</a>, <a href="https://issues.apache.org/jira/browse/KAFKA-4273" rel="noreferrer">KAFKA-4273</a>, <a href="https://groups.google.com/forum/#!topic/confluent-platform/MxmLmJJPpNY" rel="noreferrer">confluent forum question</a></p>
  
  <p>Are there any misunderstood concepts here?
  Is there a simple way to implement this topology using the DSL?
  If not, what's the suggested ways to implement it using the low level API?</p>