You are right Aaron. I would say this is like this by design as Flink doesn't require you to initialize state in the open method so it has no safe way to delete the non-referenced ones.
What you can do is restore the state and clear it on all operators and not reference it again. I know this feels like a workaround but I have no better idea at the moment. Cheers, Gyula On Wed, Nov 27, 2019 at 6:08 PM Aaron Levin <aaronle...@stripe.com> wrote: > Hi, > > Yes, we're using UNION state. I would assume, though, that if you are > not reading the UNION state it would either stop stick around as a > constant factor in your state size, or get cleared. > > Looks like I should try to recreate a small example and submit a bug > if this is true. Otherwise it's impossible to remove union state from > your operators. > > On Wed, Nov 27, 2019 at 6:50 AM Congxian Qiu <qcx978132...@gmail.com> > wrote: > > > > Hi > > > > Do you use UNION state in your scenario, when using UNION state, then JM > may encounter OOM because each TDD will contains all the state of all > subtasks[1] > > > > [1] > https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/state/state.html#using-managed-operator-state > > Best, > > Congxian > > > > > > Aaron Levin <aaronle...@stripe.com> 于2019年11月27日周三 上午3:55写道： > >> > >> Hi, > >> > >> Some context: after a refactoring, we were unable to start our jobs. > >> They started fine and checkpointed fine, but once the job restarted > >> owing to a transient failure, the application was unable to start. The > >> Job Manager was OOM'ing (even when I gave them 256GB of ram!). The > >> `_metadata` file for the checkpoint was 1.3GB (usually 11MB). Inside > >> the `_metadata` file we saw `- 1402496 offsets: > >> com.stripe.flink.backfill.kafka-archive-file-progress`. This happened > >> to be the operator state we were no longer initializing or > >> snapshotting after the refactoring. > >> > >> Before I dig further into this and try to find a smaller reproducible > >> test case I thought I would ask if someone knows what the expected > >> behaviour is for the following scenario: > >> > >> suppose you have an operator (in this case a Source) which has some > >> operator ListState. Suppose you run your flink job for some time and > >> then later refactor your job such that you no longer use that state > >> (so after the refactoring you're no longer initializing this operator > >> state in initializeState, nor are you snapshotting the operator state > >> in snapshotState). If you launch your new code from a recent > >> savepoint, what do we expect to happen to the state? Do we anticipate > >> the behaviour I explained above? > >> > >> My assumption would be that Flink would not read this state and so it > >> would be removed from the next checkpoint or savepoint. Alternatively, > >> I might assume it would not be read but would linger around every > >> future checkpoint or savepoint. However, it feels like what is > >> happening is it's not read and then possibly replicated by every > >> instance of the task every time a checkpoint happens (hence the > >> accidentally exponential behaviour). > >> > >> Thoughts? > >> > >> PS - in case someone asks: I was sure that we were calling `.clear()` > >> appropriately in `snapshotState` (we, uh, already learned that lesson > >> :D) > >> > >> Best, > >> > >> Aaron Levin >
