gistfile1.txt

  private def configure(builder: StreamsBuilder): KStream[String, String] = {
    val aggregateTable: KTable[Windowed[String], String] = builder
      .stream[String, String]("input-topic")
      .filter((key: String, _: String) => !key.isEmpty)
      .groupByKey
      .windowedBy(SessionWindows.`with`(TimeUnit.SECONDS.toMillis(windowSize)))
      .aggregate(
        () => MapperUtil.asString(ConversionAggregate(None, None)),
        ConversionAggregator.apply,
        ConversionMerger.apply
      )

    aggregateTable
      .toStream[String]((key: Windowed[String], _: String) => key.key())
      .filter((_, value) => value != null) // ignore tombstone records
  }