Anyone trying to adopt Scotty on the recent Flink versions?

[Dongwon Kim]

Hi community,

We're recently trying to adopt Scotty to overcome the poor performance
caused by too many sliding windows.

We're facing the following exception on the latest Flink-1.14.2:

switched from RUNNING to FAILED with failure cause:
java.lang.ArrayIndexOutOfBoundsException: -1
        at java.util.ArrayList.elementData(ArrayList.java:424)
        at java.util.ArrayList.get(ArrayList.java:437)
        at 
de.tub.dima.scotty.slicing.aggregationstore.LazyAggregateStore.getSlice(LazyAggregateStore.java:53)
        at 
de.tub.dima.scotty.slicing.aggregationstore.LazyAggregateStore.insertValueToSlice(LazyAggregateStore.java:64)
        at 
de.tub.dima.scotty.slicing.SliceManager.processElement(SliceManager.java:76)
        at 
de.tub.dima.scotty.slicing.SlicingWindowOperator.processElement(SlicingWindowOperator.java:43)
        at 
de.tub.dima.scotty.flinkconnector.KeyedScottyWindowOperator.processElement(KeyedScottyWindowOperator.java:62)
        at 
org.apache.flink.streaming.api.operators.KeyedProcessOperator.processElement(KeyedProcessOperator.java:83)
        at 
org.apache.flink.streaming.runtime.tasks.OneInputStreamTask$StreamTaskNetworkOutput.emitRecord(OneInputStreamTask.java:233)
        at 
org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.processElement(AbstractStreamTaskNetworkInput.java:134)
        at 
org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:105)
        at 
org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)
        at 
org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:496)
        at 
org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:203)
        at 
org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:809)
        at 
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:761)
        at 
org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:958)
        at 
org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:937)
        at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:766)
        at org.apache.flink.runtime.taskmanager.Task.run(Task.java:575)
        at java.lang.Thread.run(Thread.java:748)

We create a window operator as follows:

import de.tub.dima.scotty.core.windowType.{SlidingWindow, WindowMeasure}
import de.tub.dima.scotty.flinkconnector.KeyedScottyWindowOperator
import de.tub.dima.scotty.flinkconnector.{_}

    val windowOp = new KeyedScottyWindowOperator[(java.lang.String,
Long), NaviGpsProcessable, NaviTrafficUserResult](new
NaviTrafficUserAggregationScotty())
    windowOp.addWindow(new SlidingWindow(WindowMeasure.Time, 600_000, 60_000))

    val userAggStream = stream
      .keyBy(el => (el.id, el.trafficId))
      .process(windowOp)
      .map(_.getAggValues.get(0));

Can I get any advice on this?

Best,

Dongwon