Apache Flink：自定义触发器意外行为

Apache Flink：自定义触发器意外行为
我有一个DataStream，它由一个事件组成，该事件具有一个表示一批生成元素的属性。该属性，我们称之为'batchNumber'，在我从同一生产批次中摄取的每个事件中都是恒定的。我每批收到多个事件。

我想在'batchNumber'更改时分析批处理中的机器性能。我的方法是使用全局流并使用'batchNumber'作为密钥对其进行分区。我希望这会将全局流分区为窗口，其中每个事件都有'batchNumber'。然后我定义一个触发器，当'batchNumber'发生变化时应该触发。然后我可以分析ProcessWindowFunction中的聚合数据。

我的问题是：

当prodnr改变时，触发器并不总是触发
即使它发射，也只有一个元素被聚合。我期待接近200。
这是我正在使用的代码。

public class batchnrTrigger extends Trigger<ImaginePaperData, GlobalWindow> {

private static final long serialVersionUID = 1L;

public batchnrTrigger() {}

private final ValueStateDescriptor<Integer> prevbatchnr = new ValueStateDescriptor<>("batchnr", Integer.class);

@Override
public TriggerResult onElement(ImaginePaperData element, long timestamp, GlobalWindow window, TriggerContext ctx) throws Exception {

    ValueState<Integer> batchnrState = ctx.getPartitionedState(prevbatchnr);

    if (batchnrState == null || batchnrState.value() == null || !(element.batchnr == batchnrState.value())) {

        System.out.println("batchnr BEFORE: " + batchnrState.value() + "   NEW batchnr: " + element.batchnr + " ==> should fire and process elements from window!");
        batchnrState.update(element.batchnr);
        return TriggerResult.FIRE;

    }

    System.out.println("batchnr BEFORE: " + batchnrState.value() + "   NEW batchnr: " + element.batchnr + " ==> should not fire and continue ingesting elements!");
    batchnrState.update(element.batchnr);
    return TriggerResult.CONTINUE;
}

@Override
public TriggerResult onProcessingTime(long time, GlobalWindow window, TriggerContext ctx) throws Exception {
    return TriggerResult.CONTINUE;
}

@Override
public TriggerResult onEventTime(long time, GlobalWindow window, TriggerContext ctx) throws Exception {
    return TriggerResult.CONTINUE;
}

@Override
public void clear(GlobalWindow window, TriggerContext ctx) throws Exception {

}

}
这就是我称之为触发器的方式：

DataStream imaginePaperDataStream = nifiStreamSource

    .map(new ImaginePaperDataConverter())
    .keyBy((ImaginePaperData event) -> event.lunum)
    .window(GlobalWindows.create())
    .trigger(new LunumTrigger())
    .process(new ImaginePaperWindowReportFunction());