有没有办法对Kafka streams中的count函数应用过滤器？

import org.apache.kafka.streams.processor.AbstractProcessor;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.PunctuationType;
import org.apache.kafka.streams.state.KeyValueStore;

public class CountFilterProcessor extends AbstractProcessor<String, Long> {
    private ProcessorContext context;
    private KeyValueStore<String, Long> countStore;

    @Override
    public void init(ProcessorContext context) {
        this.context = context;
        this.countStore = (KeyValueStore<String, Long>) context.getStateStore("count-store");

        // 定期调度一个punctuate方法，用于处理过滤逻辑
        this.context.schedule(1000, PunctuationType.WALL_CLOCK_TIME, this::punctuate);
    }

    @Override
    public void process(String key, Long value) {
        // 更新count值
        countStore.put(key, value);

        // 发送过滤后的结果到下一个处理节点
        if (value > 100) {
            context.forward(key, value);
        }
    }

    private void punctuate(long timestamp) {
        // 在这里可以添加其他过滤逻辑，比如定期清理countStore中的数据

        // 提交处理进度
        context.commit();
    }

    @Override
    public void close() {
        // 关闭资源
        countStore.close();
    }
}

KStream<String, Long> inputStream = builder.stream("input-topic");
inputStream.transform(() -> new CountFilterProcessor(), "count-store")
         .to("output-topic");