You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2022/11/01 09:52:21 UTC
[GitHub] [flink] fredia commented on a diff in pull request #21050: [FLINK-29095][state] Improve logging in SharedStateRegistry
fredia commented on code in PR #21050:
URL: https://github.com/apache/flink/pull/21050#discussion_r1010263764
##########
flink-runtime/src/main/java/org/apache/flink/runtime/state/SharedStateRegistryImpl.java:
##########
@@ -95,60 +95,78 @@ public StreamStateHandle registerReference(
entry = registeredStates.get(registrationKey);
if (entry == null) {
- // Additional check that should never fail, because only state handles that are not
- // placeholders should
- // ever be inserted to the registry.
checkState(
- !isPlaceholder(state),
+ !isPlaceholder(newHandle),
"Attempt to reference unknown state: " + registrationKey);
- entry = new SharedStateEntry(state, checkpointID);
+ LOG.trace(
+ "Registered new shared state {} under key {}.", newHandle, registrationKey);
+ entry = new SharedStateEntry(newHandle, checkpointID);
registeredStates.put(registrationKey, entry);
- LOG.trace("Registered new shared state {} under key {}.", entry, registrationKey);
- } else {
- // Delete if this is a real duplicate.
- // Note that task (backend) is not required to re-upload state
- // if the confirmation notification was missing.
- // However, it's also not required to use exactly the same handle or placeholder
- if (!Objects.equals(state, entry.stateHandle)) {
- if (entry.confirmed || isPlaceholder(state)) {
- scheduledStateDeletion = state;
- } else {
- // Old entry is not in a confirmed checkpoint yet, and the new one differs.
- // This might result from (omitted KG range here for simplicity):
- // 1. Flink recovers from a failure using a checkpoint 1
- // 2. State Backend is initialized to UID xyz and a set of SST: { 01.sst }
- // 3. JM triggers checkpoint 2
- // 4. TM sends handle: "xyz-002.sst"; JM registers it under "xyz-002.sst"
- // 5. TM crashes; everything is repeated from (2)
- // 6. TM recovers from CP 1 again: backend UID "xyz", SST { 01.sst }
- // 7. JM triggers checkpoint 3
- // 8. TM sends NEW state "xyz-002.sst"
- // 9. JM discards it as duplicate
- // 10. checkpoint completes, but a wrong SST file is used
- // So we use a new entry and discard the old one:
- scheduledStateDeletion = entry.stateHandle;
- entry.stateHandle = state;
- }
- LOG.trace(
- "Identified duplicate state registration under key {}. New state {} was determined to "
- + "be an unnecessary copy of existing state {} and will be dropped.",
- registrationKey,
- state,
- entry.stateHandle);
- }
+ // no further handling
+ return entry.stateHandle;
+
+ } else if (entry.stateHandle == newHandle) {
+ // might be a bug but state backend is not required to use a place-holder
+ LOG.debug(
+ "Duplicated registration under key {} with the same object: {}",
+ registrationKey,
+ newHandle);
+ } else if (Objects.equals(entry.stateHandle, newHandle)) {
+ // might be a bug but state backend is not required to use a place-holder
+ LOG.debug(
+ "Duplicated registration under key {} with the new object: {}.",
+ registrationKey,
+ newHandle);
+ } else if (isPlaceholder(newHandle)) {
LOG.trace(
- "Updating last checkpoint for {} from {} to {}",
+ "Duplicated registration under key {} with a placeholder (normal case)",
+ registrationKey);
+ scheduledStateDeletion = newHandle;
+ } else if (entry.confirmed) {
+ LOG.info(
+ "Duplicated registration under key {} of a new state: {}. "
+ + "This might happen if checkpoint confirmation was delayed and state backend re-uploaded the state. "
+ + "Discarding the new state and keeping the old one which is included into a completed checkpoint",
registrationKey,
- entry.lastUsedCheckpointID,
- checkpointID);
- entry.advanceLastUsingCheckpointID(checkpointID);
- if (preventDiscardingCreatedCheckpoint) {
- entry.preventDiscardingCreatedCheckpoint();
- }
+ newHandle);
+ scheduledStateDeletion = newHandle;
+ } else {
+ // Old entry is not in a confirmed checkpoint yet, and the new one differs.
+ // This might result from (omitted KG range here for simplicity):
+ // 1. Flink recovers from a failure using a checkpoint 1
+ // 2. State Backend is initialized to UID xyz and a set of SST: { 01.sst }
+ // 3. JM triggers checkpoint 2
+ // 4. TM sends handle: "xyz-002.sst"; JM registers it under "xyz-002.sst"
+ // 5. TM crashes; everything is repeated from (2)
+ // 6. TM recovers from CP 1 again: backend UID "xyz", SST { 01.sst }
+ // 7. JM triggers checkpoint 3
+ // 8. TM sends NEW state "xyz-002.sst"
+ // 9. JM discards it as duplicate
+ // 10. checkpoint completes, but a wrong SST file is used
+ // So we use a new entry and discard the old one:
+ LOG.info(
Review Comment:
Sorry for the late reply, I add a UT to verify the concurrent checkpoints logic. https://github.com/fredia/flink/tree/test_concurrent_chk
> the later complete checkpoint will reuse the previous state handle, the logic is [here](https://github.com/apache/flink/blob/8e16cc8e424e352c5b45b46f1520ecf0edec70be/flink-runtime/src/main/java/org/apache/flink/runtime/state/IncrementalRemoteKeyedStateHandle.java#L345)
I found that the previous state handle is discarded when registering a new state handle with same key ([line 260](https://github.com/apache/flink/compare/master...fredia:flink:test_concurrent_chk#diff-80eee7531d8a093be7d2f4cb69dce47453e22e898662c930fb7557908efbebc4R260)), so the later complete checkpoint would use its **own** state handles not previous handles.
The previous checkpoint's state handles are discarded, although all of the checkpoints may succeed, the succeed checkpoints may be broken. So I think this is a bug.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org