[GitHub] [flink] becketqin commented on a change in pull request #12137: [FLINK-16177][FLINK-16357][checkpointing] Complete / improve checkpointing for OperatorCoordinators

[GitBox <gi...@apache.org>]

becketqin commented on a change in pull request #12137:
URL: https://github.com/apache/flink/pull/12137#discussion_r426112436



##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV2V3SerializerBase.java
##########
@@ -468,6 +468,7 @@ public static void serializeStreamStateHandle(
 		dos.flush();
 	}
 
+	@Nullable

Review comment:
       Minor: another option is to return an `Optional`. Currently we have both used in mix. It might be good to have a convention.

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV2V3SerializerBase.java
##########
@@ -487,6 +488,15 @@ public static StreamStateHandle deserializeStreamStateHandle(DataInputStream dis
 		}
 	}
 
+	public static ByteStreamStateHandle deserializeAndCheckByteStreamStateHandle(DataInputStream dis) throws IOException {

Review comment:
       Nit: It seems the return value is also `Nullable`?

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DefaultScheduler.java
##########
@@ -186,10 +188,18 @@ private void maybeHandleTaskFailure(final TaskExecutionState taskExecutionState,
 
 	private void handleTaskFailure(final ExecutionVertexID executionVertexId, @Nullable final Throwable error) {
 		setGlobalFailureCause(error);
+		notifyCoordinatorsAboutTaskFailure(executionVertexId, error);

Review comment:
       It looks that the failover may initially restart tasks only and fall back to a global failover if the task-only failover encountered exception. This may cause the `OperatorCoordinator.subtaskFailed()` to be invoked twice consecutively on the same subtask. Currently `SourceCoordinator` throws an exception if `subtaskFailed()` is invoked on a subtask that has not been registered. We may need to remove that check.




----------------------------------------------------------------
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org