Hi all,
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, but their column families never went
away — the data continued to be carried over into all future checkpoints. In
our specific case this mean that around 50% of our checkpoint file size was
unmapped state.
None of the options we found seemed to fix this:
- allowNonRestoredState only works per operator.
- The State Processor API means stopping the job and rewriting a savepoint, and
it doesn't support ForSt.
- applyToAllKeys isn't available on the async ForSt backend.
- Adding a TTL to the existing state needs a state migration, which async ForSt
doesn't support yet
What we ended up doing is re-registering the removed descriptors in open() and
running Flink's own rescale code on just those column families. In order to
make this work we needed to get a handle on the DB insrance requiring
reflective reads and a bit of hacky code:
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
With these handles we could use the existing
ForStIncrementalCheckpointUtils.clipDBWithKeyGroupRange:
ForStIncrementalCheckpointUtils.clipDBWithKeyGroupRange(
db,
states.map { it.columnFamilyHandle },
KeyGroupRange.EMPTY_KEY_GROUP_RANGE,
backend.keyGroupRange,
computeRequiredBytesInKeyGroupPrefix(runtimeContext.taskInfo.maxNumberOfParallelSubtasks),
true,
)
Related to this, I have a few questions:
1. Is there a supported way to do this that we missed?
2. Are there pitfalls with deleteFilesInRanges on restored (NOT_OWNED) files
beyond FLINK-37686 <https://github.com/apache/flink/pull/28801>?
Thanks a lot,
Francis
