我在flink中有两个数据流（带有公共时间戳和kafka），其中一个包含一些信号值，另一个包含活动（简单的active-inactive）信息。我试过了 RichCoProcessFunction 一个简单的状态 private ValueState<Boolean> seen; 结果是不确定的。如果我使用 startFromEarliest 我有时会过滤不同的值。我怎样才能使它具有确定性？我在分享我的 KeyedCoProcessFunction 下面是骷髅。

private ValueState < Boolean > seen;

@Override
public void open(Configuration parameters) throws Exception {
    ValueStateDescriptor < Boolean > descriptor = new ValueStateDescriptor < > (
        // state name
        "have-seen-key",
        // type information of state
        TypeInformation.of(new TypeHint < Boolean > () {}));
    seen = getRuntimeContext().getState(descriptor);
}

@Override
public void processElement1(SomeEvent < Double > value, Context ctx, Collector < SomeEvent < Double >> out) throws Exception {
    if (seen.value() == Boolean.TRUE) {
        out.collect(value);
    }
}

@Override
public void processElement2(SomeEvent < Double > value, Context ctx, Collector < SomeEvent < Double >> out) throws Exception {
    if (value.value == 1) {
        seen.update(Boolean.TRUE);

    } else {
        seen.update(Boolean.FALSE);
    }

}