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