From 0efae7298435ae590380ab6944babb9cc2a30cff Mon Sep 17 00:00:00 2001 From: Purushottam Sinha Date: Tue, 28 Jul 2026 21:51:31 +0530 Subject: [PATCH] [FLINK-39876][tests] Migrate flink-streaming-java assertions to AssertJ The contributor guide asks for JUnit 5 and AssertJ and states "Don't use Hamcrest, JUnit assertions and `assert` directive". flink-streaming-java was already on JUnit 5 (FLINK-25544), but ten test classes still reached for Hamcrest. This removes the last of them. * TypeSafeMatcher and FeatureMatcher instances that were bridged into AssertJ via HamcrestCondition.matching() become native AssertJ Conditions. * MockitoHamcrest.argThat() becomes Mockito's own ArgumentMatchers.argThat(). The Hamcrest matchers it wrapped (containsInAnyOrder, contains, hasEntry, allOf) are expressed as predicates, which also removes the casts their call sites needed. CheckpointExceptionMatcher now implements Mockito's ArgumentMatcher rather than Hamcrest's BaseMatcher. * Drops the unused EqualsResourceSpecMatcher from StreamGraphGeneratorTest. The replacements keep the semantics of the matchers they replace, which AssertJ and Mockito do not provide for free: the predicates reject null explicitly, because TypeSafeMatcher never invoked matchesSafely() on a null actual; containsInAnyOrder counts duplicates; and the map matcher checks for the expected entries rather than map equality, as hasEntry did. No production code is touched. Each migrated predicate was inverted in turn to confirm the tests using it fail. Generated-by: Claude Code (Claude Opus 5) --- .../functions/sink/filesystem/BucketTest.java | 156 ++++++----------- .../sink/filesystem/BucketsTest.java | 44 ++--- ...phGeneratorExecutionModeDetectionTest.java | 159 +++++++----------- .../api/graph/StreamGraphGeneratorTest.java | 67 ++------ .../functions/InternalWindowFunctionTest.java | 81 +++++---- .../DynamicEventTimeSessionWindowsTest.java | 48 +++--- ...namicProcessingTimeSessionWindowsTest.java | 48 +++--- .../EventTimeSessionWindowsTest.java | 48 +++--- .../ProcessingTimeSessionWindowsTest.java | 48 +++--- .../StreamTaskCancellationBarrierTest.java | 20 +-- 10 files changed, 304 insertions(+), 415 deletions(-) diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/BucketTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/BucketTest.java index 372cfca3685279..e7cdea6fe88ffc 100644 --- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/BucketTest.java +++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/BucketTest.java @@ -33,8 +33,7 @@ import org.apache.flink.streaming.api.functions.sink.filesystem.utils.NoOpRecoverableWriter; import org.apache.flink.testutils.junit.utils.TempDirUtils; -import org.hamcrest.Description; -import org.hamcrest.TypeSafeMatcher; +import org.assertj.core.api.Condition; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; @@ -47,7 +46,6 @@ import static org.apache.flink.util.Preconditions.checkArgument; import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.HamcrestCondition.matching; /** Tests for the {@code Bucket}. */ class BucketTest { @@ -66,11 +64,11 @@ void shouldNotCleanupResumablesThatArePartOfTheAckedCheckpoint() throws IOExcept bucketUnderTest.write("test-element", 0L); final BucketState state = bucketUnderTest.onReceptionOfCheckpoint(0L); - assertThat(state).is(matching(hasActiveInProgressFile())); + assertThat(state).is(hasActiveInProgressFile()); bucketUnderTest.onSuccessfulCompletionOfCheckpoint(0L); assertThat(recoverableWriter) - .is(matching(hasCalledDiscard(0))); // it did not discard as this is still valid. + .is(hasCalledDiscard(0)); // it did not discard as this is still valid. } @Test @@ -85,7 +83,7 @@ void shouldCleanupOutdatedResumablesOnCheckpointAck() throws IOException { bucketUnderTest.write("test-element", 0L); final BucketState state = bucketUnderTest.onReceptionOfCheckpoint(0L); - assertThat(state).is(matching(hasActiveInProgressFile())); + assertThat(state).is(hasActiveInProgressFile()); bucketUnderTest.onSuccessfulCompletionOfCheckpoint(0L); @@ -93,8 +91,7 @@ void shouldCleanupOutdatedResumablesOnCheckpointAck() throws IOException { bucketUnderTest.onReceptionOfCheckpoint(2L); bucketUnderTest.onSuccessfulCompletionOfCheckpoint(2L); - assertThat(recoverableWriter) - .is(matching(hasCalledDiscard(2))); // that is for checkpoints 0 and 1 + assertThat(recoverableWriter).is(hasCalledDiscard(2)); // that is for checkpoints 0 and 1 } @Test @@ -107,14 +104,13 @@ void shouldNotCallCleanupWithoutInProgressPartFiles() throws Exception { createBucket(recoverableWriter, path, 0, 0, OutputFileConfig.builder().build()); final BucketState state = bucketUnderTest.onReceptionOfCheckpoint(0L); - assertThat(state).is(matching(hasNoActiveInProgressFile())); + assertThat(state).is(hasNoActiveInProgressFile()); bucketUnderTest.onReceptionOfCheckpoint(1L); bucketUnderTest.onReceptionOfCheckpoint(2L); bucketUnderTest.onSuccessfulCompletionOfCheckpoint(2L); - assertThat(recoverableWriter) - .is(matching(hasCalledDiscard(0))); // we have no in-progress file. + assertThat(recoverableWriter).is(hasCalledDiscard(0)); // we have no in-progress file. } @Test @@ -128,9 +124,9 @@ void shouldCleanupOutdatedResumablesAfterResumed() throws Exception { bucketUnderTest.write("test-element", 0L); final BucketState state0 = bucketUnderTest.onReceptionOfCheckpoint(0L); - assertThat(state0).is(matching(hasActiveInProgressFile())); + assertThat(state0).is(hasActiveInProgressFile()); bucketUnderTest.onSuccessfulCompletionOfCheckpoint(0L); - assertThat(recoverableWriter).is(matching(hasCalledDiscard(0))); + assertThat(recoverableWriter).is(hasCalledDiscard(0)); final File newOutDir = TempDirUtils.newFolder(tempFolder); final Path newPath = new Path(newOutDir.toURI()); @@ -139,9 +135,9 @@ void shouldCleanupOutdatedResumablesAfterResumed() throws Exception { restoreBucket( newRecoverableWriter, 0, 0, state0, OutputFileConfig.builder().build()); final BucketState state1 = bucketAfterResume.onReceptionOfCheckpoint(1L); - assertThat(state1).is(matching(hasActiveInProgressFile())); + assertThat(state1).is(hasActiveInProgressFile()); bucketAfterResume.onSuccessfulCompletionOfCheckpoint(1L); - assertThat(newRecoverableWriter).is(matching(hasCalledDiscard(1))); + assertThat(newRecoverableWriter).is(hasCalledDiscard(1)); } // --------------------------- Checking Restore --------------------------- @@ -152,8 +148,8 @@ void inProgressFileShouldBeCommittedIfWriterDoesNotSupportResume() throws IOExce final Bucket bucket = getRestoredBucketWithOnlyInProgressPart(nonResumableWriter); - assertThat(nonResumableWriter).is(matching(hasMethodCallCountersEqualTo(1, 0, 1))); - assertThat(bucket).is(matching(hasNullInProgressFile(true))); + assertThat(nonResumableWriter).is(hasMethodCallCountersEqualTo(1, 0, 1)); + assertThat(bucket).is(hasNullInProgressFile(true)); } @Test @@ -162,8 +158,8 @@ void inProgressFileShouldBeRestoredIfWriterSupportsResume() throws IOException { final Bucket bucket = getRestoredBucketWithOnlyInProgressPart(resumableWriter); - assertThat(resumableWriter).is(matching(hasMethodCallCountersEqualTo(1, 1, 0))); - assertThat(bucket).is(matching(hasNullInProgressFile(false))); + assertThat(resumableWriter).is(hasMethodCallCountersEqualTo(1, 1, 0)); + assertThat(bucket).is(hasNullInProgressFile(false)); } @Test @@ -174,108 +170,54 @@ void pendingFilesShouldBeRestored() throws IOException { final Bucket bucket = getRestoredBucketWithOnlyPendingParts(writer, expectedRecoverForCommitCounter); - assertThat(writer) - .is(matching(hasMethodCallCountersEqualTo(0, 0, expectedRecoverForCommitCounter))); - assertThat(bucket).is(matching(hasNullInProgressFile(true))); + assertThat(writer).is(hasMethodCallCountersEqualTo(0, 0, expectedRecoverForCommitCounter)); + assertThat(bucket).is(hasNullInProgressFile(true)); } - // ------------------------------- Matchers -------------------------------- - - private static TypeSafeMatcher hasCalledDiscard(int times) { - return new TypeSafeMatcher() { - @Override - protected boolean matchesSafely(TestRecoverableWriter writer) { - return writer.getCleanupCallCounter() == times; - } - - @Override - public void describeTo(Description description) { - description - .appendText( - "the TestRecoverableWriter to have called discardRecoverableState() ") - .appendValue(times) - .appendText(" times."); - } - }; + // ------------------------------- Conditions -------------------------------- + + private static Condition hasCalledDiscard(int times) { + return new Condition<>( + writer -> writer != null && writer.getCleanupCallCounter() == times, + "the TestRecoverableWriter to have called discardRecoverableState() %d times.", + times); } - private static TypeSafeMatcher> hasActiveInProgressFile() { - return new TypeSafeMatcher>() { - @Override - protected boolean matchesSafely(BucketState state) { - return state.getInProgressFileRecoverable() != null; - } - - @Override - public void describeTo(Description description) { - description.appendText("a BucketState with active in-progress file."); - } - }; + private static Condition> hasActiveInProgressFile() { + return new Condition<>( + state -> state != null && state.getInProgressFileRecoverable() != null, + "a BucketState with active in-progress file."); } - private static TypeSafeMatcher> hasNoActiveInProgressFile() { - return new TypeSafeMatcher>() { - @Override - protected boolean matchesSafely(BucketState state) { - return state.getInProgressFileRecoverable() == null; - } - - @Override - public void describeTo(Description description) { - description.appendText("a BucketState with no active in-progress file."); - } - }; + private static Condition> hasNoActiveInProgressFile() { + return new Condition<>( + state -> state != null && state.getInProgressFileRecoverable() == null, + "a BucketState with no active in-progress file."); } - private static TypeSafeMatcher> hasNullInProgressFile( - final boolean isNull) { - - return new TypeSafeMatcher>() { - @Override - protected boolean matchesSafely(Bucket bucket) { - final InProgressFileWriter inProgressPart = - bucket.getInProgressPart(); - return isNull == (inProgressPart == null); - } - - @Override - public void describeTo(Description description) { - description - .appendText("a Bucket with its inProgressPart being ") - .appendText(isNull ? " null." : " not null."); - } - }; + private static Condition> hasNullInProgressFile(final boolean isNull) { + return new Condition<>( + bucket -> bucket != null && isNull == (bucket.getInProgressPart() == null), + "a Bucket with its inProgressPart being %s", + isNull ? "null." : "not null."); } - private static TypeSafeMatcher hasMethodCallCountersEqualTo( + private static Condition hasMethodCallCountersEqualTo( final int supportsResumeCalls, final int recoverCalls, final int recoverForCommitCalls) { - return new TypeSafeMatcher() { - @Override - protected boolean matchesSafely(BaseStubWriter writer) { - return writer.getSupportsResumeCallCounter() == supportsResumeCalls - && writer.getRecoverCallCounter() == recoverCalls - && writer.getRecoverForCommitCallCounter() == recoverForCommitCalls; - } - - @Override - public void describeTo(Description description) { - description - .appendText("a Writer where:") - .appendText(" supportsResume was called ") - .appendValue(supportsResumeCalls) - .appendText(" times,") - .appendText(" recover was called ") - .appendValue(recoverCalls) - .appendText(" times,") - .appendText(" and recoverForCommit was called ") - .appendValue(recoverForCommitCalls) - .appendText(" times.") - .appendText("'"); - } - }; + return new Condition<>( + writer -> + writer != null + && writer.getSupportsResumeCallCounter() == supportsResumeCalls + && writer.getRecoverCallCounter() == recoverCalls + && writer.getRecoverForCommitCallCounter() == recoverForCommitCalls, + "a Writer where: supportsResume was called %d times, recover was called %d times, " + + "and recoverForCommit was called %d times.", + supportsResumeCalls, + recoverCalls, + recoverForCommitCalls); } // ------------------------------- Mock Classes -------------------------------- diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/BucketsTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/BucketsTest.java index 30e14518b59aa1..6b293ca8e86ea4 100644 --- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/BucketsTest.java +++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/BucketsTest.java @@ -31,8 +31,7 @@ import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.OnCheckpointRollingPolicy; import org.apache.flink.testutils.junit.utils.TempDirUtils; -import org.hamcrest.Description; -import org.hamcrest.TypeSafeMatcher; +import org.assertj.core.api.Condition; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; @@ -45,7 +44,6 @@ import java.util.Map; import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.HamcrestCondition.matching; /** Tests for {@link Buckets}. */ class BucketsTest { @@ -69,15 +67,15 @@ void testSnapshotAndRestore() throws Exception { buckets.snapshotState(0L, bucketStateContainer, partCounterContainer); assertThat(buckets.getActiveBuckets().get("test1")) - .is(matching(hasSinglePartFileToBeCommittedOnCheckpointAck(path, "test1"))); + .is(hasSinglePartFileToBeCommittedOnCheckpointAck(path, "test1")); buckets.onElement("test2", new TestUtils.MockSinkContext(null, 1L, 2L)); buckets.snapshotState(1L, bucketStateContainer, partCounterContainer); assertThat(buckets.getActiveBuckets().get("test1")) - .is(matching(hasSinglePartFileToBeCommittedOnCheckpointAck(path, "test1"))); + .is(hasSinglePartFileToBeCommittedOnCheckpointAck(path, "test1")); assertThat(buckets.getActiveBuckets().get("test2")) - .is(matching(hasSinglePartFileToBeCommittedOnCheckpointAck(path, "test2"))); + .is(hasSinglePartFileToBeCommittedOnCheckpointAck(path, "test2")); Buckets restoredBuckets = restoreBuckets( @@ -94,27 +92,19 @@ void testSnapshotAndRestore() throws Exception { assertThat(activeBuckets).isEmpty(); } - private static TypeSafeMatcher> - hasSinglePartFileToBeCommittedOnCheckpointAck( - final Path testTmpPath, final String bucketId) { - return new TypeSafeMatcher>() { - @Override - protected boolean matchesSafely(Bucket bucket) { - return bucket.getBucketId().equals(bucketId) - && bucket.getBucketPath().equals(new Path(testTmpPath, bucketId)) - && bucket.getInProgressPart() == null - && bucket.getPendingFileRecoverablesForCurrentCheckpoint().isEmpty() - && bucket.getPendingFileRecoverablesPerCheckpoint().size() == 1; - } - - @Override - public void describeTo(Description description) { - description - .appendText("a Bucket with a single pending part file @ ") - .appendValue(new Path(testTmpPath, bucketId)) - .appendText("'"); - } - }; + private static Condition> hasSinglePartFileToBeCommittedOnCheckpointAck( + final Path testTmpPath, final String bucketId) { + final Path bucketPath = new Path(testTmpPath, bucketId); + return new Condition<>( + bucket -> + bucket != null + && bucket.getBucketId().equals(bucketId) + && bucket.getBucketPath().equals(bucketPath) + && bucket.getInProgressPart() == null + && bucket.getPendingFileRecoverablesForCurrentCheckpoint().isEmpty() + && bucket.getPendingFileRecoverablesPerCheckpoint().size() == 1, + "a Bucket with a single pending part file @ %s", + bucketPath); } @Test diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorExecutionModeDetectionTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorExecutionModeDetectionTest.java index ff06b12d462354..6c456fa6b5d2a2 100644 --- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorExecutionModeDetectionTest.java +++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorExecutionModeDetectionTest.java @@ -35,8 +35,7 @@ import org.apache.flink.streaming.api.transformations.SourceTransformation; import org.apache.flink.streaming.api.transformations.TwoInputTransformation; -import org.hamcrest.Description; -import org.hamcrest.TypeSafeMatcher; +import org.assertj.core.api.Condition; import org.junit.jupiter.api.Test; import java.util.ArrayList; @@ -45,7 +44,6 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; -import static org.assertj.core.api.HamcrestCondition.matching; /** * Tests for the detection of the {@link RuntimeExecutionMode runtime execution mode} during stream @@ -68,12 +66,11 @@ void testExecutionModePropagationFromEnvWithDefaultAndBoundedSource() { assertThat(environment.getStreamGraph()) .is( - matching( - hasProperties( - GlobalStreamExchangeMode.ALL_EDGES_PIPELINED, - JobType.STREAMING, - true, - true))); + hasProperties( + GlobalStreamExchangeMode.ALL_EDGES_PIPELINED, + JobType.STREAMING, + true, + true)); } @Test @@ -90,12 +87,11 @@ void testExecutionModePropagationFromEnvWithDefaultAndUnboundedSource() { assertThat(environment.getStreamGraph()) .is( - matching( - hasProperties( - GlobalStreamExchangeMode.ALL_EDGES_PIPELINED, - JobType.STREAMING, - false, - true))); + hasProperties( + GlobalStreamExchangeMode.ALL_EDGES_PIPELINED, + JobType.STREAMING, + false, + true)); } @Test @@ -121,12 +117,11 @@ void testExecutionModePropagationFromEnvWithAutomaticAndBoundedSource() { final StreamGraph streamGraph = environment.getStreamGraph(); assertThat(streamGraph) .is( - matching( - hasProperties( - GlobalStreamExchangeMode.ALL_EDGES_BLOCKING, - JobType.BATCH, - false, - false))); + hasProperties( + GlobalStreamExchangeMode.ALL_EDGES_BLOCKING, + JobType.BATCH, + false, + false)); } @Test @@ -175,12 +170,11 @@ void testDetectionThroughTransitivePredecessors() { generateStreamGraph(RuntimeExecutionMode.AUTOMATIC, resultTransform); assertThat(graph) .is( - matching( - hasProperties( - GlobalStreamExchangeMode.ALL_EDGES_PIPELINED, - JobType.STREAMING, - false, - true))); + hasProperties( + GlobalStreamExchangeMode.ALL_EDGES_PIPELINED, + JobType.STREAMING, + false, + true)); } @Test @@ -192,12 +186,11 @@ void testBoundedDetection() { final StreamGraph graph = generateStreamGraph(RuntimeExecutionMode.AUTOMATIC, bounded); assertThat(graph) .is( - matching( - hasProperties( - GlobalStreamExchangeMode.ALL_EDGES_BLOCKING, - JobType.BATCH, - false, - false))); + hasProperties( + GlobalStreamExchangeMode.ALL_EDGES_BLOCKING, + JobType.BATCH, + false, + false)); } @Test @@ -209,12 +202,11 @@ void testUnboundedDetection() { final StreamGraph graph = generateStreamGraph(RuntimeExecutionMode.AUTOMATIC, unbounded); assertThat(graph) .is( - matching( - hasProperties( - GlobalStreamExchangeMode.ALL_EDGES_PIPELINED, - JobType.STREAMING, - false, - true))); + hasProperties( + GlobalStreamExchangeMode.ALL_EDGES_PIPELINED, + JobType.STREAMING, + false, + true)); } @Test @@ -230,12 +222,11 @@ void testMixedDetection() { final StreamGraph graph = generateStreamGraph(RuntimeExecutionMode.AUTOMATIC, unbounded); assertThat(graph) .is( - matching( - hasProperties( - GlobalStreamExchangeMode.ALL_EDGES_PIPELINED, - JobType.STREAMING, - false, - true))); + hasProperties( + GlobalStreamExchangeMode.ALL_EDGES_PIPELINED, + JobType.STREAMING, + false, + true)); } @Test @@ -247,23 +238,21 @@ void testExplicitOverridesDetectedMode() { final StreamGraph graph = generateStreamGraph(RuntimeExecutionMode.AUTOMATIC, bounded); assertThat(graph) .is( - matching( - hasProperties( - GlobalStreamExchangeMode.ALL_EDGES_BLOCKING, - JobType.BATCH, - false, - false))); + hasProperties( + GlobalStreamExchangeMode.ALL_EDGES_BLOCKING, + JobType.BATCH, + false, + false)); final StreamGraph streamingGraph = generateStreamGraph(RuntimeExecutionMode.STREAMING, bounded); assertThat(streamingGraph) .is( - matching( - hasProperties( - GlobalStreamExchangeMode.ALL_EDGES_PIPELINED, - JobType.STREAMING, - false, - true))); + hasProperties( + GlobalStreamExchangeMode.ALL_EDGES_PIPELINED, + JobType.STREAMING, + false, + true)); } private StreamGraph generateStreamGraph( @@ -293,50 +282,26 @@ private StreamGraph generateStreamGraph( 1); } - private static TypeSafeMatcher hasProperties( + private static Condition hasProperties( final GlobalStreamExchangeMode exchangeMode, final JobType jobType, final boolean isCheckpointingEnabled, final boolean isAllVerticesInSameSlotSharingGroupByDefault) { - return new TypeSafeMatcher() { - @Override - protected boolean matchesSafely(StreamGraph actualStreamGraph) { - return exchangeMode == actualStreamGraph.getGlobalStreamExchangeMode() - && jobType == actualStreamGraph.getJobType() - && actualStreamGraph.getCheckpointConfig().isCheckpointingEnabled() - == isCheckpointingEnabled - && actualStreamGraph.isAllVerticesInSameSlotSharingGroupByDefault() - == isAllVerticesInSameSlotSharingGroupByDefault; - } - - @Override - public void describeTo(Description description) { - description - .appendText("a StreamGraph with exchangeMode=") - .appendValue(exchangeMode) - .appendText(", jobType=") - .appendValue(jobType) - .appendText(", isCheckpointingEnabled=") - .appendValue(isCheckpointingEnabled) - .appendText(", isAllVerticesInSameSlotSharingGroupByDefault=") - .appendValue(isAllVerticesInSameSlotSharingGroupByDefault); - } - - @Override - protected void describeMismatchSafely( - StreamGraph item, Description mismatchDescription) { - mismatchDescription - .appendText("was ") - .appendText("a StreamGraph with exchangeMode=") - .appendValue(exchangeMode) - .appendText(", jobType=") - .appendValue(jobType) - .appendText(", isCheckpointingEnabled=") - .appendValue(isCheckpointingEnabled) - .appendText(", isAllVerticesInSameSlotSharingGroupByDefault=") - .appendValue(isAllVerticesInSameSlotSharingGroupByDefault); - } - }; + return new Condition<>( + actualStreamGraph -> + actualStreamGraph != null + && exchangeMode == actualStreamGraph.getGlobalStreamExchangeMode() + && jobType == actualStreamGraph.getJobType() + && actualStreamGraph.getCheckpointConfig().isCheckpointingEnabled() + == isCheckpointingEnabled + && actualStreamGraph.isAllVerticesInSameSlotSharingGroupByDefault() + == isAllVerticesInSameSlotSharingGroupByDefault, + "a StreamGraph with exchangeMode=%s, jobType=%s, isCheckpointingEnabled=%s, " + + "isAllVerticesInSameSlotSharingGroupByDefault=%s", + exchangeMode, + jobType, + isCheckpointingEnabled, + isAllVerticesInSameSlotSharingGroupByDefault); } } diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorTest.java index dbde71ef687fe3..c4d58fa8b919fc 100644 --- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorTest.java +++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorTest.java @@ -21,7 +21,6 @@ import org.apache.flink.api.common.ExecutionConfig; import org.apache.flink.api.common.RuntimeExecutionMode; import org.apache.flink.api.common.functions.Function; -import org.apache.flink.api.common.operators.ResourceSpec; import org.apache.flink.api.common.operators.SlotSharingGroup; import org.apache.flink.api.common.state.MapStateDescriptor; import org.apache.flink.api.common.typeinfo.BasicTypeInfo; @@ -80,11 +79,8 @@ import org.apache.flink.util.OutputTag; import org.assertj.core.api.Assertions; +import org.assertj.core.api.Condition; import org.assertj.core.api.InstanceOfAssertFactories; -import org.hamcrest.Description; -import org.hamcrest.FeatureMatcher; -import org.hamcrest.Matcher; -import org.hamcrest.TypeSafeMatcher; import org.junit.jupiter.api.Test; import java.util.ArrayList; @@ -100,8 +96,6 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; -import static org.assertj.core.api.HamcrestCondition.matching; -import static org.hamcrest.Matchers.equalTo; /** * Tests for {@link StreamGraphGenerator}. This only tests correct translation of split/select, @@ -378,22 +372,17 @@ public void processBroadcastElement( assertThat(streamGraph.getStreamNodes().size()).isEqualTo(7); // forward - assertThat(edge(streamGraph, source1, map1)) - .is(matching(supportsUnalignedCheckpoints(false))); + assertThat(edge(streamGraph, source1, map1)).is(supportsUnalignedCheckpoints(false)); // shuffle - assertThat(edge(streamGraph, source2, map2)) - .is(matching(supportsUnalignedCheckpoints(true))); + assertThat(edge(streamGraph, source2, map2)).is(supportsUnalignedCheckpoints(true)); // broadcast, but other channel is forwarded - assertThat(edge(streamGraph, map1, joined)) - .is(matching(supportsUnalignedCheckpoints(false))); + assertThat(edge(streamGraph, map1, joined)).is(supportsUnalignedCheckpoints(false)); // forward - assertThat(edge(streamGraph, map2, joined)) - .is(matching(supportsUnalignedCheckpoints(false))); + assertThat(edge(streamGraph, map2, joined)).is(supportsUnalignedCheckpoints(false)); // shuffle - assertThat(edge(streamGraph, joined, map3)) - .is(matching(supportsUnalignedCheckpoints(true))); + assertThat(edge(streamGraph, joined, map3)).is(supportsUnalignedCheckpoints(true)); // rescale - assertThat(edge(streamGraph, map3, map4)).is(matching(supportsUnalignedCheckpoints(false))); + assertThat(edge(streamGraph, map3, map4)).is(supportsUnalignedCheckpoints(false)); } @Test @@ -427,14 +416,11 @@ public void processBroadcastElement( assertThat(streamGraph.getStreamNodes().size()).isEqualTo(4); // single broadcast - assertThat(edge(streamGraph, source1, map1)) - .is(matching(supportsUnalignedCheckpoints(false))); + assertThat(edge(streamGraph, source1, map1)).is(supportsUnalignedCheckpoints(false)); // keyed, connected with broadcast - assertThat(edge(streamGraph, source2, joined)) - .is(matching(supportsUnalignedCheckpoints(false))); + assertThat(edge(streamGraph, source2, joined)).is(supportsUnalignedCheckpoints(false)); // broadcast, connected with keyed - assertThat(edge(streamGraph, map1, joined)) - .is(matching(supportsUnalignedCheckpoints(false))); + assertThat(edge(streamGraph, map1, joined)).is(supportsUnalignedCheckpoints(false)); } private static StreamEdge edge( @@ -444,16 +430,11 @@ private static StreamEdge edge( return streamEdges.get(0); } - private static Matcher supportsUnalignedCheckpoints(boolean enabled) { - return new FeatureMatcher( - equalTo(enabled), - "supports unaligned checkpoint", - "supports unaligned checkpoint") { - @Override - protected Boolean featureValueOf(StreamEdge actual) { - return actual.supportsUnalignedCheckpoints(); - } - }; + private static Condition supportsUnalignedCheckpoints(boolean enabled) { + return new Condition<>( + edge -> edge != null && edge.supportsUnalignedCheckpoints() == enabled, + "supports unaligned checkpoint: %s", + enabled); } /** @@ -1202,24 +1183,6 @@ public Long map2(Long value) throws Exception { } } - private static class EqualsResourceSpecMatcher extends TypeSafeMatcher { - private final ResourceSpec resources; - - EqualsResourceSpecMatcher(ResourceSpec resources) { - this.resources = resources; - } - - @Override - public void describeTo(Description description) { - description.appendText("expected resource spec ").appendValue(resources); - } - - @Override - protected boolean matchesSafely(ResourceSpec item) { - return resources.lessThanOrEqual(item) && item.lessThanOrEqual(resources); - } - } - private static class MultipleInputOperatorFactory implements StreamOperatorFactory { @Override public > T createStreamOperator( diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/windowing/functions/InternalWindowFunctionTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/windowing/functions/InternalWindowFunctionTest.java index 1b94b8cd7adad2..3737d2598e48cb 100644 --- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/windowing/functions/InternalWindowFunctionTest.java +++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/windowing/functions/InternalWindowFunctionTest.java @@ -44,29 +44,27 @@ import org.apache.flink.streaming.runtime.operators.windowing.functions.InternalWindowFunction; import org.apache.flink.streaming.util.functions.StreamingFunctionUtils; import org.apache.flink.util.Collector; +import org.apache.flink.util.IterableUtils; -import org.hamcrest.collection.IsIterableContainingInOrder; import org.junit.jupiter.api.Test; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; +import java.util.Arrays; import java.util.HashMap; import java.util.HashSet; import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.stream.Collectors; -import static org.hamcrest.CoreMatchers.is; -import static org.hamcrest.Matchers.contains; -import static org.hamcrest.collection.IsMapContaining.hasEntry; -import static org.hamcrest.core.AllOf.allOf; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.argThat; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; -import static org.mockito.hamcrest.MockitoHamcrest.argThat; /** Tests for {@link InternalWindowFunction}. */ class InternalWindowFunctionTest { @@ -301,12 +299,7 @@ void testInternalSingleValueWindowFunction() throws Exception { mock(InternalWindowFunction.InternalWindowContext.class); windowFunction.process(42L, w, ctx, 23L, c); - verify(mock) - .apply( - eq(42L), - eq(w), - (Iterable) argThat(IsIterableContainingInOrder.contains(23L)), - eq(c)); + verify(mock).apply(eq(42L), eq(w), containsExactly(23L), eq(c)); // check close windowFunction.close(); @@ -349,11 +342,7 @@ void testInternalSingleValueAllWindowFunction() throws Exception { mock(InternalWindowFunction.InternalWindowContext.class); windowFunction.process(((byte) 0), w, ctx, 23L, c); - verify(mock) - .apply( - eq(w), - (Iterable) argThat(IsIterableContainingInOrder.contains(23L)), - eq(c)); + verify(mock).apply(eq(w), containsExactly(23L), eq(c)); // check close windowFunction.close(); @@ -397,10 +386,7 @@ void testInternalSingleValueProcessAllWindowFunction() throws Exception { windowFunction.process(((byte) 0), w, ctx, 23L, c); verify(mock) - .process( - (ProcessAllWindowFunctionMock.Context) any(), - (Iterable) argThat(IsIterableContainingInOrder.contains(23L)), - eq(c)); + .process((ProcessAllWindowFunctionMock.Context) any(), containsExactly(23L), eq(c)); // check close windowFunction.close(); @@ -460,7 +446,7 @@ public Object answer(InvocationOnMock invocationOnMock) .process( eq(42L), (ProcessWindowFunctionMock.Context) any(), - (Iterable) argThat(IsIterableContainingInOrder.contains(23L)), + containsExactly(23L), eq(c)); windowFunction.process(42L, w, ctx, 23L, c); @@ -561,16 +547,7 @@ public Object answer(InvocationOnMock invocationOnMock) } }) .when(mock) - .process( - eq(42L), - any(), - (Iterable) - argThat( - contains( - allOf( - hasEntry(is(23L), is(23L)), - hasEntry(is(24L), is(24L))))), - eq(c)); + .process(eq(42L), any(), containsSingleMapWithEntries(23L, 24L), eq(c)); windowFunction.process(42L, w, ctx, args, c); verify(ctx).currentProcessingTime(); @@ -656,22 +633,42 @@ public Set merge(Set a, Set b) { mock(InternalWindowFunction.InternalWindowContext.class); windowFunction.process(((byte) 0), w, ctx, args, c); - verify(mock) - .process( - any(), - (Iterable) - argThat( - contains( - allOf( - hasEntry(is(23L), is(23L)), - hasEntry(is(24L), is(24L))))), - eq(c)); + verify(mock).process(any(), containsSingleMapWithEntries(23L, 24L), eq(c)); // check close windowFunction.close(); verify(mock).close(); } + /** Matches an {@link Iterable} containing exactly the expected elements, in order. */ + private static Iterable containsExactly(Long... expected) { + return argThat(actual -> actual != null && toList(actual).equals(Arrays.asList(expected))); + } + + /** + * Matches an {@link Iterable} containing a single map that has an entry mapping each expected + * key to itself. Extra entries are allowed, matching the {@code hasEntry} semantics this + * replaced. + */ + private static Iterable> containsSingleMapWithEntries(Long... expectedKeys) { + return argThat( + actual -> { + if (actual == null) { + return false; + } + final List> maps = toList(actual); + if (maps.size() != 1 || maps.get(0) == null) { + return false; + } + final Map map = maps.get(0); + return Arrays.stream(expectedKeys).allMatch(key -> key.equals(map.get(key))); + }); + } + + private static List toList(Iterable iterable) { + return IterableUtils.toStream(iterable).collect(Collectors.toList()); + } + private static class ProcessWindowFunctionMock extends ProcessWindowFunction implements OutputTypeConfigurable { diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/DynamicEventTimeSessionWindowsTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/DynamicEventTimeSessionWindowsTest.java index e6832729a86c26..a46b984188a2e2 100644 --- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/DynamicEventTimeSessionWindowsTest.java +++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/DynamicEventTimeSessionWindowsTest.java @@ -29,22 +29,23 @@ import org.junit.jupiter.api.Test; import org.mockito.ArgumentMatchers; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; import java.util.Collections; +import java.util.List; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; -import static org.hamcrest.Matchers.containsInAnyOrder; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyCollection; +import static org.mockito.ArgumentMatchers.argThat; import static org.mockito.Mockito.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; -import static org.mockito.hamcrest.MockitoHamcrest.argThat; /** Tests for {@link DynamicEventTimeSessionWindows}. */ class DynamicEventTimeSessionWindowsTest { @@ -121,20 +122,13 @@ void testMergeConsecutiveWindows() { verify(callback, times(1)) .merge( - (Collection) - argThat( - containsInAnyOrder( - new TimeWindow(0, 1), - new TimeWindow(1, 2), - new TimeWindow(2, 3))), + containsInAnyOrder( + new TimeWindow(0, 1), new TimeWindow(1, 2), new TimeWindow(2, 3)), eq(new TimeWindow(0, 3))); verify(callback, times(1)) .merge( - (Collection) - argThat( - containsInAnyOrder( - new TimeWindow(4, 5), new TimeWindow(5, 6))), + containsInAnyOrder(new TimeWindow(4, 5), new TimeWindow(5, 6)), eq(new TimeWindow(4, 6))); verify(callback, times(2)).merge(anyCollection(), ArgumentMatchers.any()); @@ -160,18 +154,12 @@ void testMergeCoveringWindow() { verify(callback, times(1)) .merge( - (Collection) - argThat( - containsInAnyOrder( - new TimeWindow(1, 1), new TimeWindow(0, 2))), + containsInAnyOrder(new TimeWindow(1, 1), new TimeWindow(0, 2)), eq(new TimeWindow(0, 2))); verify(callback, times(1)) .merge( - (Collection) - argThat( - containsInAnyOrder( - new TimeWindow(5, 6), new TimeWindow(4, 7))), + containsInAnyOrder(new TimeWindow(5, 6), new TimeWindow(4, 7)), eq(new TimeWindow(4, 7))); verify(callback, times(2)).merge(anyCollection(), ArgumentMatchers.any()); @@ -208,4 +196,24 @@ void testProperties() { .isEqualTo(new TimeWindow.Serializer()); assertThat(assigner.getDefaultTrigger()).isInstanceOf(EventTimeTrigger.class); } + + /** + * Matches a collection containing exactly the expected windows, in any order, counting + * duplicates. + */ + private static Collection containsInAnyOrder(TimeWindow... windows) { + return argThat( + actual -> { + if (actual == null) { + return false; + } + final List remaining = new ArrayList<>(actual); + for (TimeWindow window : windows) { + if (!remaining.remove(window)) { + return false; + } + } + return remaining.isEmpty(); + }); + } } diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/DynamicProcessingTimeSessionWindowsTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/DynamicProcessingTimeSessionWindowsTest.java index 373796b4020e08..4eea95246fd701 100644 --- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/DynamicProcessingTimeSessionWindowsTest.java +++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/DynamicProcessingTimeSessionWindowsTest.java @@ -29,22 +29,23 @@ import org.junit.jupiter.api.Test; import org.mockito.ArgumentMatchers; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; import java.util.Collections; +import java.util.List; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; -import static org.hamcrest.Matchers.containsInAnyOrder; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyCollection; +import static org.mockito.ArgumentMatchers.argThat; import static org.mockito.Mockito.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; -import static org.mockito.hamcrest.MockitoHamcrest.argThat; /** Tests for {@link DynamicProcessingTimeSessionWindows}. */ class DynamicProcessingTimeSessionWindowsTest { @@ -126,20 +127,13 @@ void testMergeConsecutiveWindows() { verify(callback, times(1)) .merge( - (Collection) - argThat( - containsInAnyOrder( - new TimeWindow(0, 1), - new TimeWindow(1, 2), - new TimeWindow(2, 3))), + containsInAnyOrder( + new TimeWindow(0, 1), new TimeWindow(1, 2), new TimeWindow(2, 3)), eq(new TimeWindow(0, 3))); verify(callback, times(1)) .merge( - (Collection) - argThat( - containsInAnyOrder( - new TimeWindow(4, 5), new TimeWindow(5, 6))), + containsInAnyOrder(new TimeWindow(4, 5), new TimeWindow(5, 6)), eq(new TimeWindow(4, 6))); verify(callback, times(2)).merge(anyCollection(), ArgumentMatchers.any()); @@ -165,18 +159,12 @@ void testMergeCoveringWindow() { verify(callback, times(1)) .merge( - (Collection) - argThat( - containsInAnyOrder( - new TimeWindow(1, 1), new TimeWindow(0, 2))), + containsInAnyOrder(new TimeWindow(1, 1), new TimeWindow(0, 2)), eq(new TimeWindow(0, 2))); verify(callback, times(1)) .merge( - (Collection) - argThat( - containsInAnyOrder( - new TimeWindow(5, 6), new TimeWindow(4, 7))), + containsInAnyOrder(new TimeWindow(5, 6), new TimeWindow(4, 7)), eq(new TimeWindow(4, 7))); verify(callback, times(2)).merge(anyCollection(), ArgumentMatchers.any()); @@ -213,4 +201,24 @@ void testProperties() { .isEqualTo(new TimeWindow.Serializer()); assertThat(assigner.getDefaultTrigger()).isInstanceOf(ProcessingTimeTrigger.class); } + + /** + * Matches a collection containing exactly the expected windows, in any order, counting + * duplicates. + */ + private static Collection containsInAnyOrder(TimeWindow... windows) { + return argThat( + actual -> { + if (actual == null) { + return false; + } + final List remaining = new ArrayList<>(actual); + for (TimeWindow window : windows) { + if (!remaining.remove(window)) { + return false; + } + } + return remaining.isEmpty(); + }); + } } diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/EventTimeSessionWindowsTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/EventTimeSessionWindowsTest.java index aa27fd3ad3dfd8..b75a50ec5dc199 100644 --- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/EventTimeSessionWindowsTest.java +++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/EventTimeSessionWindowsTest.java @@ -31,20 +31,21 @@ import org.mockito.ArgumentMatchers; import java.time.Duration; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; import java.util.Collections; +import java.util.List; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; -import static org.hamcrest.Matchers.containsInAnyOrder; import static org.mockito.ArgumentMatchers.anyCollection; +import static org.mockito.ArgumentMatchers.argThat; import static org.mockito.Mockito.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; -import static org.mockito.hamcrest.MockitoHamcrest.argThat; /** Tests for {@link EventTimeSessionWindows}. */ class EventTimeSessionWindowsTest { @@ -109,20 +110,13 @@ void testMergeConsecutiveWindows() { verify(callback, times(1)) .merge( - (Collection) - argThat( - containsInAnyOrder( - new TimeWindow(0, 1), - new TimeWindow(1, 2), - new TimeWindow(2, 3))), + containsInAnyOrder( + new TimeWindow(0, 1), new TimeWindow(1, 2), new TimeWindow(2, 3)), eq(new TimeWindow(0, 3))); verify(callback, times(1)) .merge( - (Collection) - argThat( - containsInAnyOrder( - new TimeWindow(4, 5), new TimeWindow(5, 6))), + containsInAnyOrder(new TimeWindow(4, 5), new TimeWindow(5, 6)), eq(new TimeWindow(4, 6))); verify(callback, times(2)).merge(anyCollection(), ArgumentMatchers.any()); @@ -145,18 +139,12 @@ void testMergeCoveringWindow() { verify(callback, times(1)) .merge( - (Collection) - argThat( - containsInAnyOrder( - new TimeWindow(1, 1), new TimeWindow(0, 2))), + containsInAnyOrder(new TimeWindow(1, 1), new TimeWindow(0, 2)), eq(new TimeWindow(0, 2))); verify(callback, times(1)) .merge( - (Collection) - argThat( - containsInAnyOrder( - new TimeWindow(5, 6), new TimeWindow(4, 7))), + containsInAnyOrder(new TimeWindow(5, 6), new TimeWindow(4, 7)), eq(new TimeWindow(4, 7))); verify(callback, times(2)).merge(anyCollection(), ArgumentMatchers.any()); @@ -212,4 +200,24 @@ void testDynamicGapProperties() { assertThat(assigner).isNotNull(); assertThat(assigner.isEventTime()).isTrue(); } + + /** + * Matches a collection containing exactly the expected windows, in any order, counting + * duplicates. + */ + private static Collection containsInAnyOrder(TimeWindow... windows) { + return argThat( + actual -> { + if (actual == null) { + return false; + } + final List remaining = new ArrayList<>(actual); + for (TimeWindow window : windows) { + if (!remaining.remove(window)) { + return false; + } + } + return remaining.isEmpty(); + }); + } } diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/ProcessingTimeSessionWindowsTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/ProcessingTimeSessionWindowsTest.java index 13847e8d73d777..13dc331f6dbc46 100644 --- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/ProcessingTimeSessionWindowsTest.java +++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/ProcessingTimeSessionWindowsTest.java @@ -33,19 +33,20 @@ import org.mockito.ArgumentMatchers; import java.time.Duration; +import java.util.ArrayList; import java.util.Collection; +import java.util.List; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; -import static org.hamcrest.Matchers.containsInAnyOrder; import static org.mockito.ArgumentMatchers.anyCollection; +import static org.mockito.ArgumentMatchers.argThat; import static org.mockito.Mockito.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; -import static org.mockito.hamcrest.MockitoHamcrest.argThat; /** Tests for {@link ProcessingTimeSessionWindows}. */ class ProcessingTimeSessionWindowsTest { @@ -116,20 +117,13 @@ void testMergeConsecutiveWindows() { verify(callback, times(1)) .merge( - (Collection) - argThat( - containsInAnyOrder( - new TimeWindow(0, 1), - new TimeWindow(1, 2), - new TimeWindow(2, 3))), + containsInAnyOrder( + new TimeWindow(0, 1), new TimeWindow(1, 2), new TimeWindow(2, 3)), eq(new TimeWindow(0, 3))); verify(callback, times(1)) .merge( - (Collection) - argThat( - containsInAnyOrder( - new TimeWindow(4, 5), new TimeWindow(5, 6))), + containsInAnyOrder(new TimeWindow(4, 5), new TimeWindow(5, 6)), eq(new TimeWindow(4, 6))); verify(callback, times(2)).merge(anyCollection(), ArgumentMatchers.any()); @@ -153,18 +147,12 @@ void testMergeCoveringWindow() { verify(callback, times(1)) .merge( - (Collection) - argThat( - containsInAnyOrder( - new TimeWindow(1, 1), new TimeWindow(0, 2))), + containsInAnyOrder(new TimeWindow(1, 1), new TimeWindow(0, 2)), eq(new TimeWindow(0, 2))); verify(callback, times(1)) .merge( - (Collection) - argThat( - containsInAnyOrder( - new TimeWindow(5, 6), new TimeWindow(4, 7))), + containsInAnyOrder(new TimeWindow(5, 6), new TimeWindow(4, 7)), eq(new TimeWindow(4, 7))); verify(callback, times(2)).merge(anyCollection(), ArgumentMatchers.any()); @@ -224,4 +212,24 @@ void testDynamicGapProperties() { assertThat(assigner).isNotNull(); assertThat(assigner.isEventTime()).isFalse(); } + + /** + * Matches a collection containing exactly the expected windows, in any order, counting + * duplicates. + */ + private static Collection containsInAnyOrder(TimeWindow... windows) { + return argThat( + actual -> { + if (actual == null) { + return false; + } + final List remaining = new ArrayList<>(actual); + for (TimeWindow window : windows) { + if (!remaining.remove(window)) { + return false; + } + } + return remaining.isEmpty(); + }); + } } diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskCancellationBarrierTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskCancellationBarrierTest.java index 1f446c4541d9e4..52d7543b11aa8a 100644 --- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskCancellationBarrierTest.java +++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskCancellationBarrierTest.java @@ -29,19 +29,18 @@ import org.apache.flink.streaming.api.operators.StreamMap; import org.apache.flink.streaming.api.operators.co.CoStreamMap; -import org.hamcrest.BaseMatcher; -import org.hamcrest.Description; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Timeout; +import org.mockito.ArgumentMatcher; import java.util.concurrent.TimeUnit; import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.argThat; import static org.mockito.Mockito.eq; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; -import static org.mockito.hamcrest.MockitoHamcrest.argThat; /** Test checkpoint cancellation barrier. */ @Timeout(value = 10, unit = TimeUnit.SECONDS) @@ -180,7 +179,8 @@ public String map2(String value) throws Exception { } /** A validation matcher for checkpoint exception against failure reason. */ - private static class CheckpointExceptionMatcher extends BaseMatcher { + private static class CheckpointExceptionMatcher + implements ArgumentMatcher { private final CheckpointFailureReason failureReason; @@ -189,15 +189,15 @@ private CheckpointExceptionMatcher(CheckpointFailureReason failureReason) { } @Override - public boolean matches(Object o) { - return o != null - && o.getClass() == CheckpointException.class - && ((CheckpointException) o).getCheckpointFailureReason().equals(failureReason); + public boolean matches(CheckpointException exception) { + return exception != null + && exception.getClass() == CheckpointException.class + && exception.getCheckpointFailureReason().equals(failureReason); } @Override - public void describeTo(Description description) { - description.appendText("CheckpointException - reason = " + failureReason); + public String toString() { + return "CheckpointException - reason = " + failureReason; } } }