[flink] 02/04: [refactor][test] Introduce DiscardRecordedStateObject to avoid using mocks for state handles
This is an automated email from the ASF dual-hosted git repository. roman pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git commit 4467ad05bb9823d90be770d5e1c36989f3905d1b Author: wangfeifan AuthorDate: Mon Jul 24 14:44:02 2023 +0800 [refactor][test] Introduce DiscardRecordedStateObject to avoid using mocks for state handles --- .../checkpoint/CheckpointCoordinatorTest.java | 47 -- .../checkpoint/metadata/CheckpointTestUtils.java | 13 +++--- ...Handle.java => DiscardRecordedStateObject.java} | 36 +++-- .../IncrementalRemoteKeyedStateHandleTest.java | 36 ++--- ...le.java => TestingRelativeFileStateHandle.java} | 22 +- .../runtime/state/TestingStreamStateHandle.java| 8 +++- 6 files changed, 83 insertions(+), 79 deletions(-) diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java index 2f274dab3d8..79752468045 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java @@ -50,6 +50,7 @@ import org.apache.flink.runtime.state.CheckpointMetadataOutputStream; import org.apache.flink.runtime.state.CheckpointStorage; import org.apache.flink.runtime.state.CheckpointStorageAccess; import org.apache.flink.runtime.state.CheckpointStorageLocation; +import org.apache.flink.runtime.state.DiscardRecordedStateObject; import org.apache.flink.runtime.state.IncrementalKeyedStateHandle.HandleAndLocalPath; import org.apache.flink.runtime.state.IncrementalRemoteKeyedStateHandle; import org.apache.flink.runtime.state.KeyGroupRange; @@ -75,6 +76,7 @@ import org.apache.flink.testutils.TestingUtils; import org.apache.flink.testutils.executor.TestExecutorExtension; import org.apache.flink.testutils.junit.utils.TempDirUtils; import org.apache.flink.util.ExceptionUtils; +import org.apache.flink.util.TernaryBoolean; import org.apache.flink.util.TestLogger; import org.apache.flink.util.concurrent.FutureUtils; import org.apache.flink.util.concurrent.ManuallyTriggeredScheduledExecutor; @@ -137,7 +139,6 @@ import static org.assertj.core.api.Assertions.fail; import static org.mockito.ArgumentMatchers.anyList; import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.Matchers.any; -import static org.mockito.Mockito.atLeast; import static org.mockito.Mockito.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; @@ -2942,13 +2943,15 @@ class CheckpointCoordinatorTest extends TestLogger { streamStateHandle instanceof PlaceholderStreamStateHandle) .isFalse(); -verify(streamStateHandle, never()).discardState(); +DiscardRecordedStateObject.verifyDiscard( +streamStateHandle, TernaryBoolean.FALSE); ++sharedHandleCount; } for (HandleAndLocalPath handleAndLocalPath : incrementalKeyedStateHandle.getPrivateState()) { -verify(handleAndLocalPath.getHandle(), never()).discardState(); +DiscardRecordedStateObject.verifyDiscard( +handleAndLocalPath.getHandle(), TernaryBoolean.FALSE); } verify(incrementalKeyedStateHandle.getMetaStateHandle(), never()) @@ -2971,7 +2974,8 @@ class CheckpointCoordinatorTest extends TestLogger { // by CP1 for (List cpList : sharedHandlesByCheckpoint) { for (HandleAndLocalPath handleAndLocalPath : cpList) { -verify(handleAndLocalPath.getHandle(), never()).discardState(); +DiscardRecordedStateObject.verifyDiscard( +handleAndLocalPath.getHandle(), TernaryBoolean.FALSE); } } @@ -3030,14 +3034,19 @@ class CheckpointCoordinatorTest extends TestLogger { // references the state from CP1, so we expect they are not discarded. verifyDiscard( sharedHandlesByCheckpoint, -cpId -> restoreMode == RestoreMode.CLAIM && cpId == 0 ? times(1) : never()); +cpId -> +restoreMode == RestoreMode.CLAIM && cpId == 0 +? TernaryBoolean.TRUE +: TernaryBoolean.FALSE); // discard CP2
[flink] 02/04: [refactor][test] Introduce DiscardRecordedStateObject to avoid using mocks for state handles
This is an automated email from the ASF dual-hosted git repository. roman pushed a commit to branch release-1.16 in repository https://gitbox.apache.org/repos/asf/flink.git commit d00e2eb73aa5f2d2506370b5fbe9c8867cbec987 Author: wangfeifan AuthorDate: Mon Jul 24 14:44:02 2023 +0800 [refactor][test] Introduce DiscardRecordedStateObject to avoid using mocks for state handles --- .../checkpoint/CheckpointCoordinatorTest.java | 47 -- .../checkpoint/metadata/CheckpointTestUtils.java | 13 +++--- ...Handle.java => DiscardRecordedStateObject.java} | 36 +++-- .../IncrementalRemoteKeyedStateHandleTest.java | 36 ++--- ...le.java => TestingRelativeFileStateHandle.java} | 22 +- .../runtime/state/TestingStreamStateHandle.java| 8 +++- 6 files changed, 83 insertions(+), 79 deletions(-) diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java index 6fa95fb5380..4eedbc60134 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java @@ -50,6 +50,7 @@ import org.apache.flink.runtime.state.CheckpointMetadataOutputStream; import org.apache.flink.runtime.state.CheckpointStorage; import org.apache.flink.runtime.state.CheckpointStorageAccess; import org.apache.flink.runtime.state.CheckpointStorageLocation; +import org.apache.flink.runtime.state.DiscardRecordedStateObject; import org.apache.flink.runtime.state.IncrementalKeyedStateHandle.HandleAndLocalPath; import org.apache.flink.runtime.state.IncrementalRemoteKeyedStateHandle; import org.apache.flink.runtime.state.KeyGroupRange; @@ -73,6 +74,7 @@ import org.apache.flink.runtime.testutils.DirectScheduledExecutorService; import org.apache.flink.testutils.TestingUtils; import org.apache.flink.testutils.executor.TestExecutorResource; import org.apache.flink.util.ExceptionUtils; +import org.apache.flink.util.TernaryBoolean; import org.apache.flink.util.TestLogger; import org.apache.flink.util.concurrent.FutureUtils; import org.apache.flink.util.concurrent.ManuallyTriggeredScheduledExecutor; @@ -141,7 +143,6 @@ import static org.junit.Assert.fail; import static org.mockito.ArgumentMatchers.anyList; import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.Matchers.any; -import static org.mockito.Mockito.atLeast; import static org.mockito.Mockito.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; @@ -2940,13 +2941,15 @@ public class CheckpointCoordinatorTest extends TestLogger { streamStateHandle instanceof PlaceholderStreamStateHandle) .isFalse(); -verify(streamStateHandle, never()).discardState(); +DiscardRecordedStateObject.verifyDiscard( +streamStateHandle, TernaryBoolean.FALSE); ++sharedHandleCount; } for (HandleAndLocalPath handleAndLocalPath : incrementalKeyedStateHandle.getPrivateState()) { -verify(handleAndLocalPath.getHandle(), never()).discardState(); +DiscardRecordedStateObject.verifyDiscard( +handleAndLocalPath.getHandle(), TernaryBoolean.FALSE); } verify(incrementalKeyedStateHandle.getMetaStateHandle(), never()) @@ -2969,7 +2972,8 @@ public class CheckpointCoordinatorTest extends TestLogger { // by CP1 for (List cpList : sharedHandlesByCheckpoint) { for (HandleAndLocalPath handleAndLocalPath : cpList) { -verify(handleAndLocalPath.getHandle(), never()).discardState(); +DiscardRecordedStateObject.verifyDiscard( +handleAndLocalPath.getHandle(), TernaryBoolean.FALSE); } } @@ -3027,14 +3031,19 @@ public class CheckpointCoordinatorTest extends TestLogger { // references the state from CP1, so we expect they are not discarded. verifyDiscard( sharedHandlesByCheckpoint, -cpId -> restoreMode == RestoreMode.CLAIM && cpId == 0 ? times(1) : never()); +cpId -> +restoreMode == RestoreMode.CLAIM && cpId == 0 +? TernaryBoolean.TRUE +: TernaryBoolean.FALSE);
[flink] 02/04: [refactor][test] Introduce DiscardRecordedStateObject to avoid using mocks for state handles
This is an automated email from the ASF dual-hosted git repository. roman pushed a commit to branch release-1.17 in repository https://gitbox.apache.org/repos/asf/flink.git commit 46fa6a03062b90c460141aa275fa06347bc876da Author: wangfeifan AuthorDate: Mon Jul 24 14:44:02 2023 +0800 [refactor][test] Introduce DiscardRecordedStateObject to avoid using mocks for state handles --- .../checkpoint/CheckpointCoordinatorTest.java | 47 -- .../checkpoint/metadata/CheckpointTestUtils.java | 13 +++--- ...Handle.java => DiscardRecordedStateObject.java} | 36 +++-- .../IncrementalRemoteKeyedStateHandleTest.java | 36 ++--- ...le.java => TestingRelativeFileStateHandle.java} | 22 +- .../runtime/state/TestingStreamStateHandle.java| 8 +++- 6 files changed, 83 insertions(+), 79 deletions(-) diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java index baf4527ba88..83c82d6acfb 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java @@ -50,6 +50,7 @@ import org.apache.flink.runtime.state.CheckpointMetadataOutputStream; import org.apache.flink.runtime.state.CheckpointStorage; import org.apache.flink.runtime.state.CheckpointStorageAccess; import org.apache.flink.runtime.state.CheckpointStorageLocation; +import org.apache.flink.runtime.state.DiscardRecordedStateObject; import org.apache.flink.runtime.state.IncrementalKeyedStateHandle.HandleAndLocalPath; import org.apache.flink.runtime.state.IncrementalRemoteKeyedStateHandle; import org.apache.flink.runtime.state.KeyGroupRange; @@ -74,6 +75,7 @@ import org.apache.flink.testutils.TestingUtils; import org.apache.flink.testutils.executor.TestExecutorExtension; import org.apache.flink.testutils.junit.utils.TempDirUtils; import org.apache.flink.util.ExceptionUtils; +import org.apache.flink.util.TernaryBoolean; import org.apache.flink.util.TestLogger; import org.apache.flink.util.concurrent.FutureUtils; import org.apache.flink.util.concurrent.ManuallyTriggeredScheduledExecutor; @@ -135,7 +137,6 @@ import static org.assertj.core.api.Assertions.fail; import static org.mockito.ArgumentMatchers.anyList; import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.Matchers.any; -import static org.mockito.Mockito.atLeast; import static org.mockito.Mockito.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; @@ -2940,13 +2941,15 @@ class CheckpointCoordinatorTest extends TestLogger { streamStateHandle instanceof PlaceholderStreamStateHandle) .isFalse(); -verify(streamStateHandle, never()).discardState(); +DiscardRecordedStateObject.verifyDiscard( +streamStateHandle, TernaryBoolean.FALSE); ++sharedHandleCount; } for (HandleAndLocalPath handleAndLocalPath : incrementalKeyedStateHandle.getPrivateState()) { -verify(handleAndLocalPath.getHandle(), never()).discardState(); +DiscardRecordedStateObject.verifyDiscard( +handleAndLocalPath.getHandle(), TernaryBoolean.FALSE); } verify(incrementalKeyedStateHandle.getMetaStateHandle(), never()) @@ -2969,7 +2972,8 @@ class CheckpointCoordinatorTest extends TestLogger { // by CP1 for (List cpList : sharedHandlesByCheckpoint) { for (HandleAndLocalPath handleAndLocalPath : cpList) { -verify(handleAndLocalPath.getHandle(), never()).discardState(); +DiscardRecordedStateObject.verifyDiscard( +handleAndLocalPath.getHandle(), TernaryBoolean.FALSE); } } @@ -3028,14 +3032,19 @@ class CheckpointCoordinatorTest extends TestLogger { // references the state from CP1, so we expect they are not discarded. verifyDiscard( sharedHandlesByCheckpoint, -cpId -> restoreMode == RestoreMode.CLAIM && cpId == 0 ? times(1) : never()); +cpId -> +restoreMode == RestoreMode.CLAIM && cpId == 0 +? TernaryBoolean.TRUE +: TernaryBoolean.FALSE); // discard CP2