alibaba / alibaba/feathub

Failed to restore state when open checkpoint option

Open
#256 0 comments 0 reactions 0 assignees View on GitHub
Dominant language
Python
Stars
350
Forks
60
PR merge metrics
No merged PRs in 30d

Description

Thanks for you work, it seems SlidingWindow doesn't support restore state from CheckPoint
```
2023-10-07 16:22:31
java.io.IOException: Exception while applying AggregateFunction in aggregating state
at org.apache.flink.runtime.state.heap.HeapAggregatingState.add(HeapAggregatingState.java:107)
at org.apache.flink.streaming.runtime.operators.windowing.WindowOperator.processElement(WindowOperator.java:404)
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:519)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:203)
at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:804)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:753)
at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:963)
at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:942)
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:756)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:578)
at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.RuntimeException: Serializer does not support named field positions.
at org.apache.flink.api.java.typeutils.runtime.RowSerializer.copyNameBased(RowSerializer.java:173)
at org.apache.flink.api.java.typeutils.runtime.RowSerializer.copy(RowSerializer.java:144)
at org.apache.flink.api.java.typeutils.runtime.RowSerializer.copy(RowSerializer.java:72)
at org.apache.flink.runtime.state.heap.CopyOnWriteStateMap.transform(CopyOnWriteStateMap.java:375)
at org.apache.flink.runtime.state.heap.StateTable.transform(StateTable.java:211)
at org.apache.flink.runtime.state.heap.HeapAggregatingState.add(HeapAggregatingState.java:105)
... 14 more

```
I guess it is caused by we created an `name-based` row in aggregation function

[SlidingWindowUtils.java#L227](https://github.com/alibaba/feathub/blob/master/java/feathub-udf/flink-udf/src/main/java/com/alibaba/feathub/flink/udf/SlidingWindowUtils.java#L227)

but flink row serializer doesn't support name-based row.

[RowSerializer.java#L170](https://github.com/apache/flink/blob/master/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/RowSerializer.java#L170)

Contributor guide

No contributing guide indexed for this repository

Research direction

Start with SlidingWindowUtils.java at the linked aggregation-function location, then inspect the reported restore path through HeapAggregatingState and RowSerializer. Reproduce the checkpoint restore failure for SlidingWindow and verify that state can be restored and processing continues without the named-field serializer exception.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
stream-processing
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.