Search before asking

• I searched in the issues and found nothing similar.

Fluss version

0.5.0

Minimal reproduce step

Create a partitioned table, and using Flink to read the table

What doesn't meet your expectations?

It should throw exeption

Anything else?

1: currently, the logic of check partition changes is wrong, we should deduplicate the partitions from the splits assigned or pending
2: If the partitions should be deleted, the log scanner may has not buckets to substribe, it shouldn't throw exception when fetch

Are you willing to submit a PR?

• I'm willing to submit a PR!