[
https://issues.apache.org/jira/browse/FLINK-32658?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17755792#comment-17755792
]
Hangxiang Yu commented on FLINK-32658:
--------------------------------------
Thanks a lot for the proposal, I think it makes sense.
BTW, I saw FLINK-10121 also talk something related to this so just linked it.
Just as discussed offline, I think it's better to consider operator state /
corrdinater state / keyed state together which could make the interface more
clear.
And seems we have to modify the semantics of `ignore-unclaimed-state` and
modify/add new interface in the public API, I think we may need a FLIP firstly.
A blocker is that we could not make sure that users only create state in the
open method. I think it's a seprate topic but I'm big +1 to limit the user
operation to avoid creating state when processElement:
1. We have seen many cases users createing state when processElement which
often bring up many strange problems, e.g. error-prone judging null logic, job
stuck in schema evolution even if job is running... Actually, they could move
this to open method in all cases.
2. It's better to provide declartion info to state backend early which could
make state backend do some previous optimization. This is also described in
FLIP-22.
3. I also saw all SQL operators creating state in the open method, so I think
it works well for most even all cases.
Of course, it may break changes, it's better to be considered in Flink 2.0
> State should not be silently removed when ignore-unclaimed-state is false
> -------------------------------------------------------------------------
>
> Key: FLINK-32658
> URL: https://issues.apache.org/jira/browse/FLINK-32658
> Project: Flink
> Issue Type: Bug
> Components: Runtime / Checkpointing
> Affects Versions: 1.18.0, 1.17.1
> Reporter: Rui Fan
> Assignee: Rui Fan
> Priority: Major
> Labels: pull-request-available
> Attachments: screenshot-1.png
>
>
> When ignore-unclaimed-state is false and the old state is removed, flink
> should throw exception. It's similar to removing a stateful operator.
> This case occurs not only when the user removes state, but also when the
> operator is replaced.
> For example: upgrade FlinkKafkaConsumer to KafkaSource. All logical are not
> changed, so the operator id isn't changed. The KafkaSource cannot resume from
> the state of FlinkKafkaConsumer. However, the new flink job can start, and
> the state is silently removed in the new job.(The old state is not physically
> discarded, it is still stored in the state backend, but the new code will
> never use it.)
> It also brings an additional problem: the KafkaSource will snapshot 2 states,
> it includes the new state of KafkaSource, and the union list state of
> FlinkKafkaConsumer. Whenever a job resumes from checkpoint, the union List
> state is inflated. Eventually the state size of kafka offset exceeded 200MB.
> !screenshot-1.png!
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
