title: Why is this KStream/KTable topology propagating records that don't pass the filter?
body: <p>I have the following topology that: </p>
  
  <ol>
  <li>Creates a state store</li>
  <li>Filters records based on SOME_CONDITION, maps its values to a new entity and finally publishes these records to another topic STATIONS_LOW_CAPACITY_TOPIC</li>
  </ol>
  
  <p>However I am seeing this on the STATIONS_LOW_CAPACITY_TOPIC:</p>
  
  <pre><code>�   null
  �   null
  �   null
  �   {"id":140,"latitude":"40.4592351","longitude":"-3.6915330",...}
  �   {"id":137,"latitude":"40.4591366","longitude":"-3.6894151",...}
  �   null
  </code></pre>
  
  <p>That is to say, it's as if it were also publishing to the STATIONS_LOW_CAPACITY_TOPIC topic those records that didn't pass the filter. How is this possible? How can I prevent them to be published?</p>
  
  <p>This is the ksteams code:   </p>
  
  <pre><code>kStream.groupByKey().reduce({ _, newValue -&gt; newValue },
                  Materialized.`as`&lt;Int, Station, KeyValueStore&lt;Bytes, ByteArray&gt;&gt;(STATIONS_STORE)
                          .withKeySerde(Serdes.Integer())
                          .withValueSerde(stationSerde))
                  .filter { _, value -&gt; SOME_CONDITION }
                  .mapValues { station -&gt;
                      Stats(XXX)
                  }
                  .toStream().to(STATIONS_LOW_CAPACITY_TOPIC, Produced.with(Serdes.Integer(), stationStatsSerde))
  </code></pre>
  
  <p><strong>UPDATE:</strong> I've simplyfied to topology and printed the resulting table. For some reason the final KTable also contains null valued records corresponding to upstream records that didn't pass the filter:</p>
  
  <pre><code>kStream.groupByKey().reduce({ _, newValue -&gt; newValue },
                  Materialized.`as`&lt;Int, BiciMadStation, KeyValueStore&lt;Bytes, ByteArray&gt;&gt;(STATIONS_STORE)
                          .withKeySerde(Serdes.Integer())
                          .withValueSerde(stationSerde))
                  .filter { _, value -&gt;
                      val conditionResult = (SOME_CONDITION)
                      println(conditionResult)
                      conditionResult
                  }
                  .print()
  </code></pre>
  
  <p><strong>Logs:</strong></p>
  
  <pre><code>false
  [KTABLE-FILTER-0000000002]: 1, (null&lt;-null)
  false
  [KTABLE-FILTER-0000000002]: 2, (null&lt;-null)
  false
  [KTABLE-FILTER-0000000002]: 3, (null&lt;-null)
  false
  [KTABLE-FILTER-0000000002]: 4, (null&lt;-null)
  true
  [KTABLE-FILTER-0000000002]: 5, (Station(id=5, latitude=40.4285524, longitude=-3.7025875, ...)&lt;-null)
  </code></pre>