aggregate does not allow RichAggregateFunction ?

[Vishal Santoshi]

I have a simple Aggregation with one caveat. For some reason I have to keep
a large amount of state till the window is GCed. The state is within the
Accumulator ( ACC ). I am hitting a memory bottleneck and would like to
offload the state  to the states backend ( ROCKSDB), keeping the between
checkpoint state in memory ( seems to be an obvious fix). I am not though
allowed to have a RichAggregateFunction in the aggregate method of a
windowed stream . That begs 2 questions

1. Why
2. Is there an alternative for stateful window aggregation where we manage
the state. ?

Thanks Vishal


Here is the code ( generics but it works  )

SingleOutputStreamOperator<OUT> retVal = input
        .keyBy(keySelector)
        .window(EventTimeSessionWindows.withGap(gap))
        .aggregate(
                new AggregateFunction<IN, ACC, OUT>() {

                    @Override
                    public ACC createAccumulator() {
                        ACC newInstance = (ACC) accumulator.clone();
                        newInstance.resetLocal();
                        return newInstance;
                    }

                    @Override
                    public void add(IN value, ACC accumulator) {
                        accumulator.add(value);

                    }

                    @Override
                    public OUT getResult(ACC accumulator) {
                        return accumulator.getLocalValue();
                    }

                    @Override
                    public ACC merge(ACC a, ACC b) {
                        a.merge(b);
                        return a;
                    }
                }, new WindowFunction<OUT, OUT, KEY, TimeWindow>() {
                    @Override
                    public void apply(KEY s, TimeWindow window,
Iterable<OUT> input, Collector<OUT> out) throws Exception {
                            out.collect(input.iterator().next());
                    }
                }, accType, aggregationResultType, aggregationResultType);