featzhang commented on code in PR #27814:
URL: https://github.com/apache/flink/pull/27814#discussion_r3008537747
##########
flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackendSnapshotStrategy.java:
##########
@@ -122,6 +122,21 @@ public SnapshotResultSupplier<OperatorStateHandle>
asyncSnapshot(
Map<String, BackendWritableBroadcastState<?, ?>>
registeredBroadcastStatesDeepCopies =
syncPartResource.getRegisteredBroadcastStatesDeepCopies();
+ if (registeredBroadcastStatesDeepCopies.isEmpty()
+ &&
hasOnlyEmptyOperatorListStates(registeredOperatorStatesDeepCopies)) {
+ if (streamFactory instanceof FsMergingCheckpointStorageLocation) {
Review Comment:
After casting to FsMergingCheckpointStorageLocation, it directly returns
EmptyFileMergingOperatorStreamStateHandle. Add a comment explaining why empty
snapshots take this branch.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/state/OperatorStateBackendTest.java:
##########
@@ -489,6 +490,124 @@ void testFileMergingSnapshotEmpty(@TempDir File
tmpFolder) throws Exception {
assertThat(stateHandle).isInstanceOf(FileMergingOperatorStreamStateHandle.class);
}
+ @Test
+ void testSnapshotWithEmptyRegisteredOperatorState() throws Exception {
+ final AbstractStateBackend abstractStateBackend = new
HashMapStateBackend();
+ CloseableRegistry cancelStreamRegistry = new CloseableRegistry();
+
+ Environment env = createMockEnvironment();
+ final OperatorStateBackend operatorStateBackend =
+ abstractStateBackend.createOperatorStateBackend(
+ new OperatorStateBackendParametersImpl(
+ env, "testOperator", emptyStateHandles,
cancelStreamRegistry));
+
+ ListStateDescriptor<Integer> stateDescriptor =
+ new ListStateDescriptor<>("test-empty-list-state",
IntSerializer.INSTANCE);
+ try {
+ ListState<Integer> listState =
operatorStateBackend.getListState(stateDescriptor);
+ assertThat(listState.get()).isEmpty();
+
+ CheckpointStreamFactory streamFactory = new
MemCheckpointStreamFactory(4096);
+
+ RunnableFuture<SnapshotResult<OperatorStateHandle>> snapshot =
+ operatorStateBackend.snapshot(
+ 0L,
+ 0L,
+ streamFactory,
+
CheckpointOptions.forCheckpointWithDefaultLocation());
+
+ SnapshotResult<OperatorStateHandle> snapshotResult =
+ FutureUtils.runIfNotDoneAndGet(snapshot);
+ OperatorStateHandle stateHandle =
snapshotResult.getJobManagerOwnedSnapshot();
+ assertThat(stateHandle).isNull();
+ } finally {
+ operatorStateBackend.close();
+ operatorStateBackend.dispose();
+ }
+ }
+
+ @Test
+ void testFileMergingSnapshotRestoreWithEmptyRegisteredUnionState(@TempDir
File tmpFolder)
Review Comment:
In testFileMergingSnapshotRestoreWithEmptyRegisteredUnionState, the finally
block only calls discardState() on stateHandle. Consider additionally verifying
that FsMerging-related resources (temp files, snapshotManager) are correctly
released to prevent leaks.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackendSnapshotStrategy.java:
##########
@@ -122,6 +122,21 @@ public SnapshotResultSupplier<OperatorStateHandle>
asyncSnapshot(
Map<String, BackendWritableBroadcastState<?, ?>>
registeredBroadcastStatesDeepCopies =
syncPartResource.getRegisteredBroadcastStatesDeepCopies();
+ if (registeredBroadcastStatesDeepCopies.isEmpty()
+ &&
hasOnlyEmptyOperatorListStates(registeredOperatorStatesDeepCopies)) {
+ if (streamFactory instanceof FsMergingCheckpointStorageLocation) {
+ FsMergingCheckpointStorageLocation location =
+ (FsMergingCheckpointStorageLocation) streamFactory;
+ return snapshotCloseableRegistry ->
+ SnapshotResult.of(
+
EmptyFileMergingOperatorStreamStateHandle.create(
+ location.getExclusiveStateHandle(),
+ location.getSharedStateHandle()));
+ } else {
+ return snapshotCloseableRegistry -> SnapshotResult.empty();
Review Comment:
The else branch returns SnapshotResult.empty(). Add a comment explaining
that this is the handling strategy for empty snapshots when not using
FsMerging, for clarity.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/state/OperatorStateBackendTest.java:
##########
@@ -489,6 +490,124 @@ void testFileMergingSnapshotEmpty(@TempDir File
tmpFolder) throws Exception {
assertThat(stateHandle).isInstanceOf(FileMergingOperatorStreamStateHandle.class);
}
+ @Test
+ void testSnapshotWithEmptyRegisteredOperatorState() throws Exception {
Review Comment:
In testSnapshotWithEmptyRegisteredOperatorState, currently asserts
stateHandle is null. Also assert snapshotResult.isEmpty() or its type to ensure
empty snapshot behavior is as expected.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/state/OperatorStateBackendTest.java:
##########
@@ -489,6 +490,124 @@ void testFileMergingSnapshotEmpty(@TempDir File
tmpFolder) throws Exception {
assertThat(stateHandle).isInstanceOf(FileMergingOperatorStreamStateHandle.class);
}
+ @Test
+ void testSnapshotWithEmptyRegisteredOperatorState() throws Exception {
+ final AbstractStateBackend abstractStateBackend = new
HashMapStateBackend();
+ CloseableRegistry cancelStreamRegistry = new CloseableRegistry();
+
+ Environment env = createMockEnvironment();
+ final OperatorStateBackend operatorStateBackend =
+ abstractStateBackend.createOperatorStateBackend(
+ new OperatorStateBackendParametersImpl(
+ env, "testOperator", emptyStateHandles,
cancelStreamRegistry));
+
+ ListStateDescriptor<Integer> stateDescriptor =
+ new ListStateDescriptor<>("test-empty-list-state",
IntSerializer.INSTANCE);
+ try {
+ ListState<Integer> listState =
operatorStateBackend.getListState(stateDescriptor);
+ assertThat(listState.get()).isEmpty();
+
+ CheckpointStreamFactory streamFactory = new
MemCheckpointStreamFactory(4096);
+
+ RunnableFuture<SnapshotResult<OperatorStateHandle>> snapshot =
+ operatorStateBackend.snapshot(
+ 0L,
+ 0L,
+ streamFactory,
+
CheckpointOptions.forCheckpointWithDefaultLocation());
+
+ SnapshotResult<OperatorStateHandle> snapshotResult =
+ FutureUtils.runIfNotDoneAndGet(snapshot);
+ OperatorStateHandle stateHandle =
snapshotResult.getJobManagerOwnedSnapshot();
+ assertThat(stateHandle).isNull();
+ } finally {
+ operatorStateBackend.close();
+ operatorStateBackend.dispose();
+ }
+ }
+
+ @Test
+ void testFileMergingSnapshotRestoreWithEmptyRegisteredUnionState(@TempDir
File tmpFolder)
+ throws Exception {
+ final AbstractStateBackend abstractStateBackend = new
HashMapStateBackend();
+ CloseableRegistry cancelStreamRegistry = new CloseableRegistry();
+
+ Environment env = createMockEnvironment();
+ OperatorStateBackend operatorStateBackend =
+ abstractStateBackend.createOperatorStateBackend(
+ new OperatorStateBackendParametersImpl(
+ env, "testOperator", emptyStateHandles,
cancelStreamRegistry));
+
+ ListStateDescriptor<Integer> stateDescriptor =
+ new ListStateDescriptor<>("test-empty-union-state",
IntSerializer.INSTANCE);
+ ListState<Integer> unionState =
operatorStateBackend.getUnionListState(stateDescriptor);
+ assertThat(unionState.get()).isEmpty();
+
+ Path checkpointBaseDir = new Path(tmpFolder.toString());
+ Path sharedStateDir =
+ new Path(
+ checkpointBaseDir,
+
AbstractFsCheckpointStorageAccess.CHECKPOINT_SHARED_STATE_DIR);
+ Path taskOwnedStateDir =
+ new Path(
+ checkpointBaseDir,
+
AbstractFsCheckpointStorageAccess.CHECKPOINT_TASK_OWNED_STATE_DIR);
+
+ JobID jobId = JobID.generate();
+ final FileMergingSnapshotManager.SubtaskKey subtaskKey =
+ new FileMergingSnapshotManager.SubtaskKey(jobId.toHexString(),
"opId", 1, 1);
+ LocalFileSystem fs = getSharedInstance();
+ CheckpointStorageLocationReference cslReference =
+ AbstractFsCheckpointStorageAccess.encodePathAsReference(
+ fromLocalFile(fs.pathToFile(checkpointBaseDir)));
+ FileMergingSnapshotManager snapshotManager =
+ createFileMergingSnapshotManager(
+ checkpointBaseDir, sharedStateDir, taskOwnedStateDir,
subtaskKey);
+ CheckpointStreamFactory streamFactory =
+ new FsMergingCheckpointStorageLocation(
+ subtaskKey,
+ fs,
+ checkpointBaseDir,
+ sharedStateDir,
+ taskOwnedStateDir,
+ cslReference,
+ 1024,
+ 1024,
+ snapshotManager,
+ 0);
+
+ OperatorStateHandle stateHandle = null;
+ try {
+ RunnableFuture<SnapshotResult<OperatorStateHandle>> snapshot =
+ operatorStateBackend.snapshot(
+ 0L,
+ 0L,
+ streamFactory,
+
CheckpointOptions.forCheckpointWithDefaultLocation());
+
+ SnapshotResult<OperatorStateHandle> snapshotResult =
+ FutureUtils.runIfNotDoneAndGet(snapshot);
+ stateHandle = snapshotResult.getJobManagerOwnedSnapshot();
+
+
assertThat(stateHandle).isInstanceOf(EmptyFileMergingOperatorStreamStateHandle.class);
Review Comment:
Asserting stateHandle type as EmptyFileMergingOperatorStreamStateHandle is
correct. Add a comment explaining this is a special case for empty operator
state to improve readability.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
