[FLINK-33341][state] Add support for rescaling from local keyed state #23591
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
StefanRRichter
force-pushed
the
srichter-local-rescaling-FLINK-33341
branch
from
3d2cd24
to
27090a0
Compare
StefanRRichter
force-pushed
the
srichter-local-rescaling-FLINK-33341
branch
from
27090a0
to
b8b8c48
Compare
StefanRRichter
changed the title
StefanRRichter
force-pushed
the
srichter-local-rescaling-FLINK-33341
branch
from
b8b8c48
to
aa59dc1
Compare
rkhachatryan
approved these changes
Oct 26, 2023
There was a problem hiding this comment.
Nice changes @StefanRRichter!
I've only spotted one small issue with equals, other things are mostly NITs.
Other than that LGTM.
Comment on lines
115
to
118
| public int hashCode() { | ||
| int result = super.hashCode(); | ||
| result = 31 * result + getMetaDataState().hashCode(); | ||
| int result = directoryStateHandle.hashCode(); | ||
| result = 31 * result + getKeyGroupRange().hashCode(); | ||
| result = 31 * result + getMetaDataStateHandle().hashCode(); |
There was a problem hiding this comment.
- Doesn't
equalsneed to be updated as well? I think it's currently doesn't matchhashCode - nit: move closer to
equals?
There was a problem hiding this comment.
Good catch, I think the hashCode/equals/toString in those classes should be consolidated. I'll create a separate commit to streamline them.
...k-runtime/src/main/java/org/apache/flink/runtime/state/IncrementalLocalKeyedStateHandle.java
Show resolved
Hide resolved
| * | ||
| * @param jobManagerState the state that the task got assigned from the job manager (this | ||
| * state lives in remote storage). | ||
| * @param alternativesByPriority local alternatives to the job manager starte, ordered by |
| * | ||
| * @param jobManagerState the state that the task got assigned from the job manager (this | ||
| * state lives in remote storage). | ||
| * @param alternativesByPriority local alternatives to the job manager starte, ordered by |
...e/src/test/java/org/apache/flink/runtime/checkpoint/PrioritizedOperatorSubtaskStateTest.java
Show resolved
Hide resolved
...e/src/test/java/org/apache/flink/runtime/checkpoint/PrioritizedOperatorSubtaskStateTest.java
Outdated
Show resolved
Hide resolved
… IncrementalKeyedStateHandle.
StefanRRichter
force-pushed
the
srichter-local-rescaling-FLINK-33341
branch
2 times, most recently
from
ca38a5e
to
31b79ef
Compare
…g for incremental state handle classes.
… to RockDBIncrementalRestoreOperation.
… to PrioritizedOperatorSubtaskState.
StefanRRichter
force-pushed
the
srichter-local-rescaling-FLINK-33341
branch
from
31b79ef
to
7013a86
Compare
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
What is the purpose of the change
This PR extends the use case for local (keyed) state from recovery to rescaling. Previously, Flink cannot make use of relevant local state in rescaling scenarios, even if such state would be available. For example, when scaling our we will often find already a superset of the state after scale out in the previously existing slots. And for scaling in, we should find a subset of the state in the existing slot. Utilizing such existing state can reduce the download from remote storage in rescaling scenarios.
Brief change log
AutoRescalingITCaseas a variant ofRescalingITCase, that uses checkpoint & changing resource requirements to trigger a rescale that preservers local state (like an autoscaler would), instead of using savepoint & restart (losing local state).AbstractIncrementalStateHandleas common superclass toIncrementalRemoteKeyedStateHandleandIncrementalLocalKeyedStateHandle. See next change for reason.RocksDBIncrementalRestoreOperationin such way that we can transparently rescale fromIncrementalRemoteKeyedStateHandleandIncrementalLocalKeyedStateHandle(or a mix of both).PrioritizedOperatorSubtaskStateto support creating an alternative as a mix of local and remote (=assigned from job manager) state, instead of only alternatives that are either purely local ore purely remote.Verifying this change
This change added tests:
AutoRescalingITCasePrioritizedOperatorSubtaskStateTestDoes this pull request potentially affect one of the following parts:
@Public(Evolving): (no)Documentation