java - Flink ウィンドウと折り畳み機能を使用すると、要素が欠落していますか?

Question

ウィンドウとフォールド関数を使用して要素を集約しようとすると、一部の要素が集約されません。Kafka からの要素を消費(value:0, value:1, value:2, value:3)し、それらを奇数および偶数の値として集計します。

出力は次のとおりです。

{even=[0, 2, 4], odd=[1, 3]}
{even=[6, 8], odd=[5, 7, 9]}
{even=[14, 16, 18], odd=[15, 17]}
{even=[20, 22], odd=[19, 21, 23]}
{even=[24, 26, 28], odd=[25, 27]}

10 ～ 13 の数字が欠落しており、これはランダムな数字のセットで発生します。以下のコードで何が欠けているか、またすべての要素を確実に処理するにはどうすればよいか、誰かが提案できますか?

public static class Splitter implements FlatMapFunction<String, 
    Tuple3<String, String, List<String>>{
    private static final long serialVersionUID = 1L;

    @Override
    public void flatMap(String value, Collector<Tuple3<String, String, 
        List<String>>out) throws Exception {
        String[] vals = value.split(":");

        if(vals.length 1 && Integer.parseInt(vals[1]) % 2 == 0){
            out.collect(new Tuple3<String, String, List<String>>
             ("test","even", Arrays.asList(vals[1])));
        }else{
            out.collect(new Tuple3<String, String, List<String>>
            ("test","odd", Arrays.asList(vals[1])));
        }
    }
}


    DataStream<Map<String, List<String>>streamValue = 
    kafkaStream.flatMap(new Splitter()).keyBy(0)
    .window(TumblingEventTimeWindows.of(Time.milliseconds(3000))).
    trigger(CustomizedCountTrigger.of(5L))//.trigger(CountTrigger.of(2))
    .fold(new HashMap<String, List<String>>(), new 
    FoldFunction<Tuple3<String, String, List<String>>, Map<String, 
    List<String>>>() {
        private static final long serialVersionUID = 1L;

        @Override
        public Map<String, List<String>fold(Map<String, 
        List<String>accumulator,
        Tuple3<String, String, List<String>value) throws 
        Exception {
            if(accumulator.get(value.f1) != null){
                List<Stringlist = new ArrayList<>();
                list.addAll(accumulator.get(value.f1));
                list.addAll(value.f2);
                accumulator.put(value.f1, list);
            }else{
                accumulator.put(value.f1, value.f2);
            }
            return accumulator;
        }
    });

    streamValue.print();
    env.execute("window test");
}


public class CustomizedCountTrigger<W extends Windowextends 
Trigger<Object, W{

    private static final long serialVersionUID = 1L;
    private final long maxCount;

    private final ReducingStateDescriptor<LongstateDesc =
    new ReducingStateDescriptor<>("count", new Sum(), 
    LongSerializer.INSTANCE);

    private CustomizedCountTrigger(long maxCount) {
        this.maxCount = maxCount;
    }

    @Override
    public TriggerResult onElement(Object element, long timestamp, W window,
    TriggerContext ctx) throws Exception {
        ReducingState<Longcount = ctx.getPartitionedState(stateDesc);
        count.add(1L);
        if (count.get() >= maxCount) {
            count.clear();
            return TriggerResult.FIRE_AND_PURGE;
        }
        return TriggerResult.CONTINUE;
    }

    @Override
    public TriggerResult onProcessingTime(long time, W window,

    org.apache.flink.streaming.api.windowing.triggers.Trigger.TriggerContext

    ctx) throws Exception {
        return TriggerResult.CONTINUE;
    }

    @Override
    public TriggerResult onEventTime(long time, W window,

    org.apache.flink.streaming.api.windowing.triggers.Trigger.TriggerContext

    ctx) throws Exception {
        return TriggerResult.CONTINUE;
    }

    @Override
    public void clear(W window, 
    org.apache.flink.streaming.api.windowing.triggers.Trigger.TriggerContext

    ctx)
    throws Exception {
        ctx.getPartitionedState(stateDesc).clear();
    }

    @Override
    public String toString() {
        return "CountTrigger(" +  maxCount + ")";
    }

    public static <W extends WindowCustomizedCountTrigger<Wof(long 
    maxCount) {
        return new CustomizedCountTrigger<>(maxCount);
    }

    private static class Sum implements ReduceFunction<Long{
        private static final long serialVersionUID = 1L;

        @Override
        public Long reduce(Long value1, Long value2) throws Exception {
            return value1 + value2;
        }

    }
}