可以将处理器api与dsl混合并匹配，但不能处理ktable。您需要转换为kstream。或者，您可以使用与状态存储交互的处理器创建一个新拓扑。
您需要将该状态存储在某个位置—如何确定记录是否早于一小时。一种方法是为状态存储中的每条记录添加一个时间戳。
在处理器的init方法中，可以调用schedule（标点符号）来迭代状态存储中的记录并删除旧记录：

context.schedule(Duration.ofMillis(everyHourInMillis), PunctuationType.WALL_CLOCK_TIME, timestamp -> {
    myStateStore.all().forEachRemaining(keyValue -> {
        if (Instant.ofEpochMilli(valueInStateStore).compareTo(olderThanAnHour) < 0) {
            myStateStore.delete(keyValue.key);
        }
    });
});