我希望将kstream中的窗口批输出组合在一起，并将它们写入辅助存储。
我本来想看看的 .punctuate() 大概每30秒打一次电话。我得到的却保存在这里。
（原始文件有几千行）
摘要- .punctuate() 似乎是随机的，然后是反复的。它似乎不符合通过processorcontext.schedule（）设置的值。

编辑：

相同代码的另一次运行产生了对 .punctuate() 大约每四分钟。这次我没有看到疯狂的重复值。来源没有变化-只是结果不同。
使用以下代码：

主要

StreamsConfig streamsConfig = new StreamsConfig(config);
KStreamBuilder kStreamBuilder = new KStreamBuilder();
KStream<String, String> lines = kStreamBuilder.stream(TOPIC);

lines.process(new BPS2());

KafkaStreams kafkaStreams = new KafkaStreams(kStreamBuilder, streamsConfig);

kafkaStreams.start();

处理器

public class BP2 extends AbstractProcessor<String, String> {
    private static final Logger LOGGER = LoggerFactory.getLogger(BP2.class);

    private ProcessorContext context;
    private final long delay;
    private final ArrayList<String> values;

    public BP2(long delay) {
        LOGGER.debug("BatchProcessor() constructor");
        this.delay = delay;

       values = new ArrayList<>();

    }

    @Override
    public void process(String s, String s2) {
        LOGGER.debug("batched processor s:{}   s2:{}", s, s2);

        values.add(s2);
    }

    @Override
    public void init(ProcessorContext context) {
        LOGGER.info("init");

        super.init(context);

        values.clear();

        this.context = context;
        context.schedule(delay);
    }

    @Override
    public void punctuate(long timestamp) {
        super.punctuate(timestamp);

        LOGGER.info("punctuate   ts: {}   count: {}", timestamp, values.size());

        context().commit();
    }
}

处理器供应商

public class BPS2 implements ProcessorSupplier<String, String> {
    private static final Logger LOGGER = LoggerFactory.getLogger(BPS2.class);

    @Override
    public Processor<String, String> get() {
        try {
            return new BP2(30000);
        } catch(Exception exception) {
            LOGGER.error("Unable to instantiate BatchProcessor()", exception);
            throw new RuntimeException();
        }
    }
}

编辑：

为了确保我的调试器不会减慢速度，我构建了它，并在kafka进程的同一个框中运行它。这一次，它甚至没有尝试延迟4分钟或更长时间——在几秒钟内，它向用户输出虚假呼叫 .punctuate() . 其中许多（大多数）都没有任何干预电话 .process() .