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; } } }