[ https://issues.apache.org/jira/browse/FLINK-31970?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17719336#comment-17719336 ]
Piotr Nowojski edited comment on FLINK-31970 at 5/4/23 1:53 PM: ---------------------------------------------------------------- [~YordanPavlov] I was not able to reproduce your error, however you are incorrectly working with the state in your {{TestWindow}} function. Please take a look at the [documentation|https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/fault-tolerance/state/#using-keyed-state]. Especially the following paragraph {quote} It is important to keep in mind that these state objects are only used for interfacing with state. The state is not necessarily stored inside but might reside on disk or somewhere else. The second thing to keep in mind is that the value you get from the state depends on the key of the input element. So the value you get in one invocation of your user function can differ from the value in another invocation if the keys involved are different. {quote} First problem is that your {{TestWindow#snapshoState}} method is trying to access {{state}} field, in a keyed operator/function, inside a method that doesn't have key context. That's the error I was getting when I tried to run your code. Secondly, in your {{TestWindow#apply}} method is invoked within some key context, but you are storing the actual {{count}} as a regular java field. So if you ever had more then one key (that's not the case in your example, as you are using {{.keyBy(_ => 1)}} key selector), all of the keys processed by a single parallel instance of the window operator, would be collected on the same {{count}} field, which doesn't make any sense. The corrected window function should look like this: {code:scala} class CorrectTestWindow() extends WindowFunction[(Long, Int), Long, Int, TimeWindow] with CheckpointedFunction { var state: ValueState[Long] = _ override def snapshotState(functionSnapshotContext: FunctionSnapshotContext): Unit = { } override def initializeState(context: FunctionInitializationContext): Unit = { val storeDescriptor = new ValueStateDescriptor[Long]("state-xrp-dex-pricer", createTypeInformation[Long]) state = context.getKeyedStateStore.getState(storeDescriptor) } override def apply(key: Int, window: TimeWindow, input: lang.Iterable[(Long, Int)], out: Collector[Long]): Unit = { val count = state.value() + input.size state.update(count) out.collect(count) } } {code} However even that is kind of strange. You are creating a tumbling window every 1ms, only to aggregate the {{count}} across all past tumbling windows for the given key? If you want to aggregate count only across the given tumbling window, your apply could look like as simple as: {code:scala} override def apply(key: Int, window: TimeWindow, input: lang.Iterable[(Long, Int)], out: Collector[Long]): Unit = { out.collect(input.size) } {code} And you don't need any additional state for that. On the other hand, if you want to aggregate results globally, you can use {code:scala} .keyBy(_ => 1) .window(GlobalWindows.create()) .tigger(/* maybe a custom trigger here */) .apply(new TestWindow()) {code} was (Author: pnowojski): [~YordanPavlov] I was not able to reproduce your error, however you are incorrectly working with the state in your {{TestWindow}} function. Please take a look at the [documentation|https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/fault-tolerance/state/#using-keyed-state]. Especially the following paragraph {quote} It is important to keep in mind that these state objects are only used for interfacing with state. The state is not necessarily stored inside but might reside on disk or somewhere else. The second thing to keep in mind is that the value you get from the state depends on the key of the input element. So the value you get in one invocation of your user function can differ from the value in another invocation if the keys involved are different. {quote} First problem is that your {{TestWindow#snapshoState}} method is trying to access {{state}} field, in a keyed operator/function, inside a method that doesn't have key context. That's the error I was getting when I tried to run your code. Secondly, in your {{TestWindow#apply}} method is invoked within some key context, but you are storing the actual {{count}} as a regular java field. So if you ever had more then one key (that's not the case in your example, as you are using {{.keyBy(_ => 1)}} key selector), all of the keys processed by a single parallel instance of the window operator, would be collected on the same {{count}} field, which doesn't make any sense. The corrected window function should look like this: {code:scala} class CorrectTestWindow() extends WindowFunction[(Long, Int), Long, Int, TimeWindow] with CheckpointedFunction { var state: ValueState[Long] = _ override def snapshotState(functionSnapshotContext: FunctionSnapshotContext): Unit = { } override def initializeState(context: FunctionInitializationContext): Unit = { val storeDescriptor = new ValueStateDescriptor[Long]("state-xrp-dex-pricer", createTypeInformation[Long]) state = context.getKeyedStateStore.getState(storeDescriptor) } override def apply(key: Int, window: TimeWindow, input: lang.Iterable[(Long, Int)], out: Collector[Long]): Unit = { val count = state.value() + input.size state.update(count) out.collect(count) } } {code} However even that is kind of strange. You are creating a tumbling window every 1ms, only to aggregate the {{count}} across all past tumbling windows for the given key? If you want to aggregate count only across the given tumbling window, your apply could look like as simple as: {code:scala} override def apply(key: Int, window: TimeWindow, input: lang.Iterable[(Long, Int)], out: Collector[Long]): Unit = { out.collect(input.size) } {code} And you don't need any additional state for that. On the other hand, if you want to aggregate results globally, you can use {code:scala} .keyBy(_ => 1) .window(GlobalWindows.create()) .tigger(/* maybe a custom trigger here */) .apply(new TestWindow()) {code} > "Key group 0 is not in KeyGroupRange" when using CheckpointedFunction > --------------------------------------------------------------------- > > Key: FLINK-31970 > URL: https://issues.apache.org/jira/browse/FLINK-31970 > Project: Flink > Issue Type: Bug > Components: Runtime / Checkpointing > Affects Versions: 1.17.0 > Reporter: Yordan Pavlov > Priority: Major > Attachments: fill-topic.sh, main.scala > > > I am experiencing a problem where the following exception would be thrown on > Flink stop (stop with savepoint): > > {code:java} > org.apache.flink.util.SerializedThrowable: > java.lang.IllegalArgumentException: Key group 0 is not in > KeyGroupRange{startKeyGroup=86, endKeyGroup=127}.{code} > > I do not have a non deterministic keyBy() operator in fact, I use > {code:java} > .keyBy(_ => 1){code} > I believe the problem is related to using RocksDB state along with a > {code:java} > CheckpointedFunction{code} > In my test program I have commented out a reduction of the parallelism which > would make the problem go away. I am attaching a standalone program which > presents the problem and also a script which generates the input data. For > clarity I would paste here the essence of the job: > > > {code:scala} > env.fromSource(kafkaSource, watermarkStrategy, "KafkaSource") > .setParallelism(3) > .keyBy(_ => 1) > .window(TumblingEventTimeWindows.of(Time.of(1, TimeUnit.MILLISECONDS))) > .apply(new TestWindow()) > /* .setParallelism(1) this would prevent the problem */ > .uid("window tester") > .name("window tester") > .print() > class TestWindow() extends WindowFunction[(Long, Int), Long, Int, TimeWindow] > with CheckpointedFunction { > var state: ValueState[Long] = _ > var count = 0 > override def snapshotState(functionSnapshotContext: > FunctionSnapshotContext): Unit = { > state.update(count) > } > override def initializeState(context: FunctionInitializationContext): Unit > = { > val storeDescriptor = new > ValueStateDescriptor[Long]("state-xrp-dex-pricer", > createTypeInformation[Long]) > state = context.getKeyedStateStore.getState(storeDescriptor) > } > override def apply(key: Int, window: TimeWindow, input: Iterable[(Long, > Int)], out: Collector[Long]): Unit = { > count += input.size > out.collect(count) > } > }{code} -- This message was sent by Atlassian Jira (v8.20.10#820010)