KTable<Windowed<String>, CourseStatistic> windowedCourseStatisticKTable = validReviews .filter((k, review) -> !isReviewExpired(review)) .groupByKey().aggregate( this::emptyStats, this::reviewAggregator, timeWindows, courseStatisticSpecificAvroSerde );