Francis created FLINK-41424:
-------------------------------
Summary: ForSt: support dropping keyed state for removed state
descriptors
Key: FLINK-41424
URL: https://issues.apache.org/jira/browse/FLINK-41424
Project: Flink
Issue Type: Improvement
Components: Runtime / State Backends
Affects Versions: 2.2.1
Reporter: Francis
h3. Problem
Removing a state descriptor from a keyed operator does not remove its data. The
column family is restored from the checkpoint and carried into every future
checkpoint, even though no code can reach it.
We run a Flink 2.2.1 job on ForSt with the async state API, incremental
checkpoints and CLAIM restores. Over time we removed three {{ValueState}}
descriptors from a single keyed operator. Around 50% of our checkpoint file
size was state from those removed descriptors.
h3. Existing options
None of these fix it:
* {{allowNonRestoredState}} only works per operator.
* The State Processor API means stopping the job and rewriting a savepoint, and
it does not support ForSt.
* {{applyToAllKeys}} is not available on the async ForSt backend.
* Adding a TTL to the existing state needs a state migration, which async ForSt
does not support yet.
h3. Current workaround
We re-register the removed descriptors in {{open()}} and run Flink's own
rescale code on just those column families, with an empty target key group
range:
{code:java}
val states = deadStateNames.map {
runtimeContext.getState(ValueStateDescriptor(it,
BYTE_PRIMITIVE_ARRAY_TYPE_INFO))
}.filterIsInstance<ForStValueState<*, *, *>>()
// AsyncExecutionController#switchContextListener is the keyed backend
(package-private)
val backend = readField(
states.first().stateRequestHandler,
AsyncExecutionController::class.java,
"switchContextListener",
) as ForStKeyedStateBackend<*>
// ForStKeyedStateBackend#db is protected
val db = readField(backend, ForStKeyedStateBackend::class.java, "db") as RocksDB
ForStIncrementalCheckpointUtils.clipDBWithKeyGroupRange(
db,
states.map { it.columnFamilyHandle },
KeyGroupRange.EMPTY_KEY_GROUP_RANGE,
backend.keyGroupRange,
computeRequiredBytesInKeyGroupPrefix(runtimeContext.taskInfo.maxNumberOfParallelSubtasks),
true,
)
{code}
This works, but it needs two reflective field reads and depends on internal
classes. It also only empties the column families. They stay registered in the
backend.
h3. Proposal
A supported way to drop keyed state that no longer has a descriptor, without
stopping the job to rewrite a savepoint. Two possible shapes:
# An explicit API to drop a keyed state by name or descriptor.
# Extend {{ignore-unclaimed-state}} to state inside an operator (FLIP-359):
column families restored from the checkpoint but not registered after
{{open()}} are dropped.
On ForSt both can reuse existing code: {{dropColumnFamily}}, or
{{clipDBWithKeyGroupRange}} with an empty target range as above.
Checkpoints taken after the restore should no longer reference the dropped
state, and its files should be deleted once no retained checkpoint references
them.
h3. Related
* FLINK-32658 / FLIP-359: same leak, proposed for heap and RocksDB only
* FLINK-10121: remove registered operator states
* FLINK-37686: needed for {{deleteFilesInRanges}} on restored (NOT_OWNED) files
* User ML thread:
https://lists.apache.org/thread/s10jzzqb0xrgzcs741d2mfwbr90q87lf
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
