You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sn...@apache.org on 2026/08/07 15:07:24 UTC
(flink) branch master updated: [FLINK-39771][tests][JUnit5 migration] Module: flink-cep
This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 01c950f89b2 [FLINK-39771][tests][JUnit5 migration] Module: flink-cep
01c950f89b2 is described below
commit 01c950f89b259f3469e57f044cf5f91b3980b69d
Author: Purushottam Sinha <ps...@confluent.io>
AuthorDate: Fri Aug 7 20:37:17 2026 +0530
[FLINK-39771][tests][JUnit5 migration] Module: flink-cep
Generated-by: Claude Code (claude-opus-4-8)
---
.../test/java/org/apache/flink/cep/CEPITCase.java | 155 ++++-----
.../apache/flink/cep/nfa/AfterMatchSkipITCase.java | 89 ++---
.../org/apache/flink/cep/nfa/DeweyNumberTest.java | 50 +--
.../org/apache/flink/cep/nfa/GreedyITCase.java | 39 ++-
.../java/org/apache/flink/cep/nfa/GroupITCase.java | 47 ++-
.../flink/cep/nfa/IterativeConditionsITCase.java | 19 +-
.../java/org/apache/flink/cep/nfa/NFAITCase.java | 228 +++++++------
.../nfa/NFAIterativeConditionTimeContextTest.java | 9 +-
.../apache/flink/cep/nfa/NFAStateAccessTest.java | 22 +-
.../flink/cep/nfa/NFAStatusChangeITCase.java | 83 ++---
.../java/org/apache/flink/cep/nfa/NFATest.java | 31 +-
.../org/apache/flink/cep/nfa/NotPatternITCase.java | 77 +++--
.../apache/flink/cep/nfa/SameElementITCase.java | 31 +-
.../apache/flink/cep/nfa/TimesOrMoreITCase.java | 61 ++--
.../org/apache/flink/cep/nfa/TimesRangeITCase.java | 65 ++--
.../apache/flink/cep/nfa/UntilConditionITCase.java | 70 ++--
.../flink/cep/nfa/compiler/NFACompilerTest.java | 186 +++++------
.../sharedbuffer/LockableTypeSerializerTest.java | 24 +-
.../cep/nfa/sharedbuffer/SharedBufferTest.java | 96 +++---
.../flink/cep/operator/CEPMigrationTest.java | 135 ++++----
.../apache/flink/cep/operator/CEPOperatorTest.java | 252 +++++++--------
.../flink/cep/operator/CEPRescalingTest.java | 135 ++++----
.../operator/CepProcessFunctionContextTest.java | 21 +-
.../flink/cep/operator/CepRuntimeContextTest.java | 307 ++++++++----------
.../org/apache/flink/cep/pattern/PatternTest.java | 357 ++++++++++++---------
.../apache/flink/cep/utils/NFATestUtilities.java | 37 +--
.../org/apache/flink/cep/utils/OutputAsserter.java | 10 +-
.../org.junit.jupiter.api.extension.Extension | 17 +
28 files changed, 1282 insertions(+), 1371 deletions(-)
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/CEPITCase.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/CEPITCase.java
index 6e4c51778e7..320e392a53b 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/CEPITCase.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/CEPITCase.java
@@ -43,33 +43,34 @@ import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.operators.util.WatermarkStrategyWithPunctuatedWatermarks;
-import org.apache.flink.test.util.AbstractTestBaseJUnit4;
+import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameter;
+import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
import org.apache.flink.util.CloseableIterator;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
-import java.util.Comparator;
import java.util.List;
import java.util.Map;
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
/** End to end tests of both CEP operators and {@link NFA}. */
@SuppressWarnings("serial")
-@RunWith(Parameterized.class)
-public class CEPITCase extends AbstractTestBaseJUnit4 {
+@ExtendWith(ParameterizedTestExtension.class)
+class CEPITCase extends AbstractTestBase {
- @Parameterized.Parameter public Configuration envConfiguration;
+ @Parameter public Configuration envConfiguration;
- @Parameterized.Parameters
+ @Parameters
public static Collection<Configuration> prepareSharedBufferCacheConfig() {
Configuration miniCacheConfig = new Configuration();
miniCacheConfig.set(CEPCacheOptions.CEP_CACHE_STATISTICS_INTERVAL, Duration.ofSeconds(1));
@@ -87,8 +88,8 @@ public class CEPITCase extends AbstractTestBaseJUnit4 {
*
* @throws Exception
*/
- @Test
- public void testSimplePatternCEP() throws Exception {
+ @TestTemplate
+ void testSimplePatternCEP() throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment(envConfiguration);
@@ -134,11 +135,11 @@ public class CEPITCase extends AbstractTestBaseJUnit4 {
result.executeAndCollect().forEachRemaining(resultList::add);
- assertEquals(Arrays.asList("2,6,8"), resultList);
+ assertThat(resultList).containsExactly("2,6,8");
}
- @Test
- public void testSimpleKeyedPatternCEP() throws Exception {
+ @TestTemplate
+ void testSimpleKeyedPatternCEP() throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment(envConfiguration);
env.setParallelism(2);
@@ -198,13 +199,11 @@ public class CEPITCase extends AbstractTestBaseJUnit4 {
result.executeAndCollect().forEachRemaining(resultList::add);
- resultList.sort(String::compareTo);
-
- assertEquals(Arrays.asList("2,2,2", "3,3,3", "42,42,42"), resultList);
+ assertThat(resultList).containsExactlyInAnyOrder("2,2,2", "3,3,3", "42,42,42");
}
- @Test
- public void testSimplePatternEventTime() throws Exception {
+ @TestTemplate
+ void testSimplePatternEventTime() throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment(envConfiguration);
@@ -275,13 +274,11 @@ public class CEPITCase extends AbstractTestBaseJUnit4 {
result.executeAndCollect().forEachRemaining(resultList::add);
- resultList.sort(String::compareTo);
-
- assertEquals(Arrays.asList("1,5,4"), resultList);
+ assertThat(resultList).containsExactlyInAnyOrder("1,5,4");
}
- @Test
- public void testSimpleKeyedPatternEventTime() throws Exception {
+ @TestTemplate
+ void testSimpleKeyedPatternEventTime() throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment(envConfiguration);
env.setParallelism(2);
@@ -365,13 +362,11 @@ public class CEPITCase extends AbstractTestBaseJUnit4 {
result.executeAndCollect().forEachRemaining(resultList::add);
- resultList.sort(String::compareTo);
-
- assertEquals(Arrays.asList("1,1,1", "2,2,2"), resultList);
+ assertThat(resultList).containsExactlyInAnyOrder("1,1,1", "2,2,2");
}
- @Test
- public void testSimplePatternWithSingleState() throws Exception {
+ @TestTemplate
+ void testSimplePatternWithSingleState() throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment(envConfiguration);
@@ -401,16 +396,16 @@ public class CEPITCase extends AbstractTestBaseJUnit4 {
result.executeAndCollect().forEachRemaining(resultList::add);
- assertEquals(Arrays.asList(new Tuple2<>(0, 1)), resultList);
+ assertThat(resultList).containsExactly(new Tuple2<>(0, 1));
}
- @Test
- public void testProcessingTimeWithinBetweenFirstAndLast() throws Exception {
+ @TestTemplate
+ void testProcessingTimeWithinBetweenFirstAndLast() throws Exception {
testProcessingTimeWithWindow(WithinType.FIRST_AND_LAST);
}
- @Test
- public void testProcessingTimeWithinPreviousAndCurrent() throws Exception {
+ @TestTemplate
+ void testProcessingTimeWithinPreviousAndCurrent() throws Exception {
testProcessingTimeWithWindow(WithinType.PREVIOUS_AND_CURRENT);
}
@@ -443,11 +438,11 @@ public class CEPITCase extends AbstractTestBaseJUnit4 {
result.executeAndCollect().forEachRemaining(resultList::add);
- assertEquals(Arrays.asList(3), resultList);
+ assertThat(resultList).containsExactly(3);
}
- @Test
- public void testTimeoutHandlingWithinFirstAndLast() throws Exception {
+ @TestTemplate
+ void testTimeoutHandlingWithinFirstAndLast() throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment(envConfiguration);
env.setParallelism(1);
@@ -527,21 +522,15 @@ public class CEPITCase extends AbstractTestBaseJUnit4 {
result.executeAndCollect().forEachRemaining(resultList::add);
- resultList.sort(Comparator.comparing(Object::toString));
-
List<String> timeoutList = new ArrayList<>();
result.getSideOutput(outputTag).executeAndCollect().forEachRemaining(timeoutList::add);
- timeoutList.sort(Comparator.comparing(Object::toString));
- List<String> timeoutExpected = Arrays.asList("1.0", "2.0", "2.0");
- List<String> resultExpected = Arrays.asList("2.0,2.0,2.0");
-
- assertEquals(timeoutExpected, timeoutList);
- assertEquals(resultExpected, resultList);
+ assertThat(timeoutList).containsExactlyInAnyOrder("1.0", "2.0", "2.0");
+ assertThat(resultList).containsExactlyInAnyOrder("2.0,2.0,2.0");
}
- @Test
- public void testTimeoutHandlingWithinPreviousAndCurrent() throws Exception {
+ @TestTemplate
+ void testTimeoutHandlingWithinPreviousAndCurrent() throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment(envConfiguration);
env.setParallelism(1);
@@ -622,17 +611,11 @@ public class CEPITCase extends AbstractTestBaseJUnit4 {
result.executeAndCollect().forEachRemaining(resultList::add);
- resultList.sort(Comparator.comparing(Object::toString));
-
List<String> timeoutList = new ArrayList<>();
result.getSideOutput(outputTag).executeAndCollect().forEachRemaining(timeoutList::add);
- timeoutList.sort(Comparator.comparing(Object::toString));
- List<String> timeoutExpected = Arrays.asList("1.0", "2.0");
- List<String> resultExpected = Arrays.asList("1.0,2.0,2.0", "2.0,2.0,2.0");
-
- assertEquals(timeoutExpected, timeoutList);
- assertEquals(resultExpected, resultList);
+ assertThat(timeoutList).containsExactlyInAnyOrder("1.0", "2.0");
+ assertThat(resultList).containsExactlyInAnyOrder("1.0,2.0,2.0", "2.0,2.0,2.0");
}
/**
@@ -640,8 +623,8 @@ public class CEPITCase extends AbstractTestBaseJUnit4 {
*
* @throws Exception
*/
- @Test
- public void testSimpleOrFilterPatternCEP() throws Exception {
+ @TestTemplate
+ void testSimpleOrFilterPatternCEP() throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment(envConfiguration);
@@ -687,13 +670,7 @@ public class CEPITCase extends AbstractTestBaseJUnit4 {
result.executeAndCollect().forEachRemaining(resultList::add);
- List<String> expected = Arrays.asList("1,5,6", "1,2,3", "4,5,6", "1,2,6");
-
- expected.sort(String::compareTo);
-
- resultList.sort(String::compareTo);
-
- assertEquals(expected, resultList);
+ assertThat(resultList).containsExactlyInAnyOrder("1,5,6", "1,2,3", "4,5,6", "1,2,6");
}
/**
@@ -701,8 +678,8 @@ public class CEPITCase extends AbstractTestBaseJUnit4 {
*
* @throws Exception
*/
- @Test
- public void testSimplePatternEventTimeWithComparator() throws Exception {
+ @TestTemplate
+ void testSimplePatternEventTimeWithComparator() throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment(envConfiguration);
@@ -776,13 +753,7 @@ public class CEPITCase extends AbstractTestBaseJUnit4 {
result.executeAndCollect().forEachRemaining(resultList::add);
- List<String> expected = Arrays.asList("1,6,4", "1,5,4");
-
- expected.sort(String::compareTo);
-
- resultList.sort(String::compareTo);
-
- assertEquals(expected, resultList);
+ assertThat(resultList).containsExactlyInAnyOrder("1,6,4", "1,5,4");
}
private static class CustomEventComparator implements EventComparator<Event> {
@@ -792,8 +763,8 @@ public class CEPITCase extends AbstractTestBaseJUnit4 {
}
}
- @Test
- public void testSimpleAfterMatchSkip() throws Exception {
+ @TestTemplate
+ void testSimpleAfterMatchSkip() throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment(envConfiguration);
@@ -829,16 +800,11 @@ public class CEPITCase extends AbstractTestBaseJUnit4 {
result.executeAndCollect().forEachRemaining(resultList::add);
- resultList.sort(Comparator.comparing(tuple2 -> tuple2.toString()));
-
- List<Tuple2<Integer, String>> expected =
- Arrays.asList(Tuple2.of(1, "a"), Tuple2.of(3, "a"));
-
- assertEquals(expected, resultList);
+ assertThat(resultList).containsExactlyInAnyOrder(Tuple2.of(1, "a"), Tuple2.of(3, "a"));
}
- @Test
- public void testRichPatternFlatSelectFunction() throws Exception {
+ @TestTemplate
+ void testRichPatternFlatSelectFunction() throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment(envConfiguration);
@@ -916,11 +882,11 @@ public class CEPITCase extends AbstractTestBaseJUnit4 {
result.executeAndCollect().forEachRemaining(resultList::add);
- assertEquals(Arrays.asList("2,6,8"), resultList);
+ assertThat(resultList).containsExactly("2,6,8");
}
- @Test
- public void testRichPatternSelectFunction() throws Exception {
+ @TestTemplate
+ void testRichPatternSelectFunction() throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment(envConfiguration);
env.setParallelism(2);
@@ -1010,13 +976,11 @@ public class CEPITCase extends AbstractTestBaseJUnit4 {
result.executeAndCollect().forEachRemaining(resultList::add);
- resultList.sort(String::compareTo);
-
- assertEquals(Arrays.asList("2,2,2", "3,3,3", "42,42,42"), resultList);
+ assertThat(resultList).containsExactlyInAnyOrder("2,2,2", "3,3,3", "42,42,42");
}
- @Test
- public void testFlatSelectSerializationWithAnonymousClass() throws Exception {
+ @TestTemplate
+ void testFlatSelectSerializationWithAnonymousClass() throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment(envConfiguration);
@@ -1045,8 +1009,8 @@ public class CEPITCase extends AbstractTestBaseJUnit4 {
env.execute();
}
- @Test
- public void testPartialMatchTimeoutOutputCompletedMatch() throws Exception {
+ @TestTemplate
+ void testPartialMatchTimeoutOutputCompletedMatch() throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment(envConfiguration);
@@ -1107,7 +1071,6 @@ public class CEPITCase extends AbstractTestBaseJUnit4 {
try (CloseableIterator<String> iterator = result.executeAndCollect()) {
iterator.forEachRemaining(resultList::add);
}
- resultList.sort(String::compareTo);
- assertEquals(Arrays.asList("3,5"), resultList);
+ assertThat(resultList).containsExactlyInAnyOrder("3,5");
}
}
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/AfterMatchSkipITCase.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/AfterMatchSkipITCase.java
index f469aba73b7..2a1be144618 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/AfterMatchSkipITCase.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/AfterMatchSkipITCase.java
@@ -29,25 +29,24 @@ import org.apache.flink.cep.utils.NFATestHarness;
import org.apache.flink.cep.utils.TestSharedBuffer;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.util.FlinkRuntimeException;
-import org.apache.flink.util.TestLogger;
import org.apache.flink.shaded.guava33.com.google.common.collect.Lists;
-import org.hamcrest.Matchers;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import static org.apache.flink.cep.utils.NFATestUtilities.comparePatterns;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** IT tests covering {@link AfterMatchSkipStrategy}. */
-public class AfterMatchSkipITCase extends TestLogger {
+class AfterMatchSkipITCase {
@Test
- public void testNoSkip() throws Exception {
+ void testNoSkip() throws Exception {
List<StreamRecord<Event>> streamEvents = new ArrayList<>();
Event a1 = new Event(1, "a", 0.0);
@@ -83,7 +82,7 @@ public class AfterMatchSkipITCase extends TestLogger {
}
@Test
- public void testNoSkipWithFollowedByAny() throws Exception {
+ void testNoSkipWithFollowedByAny() throws Exception {
List<List<Event>> resultingPatterns =
TwoVariablesFollowedByAny.compute(AfterMatchSkipStrategy.noSkip());
@@ -99,7 +98,7 @@ public class AfterMatchSkipITCase extends TestLogger {
}
@Test
- public void testSkipToNextWithFollowedByAny() throws Exception {
+ void testSkipToNextWithFollowedByAny() throws Exception {
List<List<Event>> resultingPatterns =
TwoVariablesFollowedByAny.compute(AfterMatchSkipStrategy.skipToNext());
@@ -144,7 +143,7 @@ public class AfterMatchSkipITCase extends TestLogger {
}
@Test
- public void testNoSkipWithQuantifierAtTheEnd() throws Exception {
+ void testNoSkipWithQuantifierAtTheEnd() throws Exception {
List<List<Event>> resultingPatterns =
QuantifierAtEndOfPattern.compute(AfterMatchSkipStrategy.noSkip());
@@ -165,7 +164,7 @@ public class AfterMatchSkipITCase extends TestLogger {
}
@Test
- public void testSkipToNextWithQuantifierAtTheEnd() throws Exception {
+ void testSkipToNextWithQuantifierAtTheEnd() throws Exception {
List<List<Event>> resultingPatterns =
QuantifierAtEndOfPattern.compute(AfterMatchSkipStrategy.skipToNext());
@@ -209,7 +208,7 @@ public class AfterMatchSkipITCase extends TestLogger {
}
@Test
- public void testSkipPastLast() throws Exception {
+ void testSkipPastLast() throws Exception {
List<StreamRecord<Event>> streamEvents = new ArrayList<>();
Event a1 = new Event(1, "a", 0.0);
@@ -241,7 +240,7 @@ public class AfterMatchSkipITCase extends TestLogger {
}
@Test
- public void testSkipToFirst() throws Exception {
+ void testSkipToFirst() throws Exception {
List<StreamRecord<Event>> streamEvents = new ArrayList<>();
Event ab1 = new Event(1, "ab", 0.0);
@@ -278,7 +277,7 @@ public class AfterMatchSkipITCase extends TestLogger {
}
@Test
- public void testSkipToLast() throws Exception {
+ void testSkipToLast() throws Exception {
List<StreamRecord<Event>> streamEvents = new ArrayList<>();
Event ab1 = new Event(1, "ab", 0.0);
@@ -316,7 +315,7 @@ public class AfterMatchSkipITCase extends TestLogger {
}
@Test
- public void testSkipPastLast2() throws Exception {
+ void testSkipPastLast2() throws Exception {
List<StreamRecord<Event>> streamEvents = new ArrayList<>();
Event a1 = new Event(1, "a1", 0.0);
@@ -355,7 +354,7 @@ public class AfterMatchSkipITCase extends TestLogger {
}
@Test
- public void testSkipPastLast3() throws Exception {
+ void testSkipPastLast3() throws Exception {
List<StreamRecord<Event>> streamEvents = new ArrayList<>();
Event a1 = new Event(1, "a1", 0.0);
@@ -382,7 +381,7 @@ public class AfterMatchSkipITCase extends TestLogger {
}
@Test
- public void testSkipToFirstWithOptionalMatch() throws Exception {
+ void testSkipToFirstWithOptionalMatch() throws Exception {
List<StreamRecord<Event>> streamEvents = new ArrayList<>();
Event ab1 = new Event(1, "ab1", 0.0);
@@ -414,7 +413,7 @@ public class AfterMatchSkipITCase extends TestLogger {
}
@Test
- public void testSkipToFirstAtStartPosition() throws Exception {
+ void testSkipToFirstAtStartPosition() throws Exception {
List<StreamRecord<Event>> streamEvents = new ArrayList<>();
Event ab1 = new Event(1, "ab1", 0.0);
@@ -442,7 +441,7 @@ public class AfterMatchSkipITCase extends TestLogger {
}
@Test
- public void testSkipToFirstWithOneOrMore() throws Exception {
+ void testSkipToFirstWithOneOrMore() throws Exception {
List<StreamRecord<Event>> streamEvents = new ArrayList<>();
Event a1 = new Event(1, "a1", 0.0);
@@ -480,8 +479,8 @@ public class AfterMatchSkipITCase extends TestLogger {
Lists.newArrayList(a3, b4)));
}
- @Test(expected = FlinkRuntimeException.class)
- public void testSkipToFirstElementOfMatch() throws Exception {
+ @Test
+ void testSkipToFirstElementOfMatch() {
List<StreamRecord<Event>> streamEvents = new ArrayList<>();
Event a1 = new Event(1, "a1", 0.0);
@@ -494,24 +493,28 @@ public class AfterMatchSkipITCase extends TestLogger {
.where(SimpleCondition.of(value -> value.getName().contains("a")));
NFATestHarness nfaTestHarness = NFATestHarness.forPattern(pattern).build();
- List<List<Event>> resultingPatterns = nfaTestHarness.feedRecords(streamEvents);
-
// skip to first element of a match should throw exception if they are enabled,
// this mode is used in MATCH RECOGNIZE which assumes that skipping to first element
// would result in infinite loop. In CEP by default(with exceptions disabled), we use no
// skip
// strategy in this case.
+ assertThatThrownBy(() -> nfaTestHarness.feedRecords(streamEvents))
+ .isInstanceOf(FlinkRuntimeException.class);
}
- @Test(expected = FlinkRuntimeException.class)
- public void testSkipToFirstNonExistentPosition() throws Exception {
- MissedSkipTo.compute(AfterMatchSkipStrategy.skipToFirst("b").throwExceptionOnMiss());
-
+ @Test
+ void testSkipToFirstNonExistentPosition() {
// exception should be thrown
+ assertThatThrownBy(
+ () ->
+ MissedSkipTo.compute(
+ AfterMatchSkipStrategy.skipToFirst("b")
+ .throwExceptionOnMiss()))
+ .isInstanceOf(FlinkRuntimeException.class);
}
@Test
- public void testSkipToFirstNonExistentPositionWithoutException() throws Exception {
+ void testSkipToFirstNonExistentPositionWithoutException() throws Exception {
List<List<Event>> resultingPatterns =
MissedSkipTo.compute(AfterMatchSkipStrategy.skipToFirst("b"));
@@ -520,15 +523,19 @@ public class AfterMatchSkipITCase extends TestLogger {
Collections.singletonList(Lists.newArrayList(MissedSkipTo.a, MissedSkipTo.c)));
}
- @Test(expected = FlinkRuntimeException.class)
- public void testSkipToLastNonExistentPosition() throws Exception {
- MissedSkipTo.compute(AfterMatchSkipStrategy.skipToLast("b").throwExceptionOnMiss());
-
+ @Test
+ void testSkipToLastNonExistentPosition() {
// exception should be thrown
+ assertThatThrownBy(
+ () ->
+ MissedSkipTo.compute(
+ AfterMatchSkipStrategy.skipToLast("b")
+ .throwExceptionOnMiss()))
+ .isInstanceOf(FlinkRuntimeException.class);
}
@Test
- public void testSkipToLastNonExistentPositionWithoutException() throws Exception {
+ void testSkipToLastNonExistentPositionWithoutException() throws Exception {
List<List<Event>> resultingPatterns =
MissedSkipTo.compute(AfterMatchSkipStrategy.skipToFirst("b"));
@@ -567,7 +574,7 @@ public class AfterMatchSkipITCase extends TestLogger {
}
@Test
- public void testSkipToLastWithOneOrMore() throws Exception {
+ void testSkipToLastWithOneOrMore() throws Exception {
List<StreamRecord<Event>> streamEvents = new ArrayList<>();
Event a1 = new Event(1, "a1", 0.0);
@@ -607,7 +614,7 @@ public class AfterMatchSkipITCase extends TestLogger {
/** Example from docs. */
@Test
- public void testSkipPastLastWithOneOrMoreAtBeginning() throws Exception {
+ void testSkipPastLastWithOneOrMoreAtBeginning() throws Exception {
List<StreamRecord<Event>> streamEvents = new ArrayList<>();
Event a1 = new Event(1, "a1", 0.0);
@@ -638,7 +645,7 @@ public class AfterMatchSkipITCase extends TestLogger {
/** Example from docs. */
@Test
- public void testSkipToLastWithOneOrMoreAtBeginning() throws Exception {
+ void testSkipToLastWithOneOrMoreAtBeginning() throws Exception {
List<StreamRecord<Event>> streamEvents = new ArrayList<>();
Event a1 = new Event(1, "a1", 0.0);
@@ -670,7 +677,7 @@ public class AfterMatchSkipITCase extends TestLogger {
/** Example from docs. */
@Test
- public void testSkipToFirstWithOneOrMoreAtBeginning() throws Exception {
+ void testSkipToFirstWithOneOrMoreAtBeginning() throws Exception {
List<StreamRecord<Event>> streamEvents = new ArrayList<>();
Event a1 = new Event(1, "a1", 0.0);
@@ -705,7 +712,7 @@ public class AfterMatchSkipITCase extends TestLogger {
/** Example from docs. */
@Test
- public void testNoSkipWithOneOrMoreAtBeginning() throws Exception {
+ void testNoSkipWithOneOrMoreAtBeginning() throws Exception {
List<StreamRecord<Event>> streamEvents = new ArrayList<>();
Event a1 = new Event(1, "a1", 0.0);
@@ -740,7 +747,7 @@ public class AfterMatchSkipITCase extends TestLogger {
/** Example from docs. */
@Test
- public void testSkipToFirstDiscarding() throws Exception {
+ void testSkipToFirstDiscarding() throws Exception {
List<StreamRecord<Event>> streamEvents = new ArrayList<>();
Event a = new Event(1, "a", 0.0);
@@ -788,7 +795,7 @@ public class AfterMatchSkipITCase extends TestLogger {
}
@Test
- public void testSkipBeforeOtherAlreadyCompleted() throws Exception {
+ void testSkipBeforeOtherAlreadyCompleted() throws Exception {
List<StreamRecord<Event>> streamEvents = new ArrayList<>();
Event a1 = new Event(1, "a1", 0.0);
@@ -834,7 +841,7 @@ public class AfterMatchSkipITCase extends TestLogger {
}
@Test
- public void testSharedBufferIsProperlyCleared() throws Exception {
+ void testSharedBufferIsProperlyCleared() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
for (int i = 0; i < 4; i++) {
@@ -854,6 +861,6 @@ public class AfterMatchSkipITCase extends TestLogger {
nfaTestHarness.feedRecords(inputEvents);
- assertThat(sharedBuffer.isEmpty(), Matchers.is(true));
+ assertThat(sharedBuffer.isEmpty()).isTrue();
}
}
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/DeweyNumberTest.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/DeweyNumberTest.java
index fabdca0bb18..f18a5bcc180 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/DeweyNumberTest.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/DeweyNumberTest.java
@@ -18,19 +18,16 @@
package org.apache.flink.cep.nfa;
-import org.apache.flink.util.TestLogger;
+import org.junit.jupiter.api.Test;
-import org.junit.Test;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link DeweyNumber}. */
-public class DeweyNumberTest extends TestLogger {
+class DeweyNumberTest {
@Test
- public void testDeweyNumberGeneration() {
+ void testDeweyNumberGeneration() {
DeweyNumber start = new DeweyNumber(1);
DeweyNumber increased = start.increase();
DeweyNumber increaseAddStage = increased.addStage();
@@ -38,24 +35,27 @@ public class DeweyNumberTest extends TestLogger {
DeweyNumber startAddStageIncreased = startAddStage.increase();
DeweyNumber startAddStageIncreasedAddStage = startAddStageIncreased.addStage();
- assertEquals(DeweyNumber.fromString("1"), start);
- assertEquals(DeweyNumber.fromString("2"), increased);
- assertEquals(DeweyNumber.fromString("2.0"), increaseAddStage);
- assertEquals(DeweyNumber.fromString("1.0"), startAddStage);
- assertEquals(DeweyNumber.fromString("1.1"), startAddStageIncreased);
- assertEquals(DeweyNumber.fromString("1.1.0"), startAddStageIncreasedAddStage);
-
- assertTrue(startAddStage.isCompatibleWith(start));
- assertTrue(startAddStageIncreased.isCompatibleWith(startAddStage));
- assertTrue(startAddStageIncreasedAddStage.isCompatibleWith(startAddStageIncreased));
- assertFalse(startAddStageIncreasedAddStage.isCompatibleWith(startAddStage));
- assertFalse(increaseAddStage.isCompatibleWith(startAddStage));
- assertFalse(startAddStage.isCompatibleWith(increaseAddStage));
- assertFalse(startAddStageIncreased.isCompatibleWith(startAddStageIncreasedAddStage));
+ assertThat(start).isEqualTo(DeweyNumber.fromString("1"));
+ assertThat(increased).isEqualTo(DeweyNumber.fromString("2"));
+ assertThat(increaseAddStage).isEqualTo(DeweyNumber.fromString("2.0"));
+ assertThat(startAddStage).isEqualTo(DeweyNumber.fromString("1.0"));
+ assertThat(startAddStageIncreased).isEqualTo(DeweyNumber.fromString("1.1"));
+ assertThat(startAddStageIncreasedAddStage).isEqualTo(DeweyNumber.fromString("1.1.0"));
+
+ assertThat(startAddStage.isCompatibleWith(start)).isTrue();
+ assertThat(startAddStageIncreased.isCompatibleWith(startAddStage)).isTrue();
+ assertThat(startAddStageIncreasedAddStage.isCompatibleWith(startAddStageIncreased))
+ .isTrue();
+ assertThat(startAddStageIncreasedAddStage.isCompatibleWith(startAddStage)).isFalse();
+ assertThat(increaseAddStage.isCompatibleWith(startAddStage)).isFalse();
+ assertThat(startAddStage.isCompatibleWith(increaseAddStage)).isFalse();
+ assertThat(startAddStageIncreased.isCompatibleWith(startAddStageIncreasedAddStage))
+ .isFalse();
}
- @Test(expected = IllegalArgumentException.class)
- public void testZeroSplitsDeweyNumber() {
- DeweyNumber.fromString(".");
+ @Test
+ void testZeroSplitsDeweyNumber() {
+ assertThatThrownBy(() -> DeweyNumber.fromString("."))
+ .isInstanceOf(IllegalArgumentException.class);
}
}
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/GreedyITCase.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/GreedyITCase.java
index 52a6002deda..747c325577b 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/GreedyITCase.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/GreedyITCase.java
@@ -22,11 +22,10 @@ import org.apache.flink.cep.Event;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-import org.apache.flink.util.TestLogger;
import org.apache.flink.shaded.guava33.com.google.common.collect.Lists;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.List;
@@ -36,10 +35,10 @@ import static org.apache.flink.cep.utils.NFATestUtilities.feedNFA;
import static org.apache.flink.cep.utils.NFAUtils.compile;
/** IT tests covering {@link Pattern#greedy()}. */
-public class GreedyITCase extends TestLogger {
+class GreedyITCase {
@Test
- public void testGreedyZeroOrMore() throws Exception {
+ void testGreedyZeroOrMore() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event c = new Event(40, "c", 1.0);
@@ -76,7 +75,7 @@ public class GreedyITCase extends TestLogger {
}
@Test
- public void testGreedyZeroOrMoreInBetween() throws Exception {
+ void testGreedyZeroOrMoreInBetween() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event c = new Event(40, "c", 1.0);
@@ -116,7 +115,7 @@ public class GreedyITCase extends TestLogger {
}
@Test
- public void testGreedyZeroOrMoreWithDummyEventsAfterQuantifier() throws Exception {
+ void testGreedyZeroOrMoreWithDummyEventsAfterQuantifier() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event c = new Event(40, "c", 1.0);
@@ -152,7 +151,7 @@ public class GreedyITCase extends TestLogger {
}
@Test
- public void testGreedyZeroOrMoreWithDummyEventsBeforeQuantifier() throws Exception {
+ void testGreedyZeroOrMoreWithDummyEventsBeforeQuantifier() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event c = new Event(40, "c", 1.0);
@@ -183,7 +182,7 @@ public class GreedyITCase extends TestLogger {
}
@Test
- public void testGreedyUntilZeroOrMoreWithDummyEventsAfterQuantifier() throws Exception {
+ void testGreedyUntilZeroOrMoreWithDummyEventsAfterQuantifier() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event c = new Event(40, "c", 1.0);
@@ -222,7 +221,7 @@ public class GreedyITCase extends TestLogger {
}
@Test
- public void testGreedyUntilWithDummyEventsBeforeQuantifier() throws Exception {
+ void testGreedyUntilWithDummyEventsBeforeQuantifier() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event c = new Event(40, "c", 1.0);
@@ -260,7 +259,7 @@ public class GreedyITCase extends TestLogger {
}
@Test
- public void testGreedyOneOrMore() throws Exception {
+ void testGreedyOneOrMore() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event c = new Event(40, "c", 1.0);
@@ -296,7 +295,7 @@ public class GreedyITCase extends TestLogger {
}
@Test
- public void testGreedyOneOrMoreInBetween() throws Exception {
+ void testGreedyOneOrMoreInBetween() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event c = new Event(40, "c", 1.0);
@@ -335,7 +334,7 @@ public class GreedyITCase extends TestLogger {
}
@Test
- public void testGreedyOneOrMoreWithDummyEventsAfterQuantifier() throws Exception {
+ void testGreedyOneOrMoreWithDummyEventsAfterQuantifier() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event c = new Event(40, "c", 1.0);
@@ -370,7 +369,7 @@ public class GreedyITCase extends TestLogger {
}
@Test
- public void testGreedyOneOrMoreWithDummyEventsBeforeQuantifier() throws Exception {
+ void testGreedyOneOrMoreWithDummyEventsBeforeQuantifier() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event c = new Event(40, "c", 1.0);
@@ -399,7 +398,7 @@ public class GreedyITCase extends TestLogger {
}
@Test
- public void testGreedyUntilOneOrMoreWithDummyEventsAfterQuantifier() throws Exception {
+ void testGreedyUntilOneOrMoreWithDummyEventsAfterQuantifier() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event c = new Event(40, "c", 1.0);
@@ -437,7 +436,7 @@ public class GreedyITCase extends TestLogger {
}
@Test
- public void testGreedyUntilOneOrMoreWithDummyEventsBeforeQuantifier() throws Exception {
+ void testGreedyUntilOneOrMoreWithDummyEventsBeforeQuantifier() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event c = new Event(40, "c", 1.0);
@@ -473,7 +472,7 @@ public class GreedyITCase extends TestLogger {
}
@Test
- public void testGreedyZeroOrMoreBeforeGroupPattern() throws Exception {
+ void testGreedyZeroOrMoreBeforeGroupPattern() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event c = new Event(40, "c", 1.0);
@@ -530,7 +529,7 @@ public class GreedyITCase extends TestLogger {
}
@Test
- public void testEndWithZeroOrMoreGreedy() throws Exception {
+ void testEndWithZeroOrMoreGreedy() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event c = new Event(40, "c", 1.0);
@@ -568,7 +567,7 @@ public class GreedyITCase extends TestLogger {
}
@Test
- public void testEndWithZeroOrMoreConsecutiveGreedy() throws Exception {
+ void testEndWithZeroOrMoreConsecutiveGreedy() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event c = new Event(40, "c", 1.0);
@@ -606,7 +605,7 @@ public class GreedyITCase extends TestLogger {
}
@Test
- public void testEndWithGreedyTimesRange() throws Exception {
+ void testEndWithGreedyTimesRange() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event c = new Event(40, "c", 1.0);
@@ -644,7 +643,7 @@ public class GreedyITCase extends TestLogger {
}
@Test
- public void testGreedyTimesRange() throws Exception {
+ void testGreedyTimesRange() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event c = new Event(40, "c", 1.0);
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/GroupITCase.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/GroupITCase.java
index bdeb56a01f8..c5c64dbba6c 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/GroupITCase.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/GroupITCase.java
@@ -24,11 +24,10 @@ import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.cep.utils.NFATestHarness;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-import org.apache.flink.util.TestLogger;
import org.apache.flink.shaded.guava33.com.google.common.collect.Lists;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.List;
@@ -36,14 +35,14 @@ import java.util.List;
import static org.apache.flink.cep.utils.NFATestUtilities.comparePatterns;
import static org.apache.flink.cep.utils.NFATestUtilities.feedNFA;
import static org.apache.flink.cep.utils.NFAUtils.compile;
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
/** IT tests covering {@link GroupPattern}. */
@SuppressWarnings("unchecked")
-public class GroupITCase extends TestLogger {
+class GroupITCase {
@Test
- public void testGroupFollowedByTimes() throws Exception {
+ void testGroupFollowedByTimes() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event c = new Event(40, "c", 1.0);
@@ -87,7 +86,7 @@ public class GroupITCase extends TestLogger {
}
@Test
- public void testGroupFollowedByOptional() throws Exception {
+ void testGroupFollowedByOptional() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event c = new Event(40, "c", 1.0);
@@ -130,7 +129,7 @@ public class GroupITCase extends TestLogger {
}
@Test
- public void testFollowedByGroupTimesOptional() throws Exception {
+ void testFollowedByGroupTimesOptional() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event c = new Event(40, "c", 1.0);
@@ -168,7 +167,7 @@ public class GroupITCase extends TestLogger {
}
@Test
- public void testGroupFollowedByOneOrMore() throws Exception {
+ void testGroupFollowedByOneOrMore() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event c = new Event(40, "c", 1.0);
@@ -214,7 +213,7 @@ public class GroupITCase extends TestLogger {
}
@Test
- public void testGroupFollowedByZeroOrMore() throws Exception {
+ void testGroupFollowedByZeroOrMore() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event c = new Event(40, "c", 1.0);
@@ -262,7 +261,7 @@ public class GroupITCase extends TestLogger {
}
@Test
- public void testGroupFollowedByAnyTimesCombinations() throws Exception {
+ void testGroupFollowedByAnyTimesCombinations() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event c = new Event(40, "c", 1.0);
@@ -314,7 +313,7 @@ public class GroupITCase extends TestLogger {
}
@Test
- public void testGroupFollowedByAnyTimesOptional() throws Exception {
+ void testGroupFollowedByAnyTimesOptional() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event c = new Event(40, "c", 1.0);
@@ -366,7 +365,7 @@ public class GroupITCase extends TestLogger {
}
@Test
- public void testGroupFollowedByAnyOneOrMore() throws Exception {
+ void testGroupFollowedByAnyOneOrMore() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event c = new Event(40, "c", 1.0);
@@ -420,7 +419,7 @@ public class GroupITCase extends TestLogger {
}
@Test
- public void testGroupNextZeroOrMore() throws Exception {
+ void testGroupNextZeroOrMore() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event c = new Event(40, "c", 1.0);
@@ -474,7 +473,7 @@ public class GroupITCase extends TestLogger {
}
@Test
- public void testGroupNotFollowedBy() throws Exception {
+ void testGroupNotFollowedBy() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event c = new Event(40, "c", 1.0);
@@ -521,7 +520,7 @@ public class GroupITCase extends TestLogger {
}
@Test
- public void testGroupNotNext() throws Exception {
+ void testGroupNotNext() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event c = new Event(40, "c", 1.0);
@@ -568,7 +567,7 @@ public class GroupITCase extends TestLogger {
}
@Test
- public void testGroupNest() throws Exception {
+ void testGroupNest() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event d = new Event(40, "d", 1.0);
@@ -631,7 +630,7 @@ public class GroupITCase extends TestLogger {
}
@Test
- public void testGroupNestTimes() throws Exception {
+ void testGroupNestTimes() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event d = new Event(40, "d", 1.0);
@@ -712,7 +711,7 @@ public class GroupITCase extends TestLogger {
}
@Test
- public void testGroupNestTimesConsecutive() throws Exception {
+ void testGroupNestTimesConsecutive() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event d = new Event(40, "d", 1.0);
@@ -793,7 +792,7 @@ public class GroupITCase extends TestLogger {
}
@Test
- public void testGroupBegin() throws Exception {
+ void testGroupBegin() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event a1 = new Event(41, "a", 2.0);
@@ -836,7 +835,7 @@ public class GroupITCase extends TestLogger {
}
@Test
- public void testGroupFollowedByOneOrMoreWithUntilCondition() throws Exception {
+ void testGroupFollowedByOneOrMoreWithUntilCondition() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event c = new Event(40, "c", 1.0);
@@ -884,12 +883,12 @@ public class GroupITCase extends TestLogger {
Lists.newArrayList(c, a1, b1, d),
Lists.newArrayList(c, a1, b1, a2, b2, d)));
- assertEquals(1, nfaState.getPartialMatches().size());
- assertEquals("start", nfaState.getPartialMatches().peek().getCurrentStateName());
+ assertThat(nfaState.getPartialMatches()).hasSize(1);
+ assertThat(nfaState.getPartialMatches().peek().getCurrentStateName()).isEqualTo("start");
}
@Test
- public void testGroupStartsWithOptionalPattern() throws Exception {
+ void testGroupStartsWithOptionalPattern() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event a = new Event(40, "a", 1.0);
@@ -929,7 +928,7 @@ public class GroupITCase extends TestLogger {
}
@Test
- public void testFollowedByOptionalGroupPattern() throws Exception {
+ void testFollowedByOptionalGroupPattern() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event a = new Event(40, "a", 1.0);
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/IterativeConditionsITCase.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/IterativeConditionsITCase.java
index ab7df15aae6..9390a02265d 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/IterativeConditionsITCase.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/IterativeConditionsITCase.java
@@ -24,11 +24,10 @@ import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-import org.apache.flink.util.TestLogger;
import org.apache.flink.shaded.guava33.com.google.common.collect.Lists;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.List;
@@ -39,7 +38,7 @@ import static org.apache.flink.cep.utils.NFAUtils.compile;
/** IT tests covering {@link IterativeCondition} usage. */
@SuppressWarnings("unchecked")
-public class IterativeConditionsITCase extends TestLogger {
+class IterativeConditionsITCase {
////////////////////// Iterative BooleanConditions /////////////////////////
@@ -55,7 +54,7 @@ public class IterativeConditionsITCase extends TestLogger {
private final Event endEvent = new Event(46, "end", 1.0);
@Test
- public void testIterativeWithBranchingPatternEager() throws Exception {
+ void testIterativeWithBranchingPatternEager() throws Exception {
List<List<Event>> actual = testIterativeWithBranchingPattern(true);
comparePatterns(
@@ -70,7 +69,7 @@ public class IterativeConditionsITCase extends TestLogger {
}
@Test
- public void testIterativeWithBranchingPatternCombinations() throws Exception {
+ void testIterativeWithBranchingPatternCombinations() throws Exception {
List<List<Event>> actual = testIterativeWithBranchingPattern(false);
comparePatterns(
@@ -143,7 +142,7 @@ public class IterativeConditionsITCase extends TestLogger {
}
@Test
- public void testIterativeWithLoopingStartingEager() throws Exception {
+ void testIterativeWithLoopingStartingEager() throws Exception {
List<List<Event>> actual = testIterativeWithLoopingStarting(true);
comparePatterns(
@@ -157,7 +156,7 @@ public class IterativeConditionsITCase extends TestLogger {
}
@Test
- public void testIterativeWithLoopingStartingCombination() throws Exception {
+ void testIterativeWithLoopingStartingCombination() throws Exception {
List<List<Event>> actual = testIterativeWithLoopingStarting(false);
comparePatterns(
@@ -226,7 +225,7 @@ public class IterativeConditionsITCase extends TestLogger {
}
@Test
- public void testIterativeWithPrevPatternDependency() throws Exception {
+ void testIterativeWithPrevPatternDependency() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(startEvent1, 1L));
@@ -270,7 +269,7 @@ public class IterativeConditionsITCase extends TestLogger {
}
@Test
- public void testIterativeWithABACPattern() throws Exception {
+ void testIterativeWithABACPattern() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(startEvent1, 1L)); // 1
@@ -332,7 +331,7 @@ public class IterativeConditionsITCase extends TestLogger {
}
@Test
- public void testIterativeWithPrevPatternDependencyAfterBranching() throws Exception {
+ void testIterativeWithPrevPatternDependencyAfterBranching() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(startEvent1, 1L));
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAITCase.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAITCase.java
index 43363b7ab9e..8957137dd01 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAITCase.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAITCase.java
@@ -32,14 +32,12 @@ import org.apache.flink.cep.utils.NFATestHarness;
import org.apache.flink.cep.utils.TestSharedBuffer;
import org.apache.flink.cep.utils.TestTimerService;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-import org.apache.flink.util.TestLogger;
import org.apache.flink.shaded.guava33.com.google.common.collect.Lists;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import java.time.Duration;
@@ -56,9 +54,7 @@ import java.util.stream.Collectors;
import static org.apache.flink.cep.utils.NFATestUtilities.comparePatterns;
import static org.apache.flink.cep.utils.NFATestUtilities.feedNFA;
import static org.apache.flink.cep.utils.NFAUtils.compile;
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.anyLong;
/**
@@ -66,24 +62,24 @@ import static org.mockito.ArgumentMatchers.anyLong;
* NotPatternITCase}, {@link SameElementITCase} for more specific tests.
*/
@SuppressWarnings("unchecked")
-public class NFAITCase extends TestLogger {
+class NFAITCase {
private SharedBuffer<Event> sharedBuffer;
private SharedBufferAccessor<Event> sharedBufferAccessor;
- @Before
- public void init() {
+ @BeforeEach
+ void init() {
sharedBuffer = TestSharedBuffer.createTestBuffer(Event.createTypeSerializer());
sharedBufferAccessor = sharedBuffer.getAccessor();
}
- @After
- public void clear() throws Exception {
+ @AfterEach
+ void clear() throws Exception {
sharedBufferAccessor.close();
}
@Test
- public void testNoConditionNFA() throws Exception {
+ void testNoConditionNFA() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event a = new Event(40, "a", 1.0);
@@ -114,7 +110,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testNoConditionLoopingNFA() throws Exception {
+ void testNoConditionLoopingNFA() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event a = new Event(40, "a", 1.0);
@@ -151,7 +147,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testAnyWithNoConditionNFA() throws Exception {
+ void testAnyWithNoConditionNFA() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event a = new Event(40, "a", 1.0);
@@ -188,7 +184,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testSimplePatternNFA() throws Exception {
+ void testSimplePatternNFA() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(41, "start", 1.0);
@@ -222,7 +218,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testStrictContinuityWithResults() throws Exception {
+ void testStrictContinuityWithResults() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event middleEvent1 = new Event(41, "a", 2.0);
@@ -247,7 +243,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testStrictContinuityNoResults() throws Exception {
+ void testStrictContinuityNoResults() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event middleEvent1 = new Event(41, "a", 2.0);
@@ -276,7 +272,7 @@ public class NFAITCase extends TestLogger {
* length between the first and last event.
*/
@Test
- public void testSimplePatternWithTimeWindowNFAWithinFirstAndLast() throws Exception {
+ void testSimplePatternWithTimeWindowNFAWithinFirstAndLast() throws Exception {
List<StreamRecord<Event>> events = new ArrayList<>();
final Event startEvent;
@@ -314,7 +310,7 @@ public class NFAITCase extends TestLogger {
* length between the previous and current event.
*/
@Test
- public void testSimplePatternWithTimeWindowNFAWithinPreviousAndCurrent() throws Exception {
+ void testSimplePatternWithTimeWindowNFAWithinPreviousAndCurrent() throws Exception {
List<StreamRecord<Event>> events = new ArrayList<>();
final Event startEvent1;
@@ -354,7 +350,7 @@ public class NFAITCase extends TestLogger {
* out within the first and last event.
*/
@Test
- public void testSimplePatternWithTimeoutHandlingWithinFirstAndLast() throws Exception {
+ void testSimplePatternWithTimeoutHandlingWithinFirstAndLast() throws Exception {
List<StreamRecord<Event>> events = new ArrayList<>();
List<Map<String, List<Event>>> resultingPatterns = new ArrayList<>();
Set<Tuple2<Map<String, List<Event>>, Long>> resultingTimeoutPatterns = new HashSet<>();
@@ -421,10 +417,10 @@ public class NFAITCase extends TestLogger {
resultingTimeoutPatterns.addAll(timeoutPatterns);
}
- assertEquals(1, resultingPatterns.size());
- assertEquals(expectedTimeoutPatterns.size(), resultingTimeoutPatterns.size());
+ assertThat(resultingPatterns).hasSize(1);
+ assertThat(resultingTimeoutPatterns).hasSize(expectedTimeoutPatterns.size());
- assertEquals(expectedTimeoutPatterns, resultingTimeoutPatterns);
+ assertThat(resultingTimeoutPatterns).isEqualTo(expectedTimeoutPatterns);
}
/**
@@ -432,7 +428,7 @@ public class NFAITCase extends TestLogger {
* out within the previous and current event.
*/
@Test
- public void testSimplePatternWithTimeoutHandlingWithinPreviousAndCurrent() throws Exception {
+ void testSimplePatternWithTimeoutHandlingWithinPreviousAndCurrent() throws Exception {
List<StreamRecord<Event>> events = new ArrayList<>();
List<Map<String, List<Event>>> resultingPatterns = new ArrayList<>();
Set<Tuple2<Map<String, List<Event>>, Long>> resultingTimeoutPatterns = new HashSet<>();
@@ -491,19 +487,19 @@ public class NFAITCase extends TestLogger {
resultingTimeoutPatterns.addAll(timeoutPatterns);
}
- assertEquals(2, resultingPatterns.size());
- assertEquals(expectedTimeoutPatterns.size(), resultingTimeoutPatterns.size());
+ assertThat(resultingPatterns).hasSize(2);
+ assertThat(resultingTimeoutPatterns).hasSize(expectedTimeoutPatterns.size());
- assertEquals(expectedTimeoutPatterns, resultingTimeoutPatterns);
+ assertThat(resultingTimeoutPatterns).isEqualTo(expectedTimeoutPatterns);
}
@Test
- public void testPendingStateMatchesWithinFirstAndLast() throws Exception {
+ void testPendingStateMatchesWithinFirstAndLast() throws Exception {
testPendingStateMatches(WithinType.FIRST_AND_LAST);
}
@Test
- public void testPendingStateMatchesWithinPreviousAndCurrent() throws Exception {
+ void testPendingStateMatchesWithinPreviousAndCurrent() throws Exception {
testPendingStateMatches(WithinType.PREVIOUS_AND_CURRENT);
}
@@ -556,13 +552,13 @@ public class NFAITCase extends TestLogger {
new TestTimerService());
}
- assertEquals(2, resultingPendingMatches.size());
- assertEquals(expectedPendingMatches.size(), resultingPendingMatches.size());
- assertEquals(expectedPendingMatches, resultingPendingMatches);
+ assertThat(resultingPendingMatches).hasSize(2);
+ assertThat(resultingPendingMatches).hasSize(expectedPendingMatches.size());
+ assertThat(resultingPendingMatches).isEqualTo(expectedPendingMatches);
}
@Test
- public void testBranchingPattern() throws Exception {
+ void testBranchingPattern() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "start", 1.0);
@@ -609,7 +605,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testComplexBranchingAfterZeroOrMore() throws Exception {
+ void testComplexBranchingAfterZeroOrMore() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -691,7 +687,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testZeroOrMore() throws Exception {
+ void testZeroOrMore() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -729,7 +725,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testEagerZeroOrMore() throws Exception {
+ void testEagerZeroOrMore() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -770,7 +766,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testBeginWithZeroOrMore() throws Exception {
+ void testBeginWithZeroOrMore() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event middleEvent1 = new Event(40, "a", 2.0);
@@ -808,7 +804,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testZeroOrMoreAfterZeroOrMore() throws Exception {
+ void testZeroOrMoreAfterZeroOrMore() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -856,7 +852,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testZeroOrMoreAfterBranching() throws Exception {
+ void testZeroOrMoreAfterBranching() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -910,7 +906,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testStrictContinuityNoResultsAfterZeroOrMore() throws Exception {
+ void testStrictContinuityNoResultsAfterZeroOrMore() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event start = new Event(40, "d", 2.0);
@@ -943,7 +939,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testStrictContinuityResultsAfterZeroOrMore() throws Exception {
+ void testStrictContinuityResultsAfterZeroOrMore() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event start = new Event(40, "d", 2.0);
@@ -979,7 +975,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testAtLeastOne() throws Exception {
+ void testAtLeastOne() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -1015,7 +1011,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testBeginWithAtLeastOne() throws Exception {
+ void testBeginWithAtLeastOne() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent1 = new Event(41, "a", 2.0);
@@ -1053,7 +1049,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testNextZeroOrMore() throws Exception {
+ void testNextZeroOrMore() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "start", 1.0);
@@ -1090,7 +1086,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testAtLeastOneEager() throws Exception {
+ void testAtLeastOneEager() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -1131,7 +1127,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testOptional() throws Exception {
+ void testOptional() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -1163,7 +1159,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testTimes() throws Exception {
+ void testTimes() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -1200,7 +1196,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testStartWithTimes() throws Exception {
+ void testStartWithTimes() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event middleEvent1 = new Event(41, "a", 2.0);
@@ -1233,7 +1229,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testTimesNonStrictWithNext() throws Exception {
+ void testTimesNonStrictWithNext() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
@@ -1274,7 +1270,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testTimesNotStrictWithFollowedByEager() throws Exception {
+ void testTimesNotStrictWithFollowedByEager() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
@@ -1307,7 +1303,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testTimesNotStrictWithFollowedByNotEager() throws Exception {
+ void testTimesNotStrictWithFollowedByNotEager() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
@@ -1351,7 +1347,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testTimesStrictWithNextAndConsecutive() throws Exception {
+ void testTimesStrictWithNextAndConsecutive() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
@@ -1379,7 +1375,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testStartWithOptional() throws Exception {
+ void testStartWithOptional() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -1406,7 +1402,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testEndWithZeroOrMore() throws Exception {
+ void testEndWithZeroOrMore() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -1441,7 +1437,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testStartAndEndWithZeroOrMore() throws Exception {
+ void testStartAndEndWithZeroOrMore() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -1482,7 +1478,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testEndWithOptional() throws Exception {
+ void testEndWithOptional() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -1510,7 +1506,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testEndWithOneOrMore() throws Exception {
+ void testEndWithOneOrMore() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -1546,7 +1542,7 @@ public class NFAITCase extends TestLogger {
// ////////////////////////////////////////
@Test
- public void testTimesNonStrictOptional1() throws Exception {
+ void testTimesNonStrictOptional1() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
@@ -1574,7 +1570,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testTimesNonStrictOptional2() throws Exception {
+ void testTimesNonStrictOptional2() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
@@ -1622,7 +1618,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testTimesNonStrictOptional3() throws Exception {
+ void testTimesNonStrictOptional3() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
@@ -1664,7 +1660,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testTimesStrictOptional() throws Exception {
+ void testTimesStrictOptional() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
@@ -1702,7 +1698,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testOneOrMoreStrictOptional() throws Exception {
+ void testOneOrMoreStrictOptional() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
@@ -1752,7 +1748,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testTimesStrictOptional1() throws Exception {
+ void testTimesStrictOptional1() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
@@ -1788,7 +1784,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testOptionalTimesNonStrictWithNext() throws Exception {
+ void testOptionalTimesNonStrictWithNext() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
@@ -1845,7 +1841,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testStrictOneOrMore() throws Exception {
+ void testStrictOneOrMore() throws Exception {
List<List<Event>> resultingPatterns = testOneOrMore(Quantifier.ConsumingStrategy.STRICT);
comparePatterns(
@@ -1869,7 +1865,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testSkipTillNextOneOrMore() throws Exception {
+ void testSkipTillNextOneOrMore() throws Exception {
List<List<Event>> resultingPatterns =
testOneOrMore(Quantifier.ConsumingStrategy.SKIP_TILL_NEXT);
@@ -1901,7 +1897,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testSkipTillAnyOneOrMore() throws Exception {
+ void testSkipTillAnyOneOrMore() throws Exception {
List<List<Event>> resultingPatterns =
testOneOrMore(Quantifier.ConsumingStrategy.SKIP_TILL_ANY);
@@ -1995,7 +1991,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testStrictEagerZeroOrMore() throws Exception {
+ void testStrictEagerZeroOrMore() throws Exception {
List<List<Event>> resultingPatterns = testZeroOrMore(Quantifier.ConsumingStrategy.STRICT);
comparePatterns(
@@ -2020,7 +2016,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testSkipTillAnyZeroOrMore() throws Exception {
+ void testSkipTillAnyZeroOrMore() throws Exception {
List<List<Event>> resultingPatterns =
testZeroOrMore(Quantifier.ConsumingStrategy.SKIP_TILL_ANY);
@@ -2075,7 +2071,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testSkipTillNextZeroOrMore() throws Exception {
+ void testSkipTillNextZeroOrMore() throws Exception {
List<List<Event>> resultingPatterns =
testZeroOrMore(Quantifier.ConsumingStrategy.SKIP_TILL_NEXT);
@@ -2149,7 +2145,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testTimesStrict() throws Exception {
+ void testTimesStrict() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
@@ -2185,7 +2181,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testTimesNonStrict() throws Exception {
+ void testTimesNonStrict() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
@@ -2231,7 +2227,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testStartWithZeroOrMoreStrict() throws Exception {
+ void testStartWithZeroOrMoreStrict() throws Exception {
Pattern<Event, ?> pattern =
Pattern.<Event>begin("start")
.where(SimpleCondition.of(value -> value.getName().equals("a")))
@@ -2243,7 +2239,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testStartWithOneOrMoreStrict() throws Exception {
+ void testStartWithOneOrMoreStrict() throws Exception {
Pattern<Event, ?> pattern =
Pattern.<Event>begin("start")
@@ -2281,12 +2277,12 @@ public class NFAITCase extends TestLogger {
// ////////////////////////////////////////
@Test
- public void testTimesClearingBufferWithinFirstAndLast() throws Exception {
+ void testTimesClearingBufferWithinFirstAndLast() throws Exception {
testTimesClearingBuffer(WithinType.FIRST_AND_LAST);
}
@Test
- public void testTimesClearingBufferWithinPreviousAndCurrent() throws Exception {
+ void testTimesClearingBufferWithinPreviousAndCurrent() throws Exception {
testTimesClearingBuffer(WithinType.PREVIOUS_AND_CURRENT);
}
@@ -2322,17 +2318,17 @@ public class NFAITCase extends TestLogger {
// pruning element
nfa.advanceTime(sharedBufferAccessor, nfaState, 10, AfterMatchSkipStrategy.noSkip());
- assertEquals(1, nfaState.getPartialMatches().size());
- assertEquals("start", nfaState.getPartialMatches().peek().getCurrentStateName());
+ assertThat(nfaState.getPartialMatches()).hasSize(1);
+ assertThat(nfaState.getPartialMatches().peek().getCurrentStateName()).isEqualTo("start");
}
@Test
- public void testOptionalClearingBufferWithinFirstAndLast() throws Exception {
+ void testOptionalClearingBufferWithinFirstAndLast() throws Exception {
testOptionalClearingBuffer(WithinType.FIRST_AND_LAST);
}
@Test
- public void testOptionalClearingBufferWithinPreviousAndCurrent() throws Exception {
+ void testOptionalClearingBufferWithinPreviousAndCurrent() throws Exception {
testOptionalClearingBuffer(WithinType.PREVIOUS_AND_CURRENT);
}
@@ -2363,17 +2359,17 @@ public class NFAITCase extends TestLogger {
// pruning element
nfa.advanceTime(sharedBufferAccessor, nfaState, 10, AfterMatchSkipStrategy.noSkip());
- assertEquals(1, nfaState.getPartialMatches().size());
- assertEquals("start", nfaState.getPartialMatches().peek().getCurrentStateName());
+ assertThat(nfaState.getPartialMatches()).hasSize(1);
+ assertThat(nfaState.getPartialMatches().peek().getCurrentStateName()).isEqualTo("start");
}
@Test
- public void testAtLeastOneClearingBufferWithinFirstAndLast() throws Exception {
+ void testAtLeastOneClearingBufferWithinFirstAndLast() throws Exception {
testAtLeastOneClearingBuffer(WithinType.FIRST_AND_LAST);
}
@Test
- public void testAtLeastOneClearingBufferWithPreviousAndCurrent() throws Exception {
+ void testAtLeastOneClearingBufferWithPreviousAndCurrent() throws Exception {
testAtLeastOneClearingBuffer(WithinType.PREVIOUS_AND_CURRENT);
}
@@ -2407,12 +2403,12 @@ public class NFAITCase extends TestLogger {
// pruning element
nfa.advanceTime(sharedBufferAccessor, nfaState, 10, AfterMatchSkipStrategy.noSkip());
- assertEquals(1, nfaState.getPartialMatches().size());
- assertEquals("start", nfaState.getPartialMatches().peek().getCurrentStateName());
+ assertThat(nfaState.getPartialMatches()).hasSize(1);
+ assertThat(nfaState.getPartialMatches().peek().getCurrentStateName()).isEqualTo("start");
}
@Test
- public void testZeroOrMoreClearingBufferWithinFirstAndLast() throws Exception {
+ void testZeroOrMoreClearingBufferWithinFirstAndLast() throws Exception {
Event startEvent = new Event(40, "c", 1.0);
Event middleEvent1 = new Event(41, "a", 2.0);
Event middleEvent2 = new Event(42, "a", 3.0);
@@ -2443,12 +2439,12 @@ public class NFAITCase extends TestLogger {
// pruning element
nfa.advanceTime(sharedBufferAccessor, nfaState, 10, AfterMatchSkipStrategy.noSkip());
- assertEquals(1, nfaState.getPartialMatches().size());
- assertEquals("start", nfaState.getPartialMatches().peek().getCurrentStateName());
+ assertThat(nfaState.getPartialMatches()).hasSize(1);
+ assertThat(nfaState.getPartialMatches().peek().getCurrentStateName()).isEqualTo("start");
}
@Test
- public void testZeroOrMoreClearingBufferWithinPreviousAndCurrent() throws Exception {
+ void testZeroOrMoreClearingBufferWithinPreviousAndCurrent() throws Exception {
Event startEvent = new Event(40, "c", 1.0);
Event middleEvent1 = new Event(41, "a", 2.0);
Event middleEvent2 = new Event(42, "a", 3.0);
@@ -2479,18 +2475,18 @@ public class NFAITCase extends TestLogger {
// pruning element
nfa.advanceTime(sharedBufferAccessor, nfaState, 10, AfterMatchSkipStrategy.noSkip());
- assertEquals(3, nfaState.getPartialMatches().size());
- assertEquals(
- "middle:0middle:0start",
- nfaState.getPartialMatches().stream()
- .map(c -> c.getCurrentStateName())
- .collect(Collectors.joining()));
+ assertThat(nfaState.getPartialMatches()).hasSize(3);
+ assertThat(
+ nfaState.getPartialMatches().stream()
+ .map(c -> c.getCurrentStateName())
+ .collect(Collectors.joining()))
+ .isEqualTo("middle:0middle:0start");
}
/////////////////////////////////////// Skip till next /////////////////////////////
@Test
- public void testBranchingPatternSkipTillNext() throws Exception {
+ void testBranchingPatternSkipTillNext() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "start", 1.0);
@@ -2532,7 +2528,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testBranchingPatternMixedFollowedBy() throws Exception {
+ void testBranchingPatternMixedFollowedBy() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "start", 1.0);
@@ -2576,7 +2572,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testMultipleTakesVersionCollision() throws Exception {
+ void testMultipleTakesVersionCollision() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -2727,7 +2723,7 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testNFAResultOrdering() throws Exception {
+ void testNFAResultOrdering() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent1 = new Event(41, "a-1", 2.0);
@@ -2760,20 +2756,17 @@ public class NFAITCase extends TestLogger {
Collection<Map<String, List<Event>>> resultingPatterns =
nfaTestHarness.consumeRecords(inputEvents);
- Assert.assertEquals(1L, resultingPatterns.size());
+ assertThat(resultingPatterns).hasSize(1);
Map<String, List<Event>> match = resultingPatterns.iterator().next();
- Assert.assertArrayEquals(
- match.get("start").toArray(),
- Lists.newArrayList(startEvent1, startEvent2, startEvent3, startEvent4).toArray());
+ assertThat(match.get("start"))
+ .containsExactly(startEvent1, startEvent2, startEvent3, startEvent4);
- Assert.assertArrayEquals(
- match.get("middle").toArray(),
- Lists.newArrayList(endEvent1, endEvent2, endEvent3).toArray());
+ assertThat(match.get("middle")).containsExactly(endEvent1, endEvent2, endEvent3);
}
@Test
- public void testNFAResultKeyOrdering() throws Exception {
+ void testNFAResultKeyOrdering() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event a1 = new Event(41, "a", 2.0);
@@ -2804,7 +2797,7 @@ public class NFAITCase extends TestLogger {
Collection<Map<String, List<Event>>> resultingPatterns =
nfaTestHarness.consumeRecords(inputEvents);
- Assert.assertEquals(1L, resultingPatterns.size());
+ assertThat(resultingPatterns).hasSize(1);
Map<String, List<Event>> match = resultingPatterns.iterator().next();
@@ -2814,11 +2807,11 @@ public class NFAITCase extends TestLogger {
resultOrder.add(key);
}
- Assert.assertEquals(expectedOrder, resultOrder);
+ assertThat(resultOrder).isEqualTo(expectedOrder);
}
@Test
- public void testSharedBufferClearing() throws Exception {
+ void testSharedBufferClearing() throws Exception {
Pattern<Event, ?> pattern = Pattern.<Event>begin("start").followedBy("end");
Event a = new Event(40, "a", 1.0);
@@ -2849,12 +2842,12 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testLoopClearingWithinFirstAndLast() throws Exception {
+ void testLoopClearingWithinFirstAndLast() throws Exception {
testLoopClearing(WithinType.FIRST_AND_LAST);
}
@Test
- public void testLoopClearingWithinPreviousAndCurrent() throws Exception {
+ void testLoopClearingWithinPreviousAndCurrent() throws Exception {
testLoopClearing(WithinType.PREVIOUS_AND_CURRENT);
}
@@ -2877,8 +2870,7 @@ public class NFAITCase extends TestLogger {
nfa.advanceTime(accessor, nfaState, 4, AfterMatchSkipStrategy.noSkip());
}
- assertThat(
- sharedBuffer.getEventsBufferSize(),
- equalTo(withinType.equals(WithinType.FIRST_AND_LAST) ? 1 : 2));
+ assertThat(sharedBuffer.getEventsBufferSize())
+ .isEqualTo(withinType.equals(WithinType.FIRST_AND_LAST) ? 1 : 2);
}
}
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAIterativeConditionTimeContextTest.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAIterativeConditionTimeContextTest.java
index 92539808c05..25d82a669a9 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAIterativeConditionTimeContextTest.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAIterativeConditionTimeContextTest.java
@@ -24,9 +24,8 @@ import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.cep.utils.NFATestHarness;
import org.apache.flink.cep.utils.TestTimerService;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-import org.apache.flink.util.TestLogger;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.util.Collections;
import java.util.List;
@@ -36,10 +35,10 @@ import static org.apache.flink.cep.utils.NFATestHarness.forPattern;
import static org.apache.flink.cep.utils.NFATestUtilities.comparePatterns;
/** Tests for accessing time properties from {@link IterativeCondition}. */
-public class NFAIterativeConditionTimeContextTest extends TestLogger {
+class NFAIterativeConditionTimeContextTest {
@Test
- public void testEventTimestamp() throws Exception {
+ void testEventTimestamp() throws Exception {
final Event event = event().withId(1).build();
final long timestamp = 3;
@@ -64,7 +63,7 @@ public class NFAIterativeConditionTimeContextTest extends TestLogger {
}
@Test
- public void testCurrentProcessingTime() throws Exception {
+ void testCurrentProcessingTime() throws Exception {
final Event event1 = event().withId(1).build();
final Event event2 = event().withId(2).build();
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAStateAccessTest.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAStateAccessTest.java
index e6006251626..11657c15932 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAStateAccessTest.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAStateAccessTest.java
@@ -27,18 +27,18 @@ import org.apache.flink.cep.utils.NFATestHarness;
import org.apache.flink.cep.utils.TestSharedBuffer;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.List;
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests that check if we do not degrade NFA computation in case of State accesses. */
-public class NFAStateAccessTest {
+class NFAStateAccessTest {
@Test
- public void testComplexBranchingAfterZeroOrMore() throws Exception {
+ void testComplexBranchingAfterZeroOrMore() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -80,13 +80,13 @@ public class NFAStateAccessTest {
NFATestHarness.forPattern(pattern).withSharedBuffer(sharedBuffer).build();
nfaTestHarness.consumeRecords(inputEvents);
- assertEquals(58, sharedBuffer.getStateReads());
- assertEquals(41, sharedBuffer.getStateWrites());
- assertEquals(99, sharedBuffer.getStateAccesses());
+ assertThat(sharedBuffer.getStateReads()).isEqualTo(58);
+ assertThat(sharedBuffer.getStateWrites()).isEqualTo(41);
+ assertThat(sharedBuffer.getStateAccesses()).isEqualTo(99);
}
@Test
- public void testIterativeWithABACPattern() throws Exception {
+ void testIterativeWithABACPattern() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
final Event startEvent1 = new Event(40, "start", 1.0);
@@ -148,8 +148,8 @@ public class NFAStateAccessTest {
NFATestHarness.forPattern(pattern).withSharedBuffer(sharedBuffer).build();
nfaTestHarness.consumeRecords(inputEvents);
- assertEquals(90, sharedBuffer.getStateReads());
- assertEquals(34, sharedBuffer.getStateWrites());
- assertEquals(124, sharedBuffer.getStateAccesses());
+ assertThat(sharedBuffer.getStateReads()).isEqualTo(90);
+ assertThat(sharedBuffer.getStateWrites()).isEqualTo(34);
+ assertThat(sharedBuffer.getStateAccesses()).isEqualTo(124);
}
}
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAStatusChangeITCase.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAStatusChangeITCase.java
index 52b48068e11..d21d466285f 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAStatusChangeITCase.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAStatusChangeITCase.java
@@ -30,9 +30,9 @@ import org.apache.flink.cep.time.TimerService;
import org.apache.flink.cep.utils.TestSharedBuffer;
import org.apache.flink.cep.utils.TestTimerService;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
import java.time.Duration;
import java.util.Collection;
@@ -40,30 +40,29 @@ import java.util.List;
import java.util.Map;
import static org.apache.flink.cep.utils.NFAUtils.compile;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests if the {@link NFAState} status is changed after processing events. */
-public class NFAStatusChangeITCase {
+class NFAStatusChangeITCase {
private SharedBuffer<Event> sharedBuffer;
private SharedBufferAccessor<Event> sharedBufferAccessor;
private AfterMatchSkipStrategy skipStrategy = AfterMatchSkipStrategy.noSkip();
private TimerService timerService = new TestTimerService();
- @Before
- public void init() {
+ @BeforeEach
+ void init() {
this.sharedBuffer = TestSharedBuffer.createTestBuffer(Event.createTypeSerializer());
sharedBufferAccessor = sharedBuffer.getAccessor();
}
- @After
- public void clear() throws Exception {
+ @AfterEach
+ void clear() throws Exception {
sharedBufferAccessor.close();
}
@Test
- public void testNFAChange() throws Exception {
+ void testNFAChange() throws Exception {
Pattern<Event, ?> pattern =
Pattern.<Event>begin("start")
.where(SimpleCondition.of(value -> value.getName().equals("a")))
@@ -119,9 +118,10 @@ public class NFAStatusChangeITCase {
1L,
skipStrategy,
timerService);
- assertFalse(
- "NFA status should not change as the event does not match the take condition of the 'start' state",
- nfaState.isStateChanged());
+ assertThat(nfaState.isStateChanged())
+ .as(
+ "NFA status should not change as the event does not match the take condition of the 'start' state")
+ .isFalse();
nfaState.resetStateChanged();
nfa.process(
@@ -131,9 +131,10 @@ public class NFAStatusChangeITCase {
2L,
skipStrategy,
timerService);
- assertTrue(
- "NFA status should change as the event matches the take condition of the 'start' state",
- nfaState.isStateChanged());
+ assertThat(nfaState.isStateChanged())
+ .as(
+ "NFA status should change as the event matches the take condition of the 'start' state")
+ .isTrue();
// the status of the queue of ComputationStatus changed,
// more than one ComputationStatus is generated by the event from some ComputationStatus
@@ -145,9 +146,10 @@ public class NFAStatusChangeITCase {
3L,
skipStrategy,
timerService);
- assertTrue(
- "NFA status should change as the event matches the ignore condition and proceed condition of the 'middle:1' state",
- nfaState.isStateChanged());
+ assertThat(nfaState.isStateChanged())
+ .as(
+ "NFA status should change as the event matches the ignore condition and proceed condition of the 'middle:1' state")
+ .isTrue();
// both the queue of ComputationStatus and eventSharedBuffer have not changed
nfaState.resetStateChanged();
@@ -158,9 +160,10 @@ public class NFAStatusChangeITCase {
4L,
skipStrategy,
timerService);
- assertFalse(
- "NFA status should not change as the event only matches the ignore condition of the 'middle:2' state and the target state is still 'middle:2'",
- nfaState.isStateChanged());
+ assertThat(nfaState.isStateChanged())
+ .as(
+ "NFA status should not change as the event only matches the ignore condition of the 'middle:2' state and the target state is still 'middle:2'")
+ .isFalse();
// both the queue of ComputationStatus and eventSharedBuffer have changed
nfaState.resetStateChanged();
@@ -171,9 +174,10 @@ public class NFAStatusChangeITCase {
5L,
skipStrategy,
timerService);
- assertTrue(
- "NFA status should change as the event matches the take condition of 'middle:2' state",
- nfaState.isStateChanged());
+ assertThat(nfaState.isStateChanged())
+ .as(
+ "NFA status should change as the event matches the take condition of 'middle:2' state")
+ .isTrue();
// both the queue of ComputationStatus and eventSharedBuffer have changed
nfaState.resetStateChanged();
@@ -184,17 +188,18 @@ public class NFAStatusChangeITCase {
6L,
skipStrategy,
timerService);
- assertTrue(
- "NFA status should change as the event matches the take condition of 'middle2' state",
- nfaState.isStateChanged());
+ assertThat(nfaState.isStateChanged())
+ .as(
+ "NFA status should change as the event matches the take condition of 'middle2' state")
+ .isTrue();
// both the queue of ComputationStatus and eventSharedBuffer have not changed
// as the timestamp is within the window
nfaState.resetStateChanged();
nfa.advanceTime(sharedBufferAccessor, nfaState, 8L, skipStrategy);
- assertFalse(
- "NFA status should not change as the timestamp is within the window",
- nfaState.isStateChanged());
+ assertThat(nfaState.isStateChanged())
+ .as("NFA status should not change as the timestamp is within the window")
+ .isFalse();
// timeout ComputationStatus will be removed from the queue of ComputationStatus and timeout
// event will
@@ -202,13 +207,13 @@ public class NFAStatusChangeITCase {
nfaState.resetStateChanged();
Collection<Tuple2<Map<String, List<Event>>, Long>> timeoutResults =
nfa.advanceTime(sharedBufferAccessor, nfaState, 12L, skipStrategy).f1;
- assertTrue(
- "NFA status should change as timeout happens",
- nfaState.isStateChanged() && !timeoutResults.isEmpty());
+ assertThat(nfaState.isStateChanged() && !timeoutResults.isEmpty())
+ .as("NFA status should change as timeout happens")
+ .isTrue();
}
@Test
- public void testNFAChangedOnOneNewComputationState() throws Exception {
+ void testNFAChangedOnOneNewComputationState() throws Exception {
Pattern<Event, ?> pattern =
Pattern.<Event>begin("start")
.where(SimpleCondition.of(value -> value.getName().equals("start")))
@@ -251,11 +256,11 @@ public class NFAStatusChangeITCase {
7L,
skipStrategy,
timerService);
- assertTrue(nfaState.isStateChanged());
+ assertThat(nfaState.isStateChanged()).isTrue();
}
@Test
- public void testNFAChangedOnTimeoutWithoutPrune() throws Exception {
+ void testNFAChangedOnTimeoutWithoutPrune() throws Exception {
Pattern<Event, ?> pattern =
Pattern.<Event>begin("start")
.where(
@@ -296,6 +301,6 @@ public class NFAStatusChangeITCase {
nfaState.resetStateChanged();
nfa.advanceTime(sharedBufferAccessor, nfaState, 17L, skipStrategy);
- assertTrue(nfaState.isStateChanged());
+ assertThat(nfaState.isStateChanged()).isTrue();
}
}
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFATest.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFATest.java
index c1467602901..a5e2fef45f7 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFATest.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFATest.java
@@ -28,9 +28,8 @@ import org.apache.flink.cep.utils.NFATestHarness;
import org.apache.flink.core.memory.DataInputViewStreamWrapper;
import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-import org.apache.flink.util.TestLogger;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
@@ -42,12 +41,12 @@ import java.util.List;
import java.util.Map;
import static org.apache.flink.cep.utils.NFAUtils.compile;
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests for {@link NFA}. */
-public class NFATest extends TestLogger {
+class NFATest {
@Test
- public void testSimpleNFA() throws Exception {
+ void testSimpleNFA() throws Exception {
List<StreamRecord<Event>> streamEvents = new ArrayList<>();
streamEvents.add(new StreamRecord<>(new Event(1, "start", 1.0), 1L));
@@ -87,11 +86,11 @@ public class NFATest extends TestLogger {
Collection<Map<String, List<Event>>> actualPatterns =
nfaTestHarness.consumeRecords(streamEvents);
- assertEquals(expectedPatterns, actualPatterns);
+ assertThat(actualPatterns).isEqualTo(expectedPatterns);
}
@Test
- public void testTimeoutWindowPruningWithinFirstAndLast() throws Exception {
+ void testTimeoutWindowPruningWithinFirstAndLast() throws Exception {
List<StreamRecord<Event>> streamEvents = new ArrayList<>();
streamEvents.add(new StreamRecord<>(new Event(1, "start", 1.0), 1L));
@@ -113,11 +112,11 @@ public class NFATest extends TestLogger {
Collection<Map<String, List<Event>>> actualPatterns =
nfaTestHarness.consumeRecords(streamEvents);
- assertEquals(expectedPatterns, actualPatterns);
+ assertThat(actualPatterns).isEqualTo(expectedPatterns);
}
@Test
- public void testTimeoutWindowPruningWithinPreviousAndNext() throws Exception {
+ void testTimeoutWindowPruningWithinPreviousAndNext() throws Exception {
List<StreamRecord<Event>> streamEvents = new ArrayList<>();
streamEvents.add(new StreamRecord<>(new Event(1, "start", 1.0), 1L));
@@ -147,7 +146,7 @@ public class NFATest extends TestLogger {
Collection<Map<String, List<Event>>> actualPatterns =
nfaTestHarness.consumeRecords(streamEvents);
- assertEquals(expectedPatterns, actualPatterns);
+ assertThat(actualPatterns).isEqualTo(expectedPatterns);
}
/**
@@ -155,7 +154,7 @@ public class NFATest extends TestLogger {
* The reason is that the right window side (later elements) is exclusive.
*/
@Test
- public void testWindowBorders() throws Exception {
+ void testWindowBorders() throws Exception {
List<StreamRecord<Event>> streamEvents = new ArrayList<>();
streamEvents.add(new StreamRecord<>(new Event(1, "start", 1.0), 1L));
@@ -169,7 +168,7 @@ public class NFATest extends TestLogger {
Collection<Map<String, List<Event>>> actualPatterns =
nfaTestHarness.consumeRecords(streamEvents);
- assertEquals(expectedPatterns, actualPatterns);
+ assertThat(actualPatterns).isEqualTo(expectedPatterns);
}
/**
@@ -177,7 +176,7 @@ public class NFATest extends TestLogger {
* semantics (left side inclusive and right side exclusive).
*/
@Test
- public void testTimeoutWindowPruningWindowBorders() throws Exception {
+ void testTimeoutWindowPruningWindowBorders() throws Exception {
List<StreamRecord<Event>> streamEvents = new ArrayList<>();
streamEvents.add(new StreamRecord<>(new Event(1, "start", 1.0), 1L));
@@ -199,11 +198,11 @@ public class NFATest extends TestLogger {
Collection<Map<String, List<Event>>> actualPatterns =
nfaTestHarness.consumeRecords(streamEvents);
- assertEquals(expectedPatterns, actualPatterns);
+ assertThat(actualPatterns).isEqualTo(expectedPatterns);
}
@Test
- public void testNFASerialization() throws Exception {
+ void testNFASerialization() throws Exception {
Pattern<Event, ?> pattern1 =
Pattern.<Event>begin("start")
.where(SimpleCondition.of(value -> value.getName().equals("a")))
@@ -305,7 +304,7 @@ public class NFATest extends TestLogger {
NFAState copy =
serializer.duplicate().deserialize(new DataInputViewStreamWrapper(bais));
bais.close();
- assertEquals(nfaState, copy);
+ assertThat(copy).isEqualTo(nfaState);
}
}
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NotPatternITCase.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NotPatternITCase.java
index 4111b8d27a7..bd224c4c140 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NotPatternITCase.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NotPatternITCase.java
@@ -25,11 +25,10 @@ import org.apache.flink.cep.pattern.WithinType;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.cep.utils.NFATestHarness;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-import org.apache.flink.util.TestLogger;
import org.apache.flink.shaded.guava33.com.google.common.collect.Lists;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.time.Duration;
import java.util.ArrayList;
@@ -39,14 +38,14 @@ import java.util.List;
import static org.apache.flink.cep.utils.NFATestUtilities.comparePatterns;
import static org.apache.flink.cep.utils.NFATestUtilities.feedNFA;
import static org.apache.flink.cep.utils.NFAUtils.compile;
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests for {@link Pattern#notFollowedBy(String)} and {@link Pattern#notNext(String)}. */
@SuppressWarnings("unchecked")
-public class NotPatternITCase extends TestLogger {
+class NotPatternITCase {
@Test
- public void testNotNext() throws Exception {
+ void testNotNext() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event a1 = new Event(40, "a", 1.0);
@@ -82,7 +81,7 @@ public class NotPatternITCase extends TestLogger {
}
@Test
- public void testNotNextNoMatches() throws Exception {
+ void testNotNextNoMatches() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event a1 = new Event(40, "a", 1.0);
@@ -111,11 +110,11 @@ public class NotPatternITCase extends TestLogger {
final List<List<Event>> matches = feedNFA(inputEvents, nfa);
- assertEquals(0, matches.size());
+ assertThat(matches).isEmpty();
}
@Test
- public void testNotNextNoMatchesAtTheEnd() throws Exception {
+ void testNotNextNoMatchesAtTheEnd() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event a1 = new Event(40, "a", 1.0);
@@ -144,11 +143,11 @@ public class NotPatternITCase extends TestLogger {
final List<List<Event>> matches = feedNFA(inputEvents, nfa);
- assertEquals(0, matches.size());
+ assertThat(matches).isEmpty();
}
@Test
- public void testNotFollowedBy() throws Exception {
+ void testNotFollowedBy() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event a1 = new Event(40, "a", 1.0);
@@ -181,7 +180,7 @@ public class NotPatternITCase extends TestLogger {
}
@Test
- public void testNotFollowedByBeforeOptional() throws Exception {
+ void testNotFollowedByBeforeOptional() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event a1 = new Event(40, "a", 1.0);
@@ -215,7 +214,7 @@ public class NotPatternITCase extends TestLogger {
}
@Test
- public void testTimesWithNotFollowedBy() throws Exception {
+ void testTimesWithNotFollowedBy() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event a1 = new Event(40, "a", 1.0);
@@ -249,7 +248,7 @@ public class NotPatternITCase extends TestLogger {
}
@Test
- public void testIgnoreStateOfTimesWithNotFollowedBy() throws Exception {
+ void testIgnoreStateOfTimesWithNotFollowedBy() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event a1 = new Event(40, "a", 1.0);
@@ -288,7 +287,7 @@ public class NotPatternITCase extends TestLogger {
}
@Test
- public void testTimesWithNotFollowedByAfter() throws Exception {
+ void testTimesWithNotFollowedByAfter() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event a1 = new Event(40, "a", 1.0);
@@ -326,7 +325,7 @@ public class NotPatternITCase extends TestLogger {
}
@Test
- public void testNotFollowedByBeforeOptionalAtTheEnd() throws Exception {
+ void testNotFollowedByBeforeOptionalAtTheEnd() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event a1 = new Event(40, "a", 1.0);
@@ -359,7 +358,7 @@ public class NotPatternITCase extends TestLogger {
}
@Test
- public void testNotFollowedByBeforeOptionalTimes() throws Exception {
+ void testNotFollowedByBeforeOptionalTimes() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event a1 = new Event(40, "a", 1.0);
@@ -395,7 +394,7 @@ public class NotPatternITCase extends TestLogger {
}
@Test
- public void testNotFollowedByWithBranchingAtStart() throws Exception {
+ void testNotFollowedByWithBranchingAtStart() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event a1 = new Event(40, "a", 1.0);
@@ -444,13 +443,13 @@ public class NotPatternITCase extends TestLogger {
}
@Test
- public void testNotNextAfterOneOrMoreSkipTillNext() throws Exception {
+ void testNotNextAfterOneOrMoreSkipTillNext() throws Exception {
final List<List<Event>> matches = testNotNextAfterOneOrMore(false);
- assertEquals(0, matches.size());
+ assertThat(matches).isEmpty();
}
@Test
- public void testNotNextAfterOneOrMoreSkipTillAny() throws Exception {
+ void testNotNextAfterOneOrMoreSkipTillAny() throws Exception {
final List<List<Event>> matches = testNotNextAfterOneOrMore(true);
comparePatterns(
matches,
@@ -488,13 +487,13 @@ public class NotPatternITCase extends TestLogger {
}
@Test
- public void testNotFollowedByNextAfterOneOrMoreEager() throws Exception {
+ void testNotFollowedByNextAfterOneOrMoreEager() throws Exception {
final List<List<Event>> matches = testNotFollowedByAfterOneOrMore(true, false);
- assertEquals(0, matches.size());
+ assertThat(matches).isEmpty();
}
@Test
- public void testNotFollowedByAnyAfterOneOrMoreEager() throws Exception {
+ void testNotFollowedByAnyAfterOneOrMoreEager() throws Exception {
final List<List<Event>> matches = testNotFollowedByAfterOneOrMore(true, true);
comparePatterns(
matches,
@@ -524,13 +523,13 @@ public class NotPatternITCase extends TestLogger {
}
@Test
- public void testNotFollowedByNextAfterOneOrMoreCombinations() throws Exception {
+ void testNotFollowedByNextAfterOneOrMoreCombinations() throws Exception {
final List<List<Event>> matches = testNotFollowedByAfterOneOrMore(false, false);
- assertEquals(0, matches.size());
+ assertThat(matches).isEmpty();
}
@Test
- public void testNotFollowedByAnyAfterOneOrMoreCombinations() throws Exception {
+ void testNotFollowedByAnyAfterOneOrMoreCombinations() throws Exception {
final List<List<Event>> matches = testNotFollowedByAfterOneOrMore(false, true);
comparePatterns(
matches,
@@ -600,7 +599,7 @@ public class NotPatternITCase extends TestLogger {
}
@Test
- public void testNotFollowedByAnyBeforeOneOrMoreEager() throws Exception {
+ void testNotFollowedByAnyBeforeOneOrMoreEager() throws Exception {
final List<List<Event>> matches = testNotFollowedByBeforeOneOrMore(true, true);
comparePatterns(
@@ -629,7 +628,7 @@ public class NotPatternITCase extends TestLogger {
}
@Test
- public void testNotFollowedByAnyBeforeOneOrMoreCombinations() throws Exception {
+ void testNotFollowedByAnyBeforeOneOrMoreCombinations() throws Exception {
final List<List<Event>> matches = testNotFollowedByBeforeOneOrMore(false, true);
comparePatterns(
@@ -680,7 +679,7 @@ public class NotPatternITCase extends TestLogger {
}
@Test
- public void testNotFollowedByBeforeOneOrMoreEager() throws Exception {
+ void testNotFollowedByBeforeOneOrMoreEager() throws Exception {
final List<List<Event>> matches = testNotFollowedByBeforeOneOrMore(true, false);
comparePatterns(
@@ -709,7 +708,7 @@ public class NotPatternITCase extends TestLogger {
}
@Test
- public void testNotFollowedByBeforeOneOrMoreCombinations() throws Exception {
+ void testNotFollowedByBeforeOneOrMoreCombinations() throws Exception {
final List<List<Event>> matches = testNotFollowedByBeforeOneOrMore(false, false);
comparePatterns(
@@ -794,7 +793,7 @@ public class NotPatternITCase extends TestLogger {
}
@Test
- public void testNotFollowedByBeforeZeroOrMoreEagerSkipTillNext() throws Exception {
+ void testNotFollowedByBeforeZeroOrMoreEagerSkipTillNext() throws Exception {
final List<List<Event>> matches = testNotFollowedByBeforeZeroOrMore(true, false);
comparePatterns(
matches,
@@ -822,7 +821,7 @@ public class NotPatternITCase extends TestLogger {
}
@Test
- public void testNotFollowedByBeforeZeroOrMoreCombinationsSkipTillNext() throws Exception {
+ void testNotFollowedByBeforeZeroOrMoreCombinationsSkipTillNext() throws Exception {
final List<List<Event>> matches = testNotFollowedByBeforeZeroOrMore(false, false);
comparePatterns(
matches,
@@ -872,7 +871,7 @@ public class NotPatternITCase extends TestLogger {
}
@Test
- public void testNotFollowedByBeforeZeroOrMoreEagerSkipTillAny() throws Exception {
+ void testNotFollowedByBeforeZeroOrMoreEagerSkipTillAny() throws Exception {
final List<List<Event>> matches = testNotFollowedByBeforeZeroOrMore(true, true);
comparePatterns(
matches,
@@ -900,7 +899,7 @@ public class NotPatternITCase extends TestLogger {
}
@Test
- public void testNotFollowedByBeforeZeroOrMoreCombinationsSkipTillAny() throws Exception {
+ void testNotFollowedByBeforeZeroOrMoreCombinationsSkipTillAny() throws Exception {
final List<List<Event>> matches = testNotFollowedByBeforeZeroOrMore(false, true);
comparePatterns(
matches,
@@ -985,16 +984,16 @@ public class NotPatternITCase extends TestLogger {
}
@Test
- public void testNotFollowedByWithinFirstAndLastAtEnd() throws Exception {
+ void testNotFollowedByWithinFirstAndLastAtEnd() throws Exception {
testNotFollowedByWithinAtEnd(WithinType.FIRST_AND_LAST);
}
@Test
- public void testNotFollowedByWithinPreviousAndCurrentAtEnd() throws Exception {
+ void testNotFollowedByWithinPreviousAndCurrentAtEnd() throws Exception {
testNotFollowedByWithinAtEnd(WithinType.PREVIOUS_AND_CURRENT);
}
- public void testNotFollowedByWithinAtEnd(WithinType withinType) throws Exception {
+ void testNotFollowedByWithinAtEnd(WithinType withinType) throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event a1 = new Event(40, "a", 1.0);
@@ -1028,7 +1027,7 @@ public class NotPatternITCase extends TestLogger {
}
@Test
- public void testNotFollowByBeforeTimesWithin() throws Exception {
+ void testNotFollowByBeforeTimesWithin() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event a1 = new Event(40, "a", 1.0);
@@ -1075,7 +1074,7 @@ public class NotPatternITCase extends TestLogger {
}
@Test
- public void testNotFollowedByWithinAtEndAfterMatch() throws Exception {
+ void testNotFollowedByWithinAtEndAfterMatch() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event a1 = new Event(40, "a", 1.0);
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/SameElementITCase.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/SameElementITCase.java
index b95b0b13799..c0ee2201acb 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/SameElementITCase.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/SameElementITCase.java
@@ -24,12 +24,11 @@ import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.cep.utils.NFATestHarness;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-import org.apache.flink.util.TestLogger;
import org.apache.flink.shaded.guava33.com.google.common.collect.Iterators;
import org.apache.flink.shaded.guava33.com.google.common.collect.Lists;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.List;
@@ -37,17 +36,17 @@ import java.util.List;
import static org.apache.flink.cep.utils.NFATestUtilities.comparePatterns;
import static org.apache.flink.cep.utils.NFATestUtilities.feedNFA;
import static org.apache.flink.cep.utils.NFAUtils.compile;
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
/**
* Tests for handling Events that are equal in case of {@link Object#equals(Object)} and have same
* timestamps.
*/
@SuppressWarnings("unchecked")
-public class SameElementITCase extends TestLogger {
+class SameElementITCase {
@Test
- public void testEagerZeroOrMoreSameElement() throws Exception {
+ void testEagerZeroOrMoreSameElement() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -115,7 +114,7 @@ public class SameElementITCase extends TestLogger {
}
@Test
- public void testClearingBuffer() throws Exception {
+ void testClearingBuffer() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event a1 = new Event(40, "a", 1.0);
@@ -147,12 +146,12 @@ public class SameElementITCase extends TestLogger {
comparePatterns(
resultingPatterns,
Lists.<List<Event>>newArrayList(Lists.newArrayList(a1, b1, c1, d)));
- assertEquals(1, nfaState.getPartialMatches().size());
- assertEquals("a", nfaState.getPartialMatches().peek().getCurrentStateName());
+ assertThat(nfaState.getPartialMatches()).hasSize(1);
+ assertThat(nfaState.getPartialMatches().peek().getCurrentStateName()).isEqualTo("a");
}
@Test
- public void testClearingBufferWithUntilAtTheEnd() throws Exception {
+ void testClearingBufferWithUntilAtTheEnd() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event a1 = new Event(40, "a", 1.0);
@@ -196,12 +195,12 @@ public class SameElementITCase extends TestLogger {
Lists.newArrayList(a1, d1, d2, d3),
Lists.newArrayList(a1, d1, d2),
Lists.newArrayList(a1, d1)));
- assertEquals(1, nfaState.getPartialMatches().size());
- assertEquals("a", nfaState.getPartialMatches().peek().getCurrentStateName());
+ assertThat(nfaState.getPartialMatches()).hasSize(1);
+ assertThat(nfaState.getPartialMatches().peek().getCurrentStateName()).isEqualTo("a");
}
@Test
- public void testZeroOrMoreSameElement() throws Exception {
+ void testZeroOrMoreSameElement() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -321,7 +320,7 @@ public class SameElementITCase extends TestLogger {
}
@Test
- public void testSimplePatternWSameElement() throws Exception {
+ void testSimplePatternWSameElement() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -353,7 +352,7 @@ public class SameElementITCase extends TestLogger {
}
@Test
- public void testIterativeConditionWSameElement() throws Exception {
+ void testIterativeConditionWSameElement() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -407,7 +406,7 @@ public class SameElementITCase extends TestLogger {
}
@Test
- public void testEndWLoopingWSameElement() throws Exception {
+ void testEndWLoopingWSameElement() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -448,7 +447,7 @@ public class SameElementITCase extends TestLogger {
}
@Test
- public void testRepeatingPatternWSameElement() throws Exception {
+ void testRepeatingPatternWSameElement() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/TimesOrMoreITCase.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/TimesOrMoreITCase.java
index 78855672495..f9bab61b312 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/TimesOrMoreITCase.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/TimesOrMoreITCase.java
@@ -22,13 +22,14 @@ import org.apache.flink.cep.Event;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameter;
+import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
import org.apache.flink.shaded.guava33.com.google.common.collect.Lists;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
import java.time.Duration;
import java.util.ArrayList;
@@ -41,18 +42,18 @@ import static org.apache.flink.cep.utils.NFATestUtilities.feedNFA;
import static org.apache.flink.cep.utils.NFAUtils.compile;
/** Tests for {@link Pattern#timesOrMore(int)}. */
-@RunWith(Parameterized.class)
-public class TimesOrMoreITCase extends TestLogger {
+@ExtendWith(ParameterizedTestExtension.class)
+class TimesOrMoreITCase {
- @Parameterized.Parameter public Duration time;
+ @Parameter public Duration time;
- @Parameterized.Parameters(name = "Times Range Time: {0}")
+ @Parameters(name = "Times Range Time: {0}")
public static Collection<Duration> parameters() {
return Arrays.asList(null, Duration.ofMillis(3));
}
- @Test
- public void testTimesOrMore() throws Exception {
+ @TestTemplate
+ void testTimesOrMore() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -91,8 +92,8 @@ public class TimesOrMoreITCase extends TestLogger {
Lists.newArrayList(startEvent, middleEvent1, middleEvent3, end1)));
}
- @Test
- public void testTimesOrMoreNonStrict() throws Exception {
+ @TestTemplate
+ void testTimesOrMoreNonStrict() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
@@ -149,8 +150,8 @@ public class TimesOrMoreITCase extends TestLogger {
comparePatterns(resultingPatterns, expectedPatterns);
}
- @Test
- public void testTimesOrMoreStrict() throws Exception {
+ @TestTemplate
+ void testTimesOrMoreStrict() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
@@ -186,8 +187,8 @@ public class TimesOrMoreITCase extends TestLogger {
ConsecutiveData.end)));
}
- @Test
- public void testTimesOrMoreStrictOptional() throws Exception {
+ @TestTemplate
+ void testTimesOrMoreStrictOptional() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
@@ -225,8 +226,8 @@ public class TimesOrMoreITCase extends TestLogger {
Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.end)));
}
- @Test
- public void testTimesOrMoreStrictOptional2() throws Exception {
+ @TestTemplate
+ void testTimesOrMoreStrictOptional2() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
@@ -268,8 +269,8 @@ public class TimesOrMoreITCase extends TestLogger {
Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.end)));
}
- @Test
- public void testTimesOrMoreNonStrictOptional() throws Exception {
+ @TestTemplate
+ void testTimesOrMoreNonStrictOptional() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
@@ -297,8 +298,8 @@ public class TimesOrMoreITCase extends TestLogger {
Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.end)));
}
- @Test
- public void testTimesOrMoreNonStrictOptional2() throws Exception {
+ @TestTemplate
+ void testTimesOrMoreNonStrictOptional2() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
@@ -356,8 +357,8 @@ public class TimesOrMoreITCase extends TestLogger {
comparePatterns(resultingPatterns, expectedPatterns);
}
- @Test
- public void testTimesOrMoreNonStrictOptional3() throws Exception {
+ @TestTemplate
+ void testTimesOrMoreNonStrictOptional3() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
@@ -405,8 +406,8 @@ public class TimesOrMoreITCase extends TestLogger {
Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.end)));
}
- @Test
- public void testTimesOrMoreNonStrictWithNext() throws Exception {
+ @TestTemplate
+ void testTimesOrMoreNonStrictWithNext() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
@@ -457,8 +458,8 @@ public class TimesOrMoreITCase extends TestLogger {
comparePatterns(resultingPatterns, expectedPatterns);
}
- @Test
- public void testTimesOrMoreNotStrictWithFollowedBy() throws Exception {
+ @TestTemplate
+ void testTimesOrMoreNotStrictWithFollowedBy() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
@@ -497,8 +498,8 @@ public class TimesOrMoreITCase extends TestLogger {
ConsecutiveData.end)));
}
- @Test
- public void testTimesOrMoreNotStrictWithFollowedByAny() throws Exception {
+ @TestTemplate
+ void testTimesOrMoreNotStrictWithFollowedByAny() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/TimesRangeITCase.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/TimesRangeITCase.java
index 923f5a24576..32e7ae913dc 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/TimesRangeITCase.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/TimesRangeITCase.java
@@ -22,13 +22,14 @@ import org.apache.flink.cep.Event;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameter;
+import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
import org.apache.flink.shaded.guava33.com.google.common.collect.Lists;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
import java.time.Duration;
import java.util.ArrayList;
@@ -42,18 +43,18 @@ import static org.apache.flink.cep.utils.NFAUtils.compile;
/** Tests for {@link Pattern#times(int, int)}. */
@SuppressWarnings("unchecked")
-@RunWith(Parameterized.class)
-public class TimesRangeITCase extends TestLogger {
+@ExtendWith(ParameterizedTestExtension.class)
+class TimesRangeITCase {
- @Parameterized.Parameter public Duration time;
+ @Parameter public Duration time;
- @Parameterized.Parameters(name = "Times Range Time: {0}")
+ @Parameters(name = "Times Range Time: {0}")
public static Collection<Duration> parameters() {
return Arrays.asList(null, Duration.ofMillis(3));
}
- @Test
- public void testTimesRange() throws Exception {
+ @TestTemplate
+ void testTimesRange() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -92,8 +93,8 @@ public class TimesRangeITCase extends TestLogger {
Lists.newArrayList(startEvent, middleEvent1, end1)));
}
- @Test
- public void testTimesRangeFromZero() throws Exception {
+ @TestTemplate
+ void testTimesRangeFromZero() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -131,8 +132,8 @@ public class TimesRangeITCase extends TestLogger {
Lists.newArrayList(startEvent, end1)));
}
- @Test
- public void testTimesRangeNonStrict() throws Exception {
+ @TestTemplate
+ void testTimesRangeNonStrict() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
@@ -200,8 +201,8 @@ public class TimesRangeITCase extends TestLogger {
comparePatterns(resultingPatterns, expectedPatterns);
}
- @Test
- public void testTimesRangeStrict() throws Exception {
+ @TestTemplate
+ void testTimesRangeStrict() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
@@ -249,8 +250,8 @@ public class TimesRangeITCase extends TestLogger {
Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.end)));
}
- @Test
- public void testTimesRangeStrictOptional() throws Exception {
+ @TestTemplate
+ void testTimesRangeStrictOptional() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
@@ -299,8 +300,8 @@ public class TimesRangeITCase extends TestLogger {
Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.end)));
}
- @Test
- public void testTimesRangeStrictOptional1() throws Exception {
+ @TestTemplate
+ void testTimesRangeStrictOptional1() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
@@ -345,8 +346,8 @@ public class TimesRangeITCase extends TestLogger {
Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.end)));
}
- @Test
- public void testTimesRangeNonStrictOptional1() throws Exception {
+ @TestTemplate
+ void testTimesRangeNonStrictOptional1() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
@@ -373,8 +374,8 @@ public class TimesRangeITCase extends TestLogger {
Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.end)));
}
- @Test
- public void testTimesRangeNonStrictOptional2() throws Exception {
+ @TestTemplate
+ void testTimesRangeNonStrictOptional2() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
@@ -431,8 +432,8 @@ public class TimesRangeITCase extends TestLogger {
comparePatterns(resultingPatterns, expectedPatterns);
}
- @Test
- public void testTimesRangeNonStrictOptional3() throws Exception {
+ @TestTemplate
+ void testTimesRangeNonStrictOptional3() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
@@ -479,8 +480,8 @@ public class TimesRangeITCase extends TestLogger {
Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.end)));
}
- @Test
- public void testTimesRangeNonStrictWithNext() throws Exception {
+ @TestTemplate
+ void testTimesRangeNonStrictWithNext() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
@@ -530,8 +531,8 @@ public class TimesRangeITCase extends TestLogger {
comparePatterns(resultingPatterns, expectedPatterns);
}
- @Test
- public void testTimesRangeNotStrictWithFollowedBy() throws Exception {
+ @TestTemplate
+ void testTimesRangeNotStrictWithFollowedBy() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
@@ -569,8 +570,8 @@ public class TimesRangeITCase extends TestLogger {
ConsecutiveData.end)));
}
- @Test
- public void testTimesRangeNotStrictWithFollowedByAny() throws Exception {
+ @TestTemplate
+ void testTimesRangeNotStrictWithFollowedByAny() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/UntilConditionITCase.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/UntilConditionITCase.java
index dd0f0f3c713..5486e6a68ee 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/UntilConditionITCase.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/UntilConditionITCase.java
@@ -27,7 +27,7 @@ import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.shaded.guava33.com.google.common.collect.Lists;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.List;
@@ -35,17 +35,17 @@ import java.util.List;
import static org.apache.flink.cep.utils.NFATestUtilities.comparePatterns;
import static org.apache.flink.cep.utils.NFATestUtilities.feedNFA;
import static org.apache.flink.cep.utils.NFAUtils.compile;
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests for {@link Pattern#until(IterativeCondition)}. */
-public class UntilConditionITCase {
+class UntilConditionITCase {
/** Condition used for {@link Pattern#until(IterativeCondition)} clause. */
public static final SimpleCondition<Event> UNTIL_CONDITION =
SimpleCondition.of(value -> value.getPrice() == 5.0);
@Test
- public void testUntilConditionFollowedByOneOrMore() throws Exception {
+ void testUntilConditionFollowedByOneOrMore() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -83,12 +83,12 @@ public class UntilConditionITCase {
Lists.newArrayList(startEvent, middleEvent1, middleEvent2, breaking),
Lists.newArrayList(startEvent, middleEvent1, breaking)));
- assertEquals(1, nfaState.getPartialMatches().size());
- assertEquals("start", nfaState.getPartialMatches().peek().getCurrentStateName());
+ assertThat(nfaState.getPartialMatches()).hasSize(1);
+ assertThat(nfaState.getPartialMatches().peek().getCurrentStateName()).isEqualTo("start");
}
@Test
- public void testUntilConditionFollowedByOneOrMoreCombinations() throws Exception {
+ void testUntilConditionFollowedByOneOrMoreCombinations() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -131,12 +131,12 @@ public class UntilConditionITCase {
Lists.newArrayList(startEvent, middleEvent1, middleEvent2, breaking),
Lists.newArrayList(startEvent, middleEvent1, middleEvent3, breaking),
Lists.newArrayList(startEvent, middleEvent1, breaking)));
- assertEquals(1, nfaState.getPartialMatches().size());
- assertEquals("start", nfaState.getPartialMatches().peek().getCurrentStateName());
+ assertThat(nfaState.getPartialMatches()).hasSize(1);
+ assertThat(nfaState.getPartialMatches().peek().getCurrentStateName()).isEqualTo("start");
}
@Test
- public void testUntilConditionFollowedByOneOrMoreConsecutive() throws Exception {
+ void testUntilConditionFollowedByOneOrMoreConsecutive() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -174,12 +174,12 @@ public class UntilConditionITCase {
Lists.<List<Event>>newArrayList(
Lists.newArrayList(startEvent, middleEvent1, middleEvent2, breaking),
Lists.newArrayList(startEvent, middleEvent1, breaking)));
- assertEquals(1, nfaState.getPartialMatches().size());
- assertEquals("start", nfaState.getPartialMatches().peek().getCurrentStateName());
+ assertThat(nfaState.getPartialMatches()).hasSize(1);
+ assertThat(nfaState.getPartialMatches().peek().getCurrentStateName()).isEqualTo("start");
}
@Test
- public void testUntilConditionFollowedByOneOrMoreConsecutive2() throws Exception {
+ void testUntilConditionFollowedByOneOrMoreConsecutive2() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -218,12 +218,12 @@ public class UntilConditionITCase {
resultingPatterns,
Lists.<List<Event>>newArrayList(
Lists.newArrayList(startEvent, middleEvent1, breaking)));
- assertEquals(1, nfaState.getPartialMatches().size());
- assertEquals("start", nfaState.getPartialMatches().peek().getCurrentStateName());
+ assertThat(nfaState.getPartialMatches()).hasSize(1);
+ assertThat(nfaState.getPartialMatches().peek().getCurrentStateName()).isEqualTo("start");
}
@Test
- public void testUntilConditionFollowedByZeroOrMore() throws Exception {
+ void testUntilConditionFollowedByZeroOrMore() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -262,12 +262,12 @@ public class UntilConditionITCase {
Lists.newArrayList(startEvent, middleEvent1, middleEvent2, breaking),
Lists.newArrayList(startEvent, middleEvent1, breaking),
Lists.newArrayList(startEvent, breaking)));
- assertEquals(1, nfaState.getPartialMatches().size());
- assertEquals("start", nfaState.getPartialMatches().peek().getCurrentStateName());
+ assertThat(nfaState.getPartialMatches()).hasSize(1);
+ assertThat(nfaState.getPartialMatches().peek().getCurrentStateName()).isEqualTo("start");
}
@Test
- public void testUntilConditionFollowedByZeroOrMoreCombinations() throws Exception {
+ void testUntilConditionFollowedByZeroOrMoreCombinations() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -312,12 +312,12 @@ public class UntilConditionITCase {
Lists.newArrayList(startEvent, middleEvent1, middleEvent3, breaking),
Lists.newArrayList(startEvent, middleEvent1, breaking),
Lists.newArrayList(startEvent, breaking)));
- assertEquals(1, nfaState.getPartialMatches().size());
- assertEquals("start", nfaState.getPartialMatches().peek().getCurrentStateName());
+ assertThat(nfaState.getPartialMatches()).hasSize(1);
+ assertThat(nfaState.getPartialMatches().peek().getCurrentStateName()).isEqualTo("start");
}
@Test
- public void testUntilConditionFollowedByZeroOrMoreConsecutive() throws Exception {
+ void testUntilConditionFollowedByZeroOrMoreConsecutive() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -357,12 +357,12 @@ public class UntilConditionITCase {
Lists.newArrayList(startEvent, middleEvent1, middleEvent2, breaking),
Lists.newArrayList(startEvent, middleEvent1, breaking),
Lists.newArrayList(startEvent, breaking)));
- assertEquals(1, nfaState.getPartialMatches().size());
- assertEquals("start", nfaState.getPartialMatches().peek().getCurrentStateName());
+ assertThat(nfaState.getPartialMatches()).hasSize(1);
+ assertThat(nfaState.getPartialMatches().peek().getCurrentStateName()).isEqualTo("start");
}
@Test
- public void testUntilConditionFollowedByAnyOneOrMore() throws Exception {
+ void testUntilConditionFollowedByAnyOneOrMore() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -399,7 +399,7 @@ public class UntilConditionITCase {
}
@Test
- public void testUntilConditionFollowedByAnyZeroOrMore() throws Exception {
+ void testUntilConditionFollowedByAnyZeroOrMore() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -438,7 +438,7 @@ public class UntilConditionITCase {
}
@Test
- public void testUntilConditionWithEmptyWhere() throws Exception {
+ void testUntilConditionWithEmptyWhere() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -476,12 +476,12 @@ public class UntilConditionITCase {
Lists.newArrayList(startEvent, middleEvent1, middleEvent2),
Lists.newArrayList(startEvent, middleEvent1)));
- assertEquals(1, nfaState.getPartialMatches().size());
- assertEquals("start", nfaState.getPartialMatches().peek().getCurrentStateName());
+ assertThat(nfaState.getPartialMatches()).hasSize(1);
+ assertThat(nfaState.getPartialMatches().peek().getCurrentStateName()).isEqualTo("start");
}
@Test
- public void testIterativeUntilConditionOneOrMore() throws Exception {
+ void testIterativeUntilConditionOneOrMore() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -532,12 +532,12 @@ public class UntilConditionITCase {
Lists.newArrayList(startEvent, middleEvent1, middleEvent2),
Lists.newArrayList(startEvent, middleEvent1)));
- assertEquals(1, nfaState.getPartialMatches().size());
- assertEquals("start", nfaState.getPartialMatches().peek().getCurrentStateName());
+ assertThat(nfaState.getPartialMatches()).hasSize(1);
+ assertThat(nfaState.getPartialMatches().peek().getCurrentStateName()).isEqualTo("start");
}
@Test
- public void testIterativeUntilConditionZeroOrMore() throws Exception {
+ void testIterativeUntilConditionZeroOrMore() throws Exception {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(40, "c", 1.0);
@@ -589,7 +589,7 @@ public class UntilConditionITCase {
Lists.newArrayList(startEvent, middleEvent1),
Lists.newArrayList(startEvent)));
- assertEquals(1, nfaState.getPartialMatches().size());
- assertEquals("start", nfaState.getPartialMatches().peek().getCurrentStateName());
+ assertThat(nfaState.getPartialMatches()).hasSize(1);
+ assertThat(nfaState.getPartialMatches().peek().getCurrentStateName()).isEqualTo("start");
}
}
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/compiler/NFACompilerTest.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/compiler/NFACompilerTest.java
index 8cf47217a61..08fbf0041bb 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/compiler/NFACompilerTest.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/compiler/NFACompilerTest.java
@@ -30,13 +30,8 @@ import org.apache.flink.cep.pattern.MalformedPatternException;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.WithinType;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
-import org.apache.flink.util.TestLogger;
-import org.apache.flink.shaded.guava33.com.google.common.collect.Sets;
-
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.ExpectedException;
+import org.junit.jupiter.api.Test;
import java.time.Duration;
import java.util.Collection;
@@ -46,13 +41,12 @@ import java.util.Map;
import java.util.Set;
import static org.apache.flink.cep.utils.NFAUtils.compile;
-import static org.hamcrest.Matchers.is;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assertions.entry;
/** Tests for {@link NFACompiler}. */
-public class NFACompilerTest extends TestLogger {
+class NFACompilerTest {
private static final SimpleCondition<Event> startFilter =
SimpleCondition.of(value -> value.getPrice() > 2);
@@ -60,15 +54,8 @@ public class NFACompilerTest extends TestLogger {
private static final SimpleCondition<Event> endFilter =
SimpleCondition.of(value -> value.getName().equals("end"));
- @Rule public ExpectedException expectedException = ExpectedException.none();
-
@Test
- public void testNFACompilerUniquePatternName() {
-
- // adjust the rule
- expectedException.expect(MalformedPatternException.class);
- expectedException.expectMessage("Duplicate pattern name: start. Names must be unique.");
-
+ void testNFACompilerUniquePatternName() {
Pattern<Event, ?> invalidPattern =
Pattern.<Event>begin("start")
.where(new TestFilter())
@@ -78,17 +65,13 @@ public class NFACompilerTest extends TestLogger {
.where(new TestFilter());
// here we must have an exception because of the two "start" patterns with the same name.
- compile(invalidPattern, false);
+ assertThatThrownBy(() -> compile(invalidPattern, false))
+ .isInstanceOf(MalformedPatternException.class)
+ .hasMessageContaining("Duplicate pattern name: start. Names must be unique.");
}
@Test
- public void testNFACompilerPatternEndsWithNotFollowedBy() {
-
- // adjust the rule
- expectedException.expect(MalformedPatternException.class);
- expectedException.expectMessage(
- "NotFollowedBy is not supported without windowTime as a last part of a Pattern!");
-
+ void testNFACompilerPatternEndsWithNotFollowedBy() {
Pattern<Event, ?> invalidPattern =
Pattern.<Event>begin("start")
.where(new TestFilter())
@@ -98,7 +81,10 @@ public class NFACompilerTest extends TestLogger {
.where(new TestFilter());
// here we must have an exception because the last pattern is notFollowedBy!
- compile(invalidPattern, false);
+ assertThatThrownBy(() -> compile(invalidPattern, false))
+ .isInstanceOf(MalformedPatternException.class)
+ .hasMessageContaining(
+ "NotFollowedBy is not supported without windowTime as a last part of a Pattern!");
}
/**
@@ -117,7 +103,7 @@ public class NFACompilerTest extends TestLogger {
/** Tests that the NFACompiler generates the correct NFA from a given Pattern. */
@Test
- public void testNFACompilerWithSimplePattern() {
+ void testNFACompilerWithSimplePattern() {
Pattern<Event, Event> pattern =
Pattern.<Event>begin("start")
.where(startFilter)
@@ -129,48 +115,46 @@ public class NFACompilerTest extends TestLogger {
NFA<Event> nfa = compile(pattern, false);
Collection<State<Event>> states = nfa.getStates();
- assertEquals(4, states.size());
+ assertThat(states).hasSize(4);
Map<String, State<Event>> stateMap = new HashMap<>();
for (State<Event> state : states) {
stateMap.put(state.getName(), state);
}
- assertTrue(stateMap.containsKey("start"));
+ assertThat(stateMap).containsKey("start");
State<Event> startState = stateMap.get("start");
- assertTrue(startState.isStart());
+ assertThat(startState.isStart()).isTrue();
final Set<Tuple2<String, StateTransitionAction>> startTransitions =
unfoldTransitions(startState);
- assertEquals(
- Sets.newHashSet(Tuple2.of("middle", StateTransitionAction.TAKE)), startTransitions);
+ assertThat(startTransitions)
+ .containsExactlyInAnyOrder(Tuple2.of("middle", StateTransitionAction.TAKE));
- assertTrue(stateMap.containsKey("middle"));
+ assertThat(stateMap).containsKey("middle");
State<Event> middleState = stateMap.get("middle");
final Set<Tuple2<String, StateTransitionAction>> middleTransitions =
unfoldTransitions(middleState);
- assertEquals(
- Sets.newHashSet(
+ assertThat(middleTransitions)
+ .containsExactlyInAnyOrder(
Tuple2.of("middle", StateTransitionAction.IGNORE),
- Tuple2.of("end", StateTransitionAction.TAKE)),
- middleTransitions);
+ Tuple2.of("end", StateTransitionAction.TAKE));
- assertTrue(stateMap.containsKey("end"));
+ assertThat(stateMap).containsKey("end");
State<Event> endState = stateMap.get("end");
final Set<Tuple2<String, StateTransitionAction>> endTransitions =
unfoldTransitions(endState);
- assertEquals(
- Sets.newHashSet(
- Tuple2.of(NFACompiler.ENDING_STATE_NAME, StateTransitionAction.TAKE)),
- endTransitions);
+ assertThat(endTransitions)
+ .containsExactlyInAnyOrder(
+ Tuple2.of(NFACompiler.ENDING_STATE_NAME, StateTransitionAction.TAKE));
- assertTrue(stateMap.containsKey(NFACompiler.ENDING_STATE_NAME));
+ assertThat(stateMap).containsKey(NFACompiler.ENDING_STATE_NAME);
State<Event> endingState = stateMap.get(NFACompiler.ENDING_STATE_NAME);
- assertTrue(endingState.isFinal());
- assertEquals(0, endingState.getStateTransitions().size());
+ assertThat(endingState.isFinal()).isTrue();
+ assertThat(endingState.getStateTransitions()).isEmpty();
}
@Test
- public void testNFACompilerPatternNotFollowedByWithIn() {
+ void testNFACompilerPatternNotFollowedByWithIn() {
Pattern<Event, Event> pattern =
Pattern.<Event>begin("start")
.where(startFilter)
@@ -181,49 +165,47 @@ public class NFACompilerTest extends TestLogger {
NFA<Event> nfa = compile(pattern, false);
Collection<State<Event>> states = nfa.getStates();
- assertEquals(4, states.size());
+ assertThat(states).hasSize(4);
Map<String, State<Event>> stateMap = new HashMap<>();
for (State<Event> state : states) {
stateMap.put(state.getName(), state);
}
- assertTrue(stateMap.containsKey("start"));
+ assertThat(stateMap).containsKey("start");
State<Event> startState = stateMap.get("start");
- assertTrue(startState.isStart());
+ assertThat(startState.isStart()).isTrue();
final Set<Tuple2<String, StateTransitionAction>> startTransitions =
unfoldTransitions(startState);
- assertEquals(
- Sets.newHashSet(Tuple2.of("middle", StateTransitionAction.TAKE)), startTransitions);
+ assertThat(startTransitions)
+ .containsExactlyInAnyOrder(Tuple2.of("middle", StateTransitionAction.TAKE));
- assertTrue(stateMap.containsKey("middle"));
+ assertThat(stateMap).containsKey("middle");
State<Event> middleState = stateMap.get("middle");
- assertTrue(middleState.isPending());
+ assertThat(middleState.isPending()).isTrue();
final Set<Tuple2<String, StateTransitionAction>> middleTransitions =
unfoldTransitions(middleState);
- assertEquals(
- Sets.newHashSet(
+ assertThat(middleTransitions)
+ .containsExactlyInAnyOrder(
Tuple2.of("middle", StateTransitionAction.IGNORE),
- Tuple2.of("middle:0", StateTransitionAction.PROCEED)),
- middleTransitions);
+ Tuple2.of("middle:0", StateTransitionAction.PROCEED));
- assertTrue(stateMap.containsKey("middle:0"));
+ assertThat(stateMap).containsKey("middle:0");
State<Event> middle0State = stateMap.get("middle:0");
- assertTrue(middle0State.isStop());
+ assertThat(middle0State.isStop()).isTrue();
final Set<Tuple2<String, StateTransitionAction>> middle0Transitions =
unfoldTransitions(middle0State);
- assertEquals(
- Sets.newHashSet(Tuple2.of("middle:0", StateTransitionAction.TAKE)),
- middle0Transitions);
+ assertThat(middle0Transitions)
+ .containsExactlyInAnyOrder(Tuple2.of("middle:0", StateTransitionAction.TAKE));
- assertTrue(stateMap.containsKey(NFACompiler.ENDING_STATE_NAME));
+ assertThat(stateMap).containsKey(NFACompiler.ENDING_STATE_NAME);
State<Event> endingState = stateMap.get(NFACompiler.ENDING_STATE_NAME);
- assertTrue(endingState.isFinal());
- assertEquals(0, endingState.getStateTransitions().size());
+ assertThat(endingState.isFinal()).isTrue();
+ assertThat(endingState.getStateTransitions()).isEmpty();
}
@Test
- public void testNoUnnecessaryStateCopiesCreated() {
+ void testNoUnnecessaryStateCopiesCreated() {
final Pattern<Event, Event> pattern =
Pattern.<Event>begin("start")
.where(startFilter)
@@ -246,15 +228,11 @@ public class NFACompilerTest extends TestLogger {
}
}
- assertEquals(1, endStateCount);
+ assertThat(endStateCount).isOne();
}
@Test
- public void testSkipToNotExistsMatchingPattern() {
- expectedException.expect(MalformedPatternException.class);
- expectedException.expectMessage(
- "The pattern name specified in AfterMatchSkipStrategy can not be found in the given Pattern");
-
+ void testSkipToNotExistsMatchingPattern() {
Pattern<Event, ?> invalidPattern =
Pattern.<Event>begin("start", AfterMatchSkipStrategy.skipToLast("midd"))
.where(SimpleCondition.of(value -> value.getName().contains("a")))
@@ -265,7 +243,10 @@ public class NFACompilerTest extends TestLogger {
.next("end")
.where(SimpleCondition.of(value -> value.getName().contains("c")));
- compile(invalidPattern, false);
+ assertThatThrownBy(() -> compile(invalidPattern, false))
+ .isInstanceOf(MalformedPatternException.class)
+ .hasMessageContaining(
+ "The pattern name specified in AfterMatchSkipStrategy can not be found in the given Pattern");
}
private <T> Set<Tuple2<String, StateTransitionAction>> unfoldTransitions(final State<T> state) {
@@ -278,26 +259,25 @@ public class NFACompilerTest extends TestLogger {
}
@Test
- public void testCheckingEmptyMatches() {
- assertThat(NFACompiler.canProduceEmptyMatches(Pattern.begin("a").optional()), is(true));
- assertThat(
- NFACompiler.canProduceEmptyMatches(Pattern.begin("a").oneOrMore().optional()),
- is(true));
+ void testCheckingEmptyMatches() {
+ assertThat(NFACompiler.canProduceEmptyMatches(Pattern.begin("a").optional())).isTrue();
+ assertThat(NFACompiler.canProduceEmptyMatches(Pattern.begin("a").oneOrMore().optional()))
+ .isTrue();
assertThat(
- NFACompiler.canProduceEmptyMatches(
- Pattern.begin("a").oneOrMore().optional().next("b").optional()),
- is(true));
+ NFACompiler.canProduceEmptyMatches(
+ Pattern.begin("a").oneOrMore().optional().next("b").optional()))
+ .isTrue();
- assertThat(NFACompiler.canProduceEmptyMatches(Pattern.begin("a")), is(false));
- assertThat(NFACompiler.canProduceEmptyMatches(Pattern.begin("a").oneOrMore()), is(false));
+ assertThat(NFACompiler.canProduceEmptyMatches(Pattern.begin("a"))).isFalse();
+ assertThat(NFACompiler.canProduceEmptyMatches(Pattern.begin("a").oneOrMore())).isFalse();
assertThat(
- NFACompiler.canProduceEmptyMatches(
- Pattern.begin("a").oneOrMore().next("b").optional()),
- is(false));
+ NFACompiler.canProduceEmptyMatches(
+ Pattern.begin("a").oneOrMore().next("b").optional()))
+ .isFalse();
}
@Test
- public void testWindowTimeCorrectlySet() {
+ void testWindowTimeCorrectlySet() {
Pattern<Event, ?> pattern =
Pattern.<Event>begin("start")
.followedBy("middle")
@@ -309,11 +289,11 @@ public class NFACompilerTest extends TestLogger {
NFACompiler.NFAFactoryCompiler<Event> factory =
new NFACompiler.NFAFactoryCompiler<>(pattern);
factory.compileFactory();
- assertEquals(10000, factory.getWindowTime());
+ assertThat(factory.getWindowTime()).isEqualTo(10000);
}
@Test
- public void testWindowTimesCorrectlySet() {
+ void testWindowTimesCorrectlySet() {
Pattern<Event, ?> pattern =
Pattern.<Event>begin("start")
.followedBy("middle")
@@ -326,14 +306,14 @@ public class NFACompilerTest extends TestLogger {
new NFACompiler.NFAFactoryCompiler<>(pattern);
factory.compileFactory();
- Map<String, Long> expectedWindowTimes = new HashMap<>();
- expectedWindowTimes.put("middle", Duration.ofSeconds(10).toMillis());
- expectedWindowTimes.put("then", Duration.ofSeconds(20).toMillis());
- assertEquals(expectedWindowTimes, factory.getWindowTimes());
+ assertThat(factory.getWindowTimes())
+ .containsOnly(
+ entry("middle", Duration.ofSeconds(10).toMillis()),
+ entry("then", Duration.ofSeconds(20).toMillis()));
}
@Test
- public void testMultipleWindowTimeWithZeroLength() {
+ void testMultipleWindowTimeWithZeroLength() {
Pattern<Event, ?> pattern =
Pattern.<Event>begin("start")
.followedBy("middle")
@@ -345,15 +325,11 @@ public class NFACompilerTest extends TestLogger {
NFACompiler.NFAFactoryCompiler<Event> factory =
new NFACompiler.NFAFactoryCompiler<>(pattern);
factory.compileFactory();
- assertEquals(0, factory.getWindowTime());
+ assertThat(factory.getWindowTime()).isZero();
}
@Test
- public void testCheckPatternWindowTimes() {
- expectedException.expect(MalformedPatternException.class);
- expectedException.expectMessage(
- "The window length between the previous and current event cannot be larger than the window length between the first and last event for a Pattern.");
-
+ void testCheckPatternWindowTimes() {
Pattern<Event, ?> pattern =
Pattern.<Event>begin("start")
.followedBy("middle")
@@ -365,6 +341,10 @@ public class NFACompilerTest extends TestLogger {
NFACompiler.NFAFactoryCompiler<Event> factory =
new NFACompiler.NFAFactoryCompiler<>(pattern);
- factory.compileFactory();
+
+ assertThatThrownBy(factory::compileFactory)
+ .isInstanceOf(MalformedPatternException.class)
+ .hasMessageContaining(
+ "The window length between the previous and current event cannot be larger than the window length between the first and last event for a Pattern.");
}
}
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/sharedbuffer/LockableTypeSerializerTest.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/sharedbuffer/LockableTypeSerializerTest.java
index bff530fbdef..0e8c70eacb1 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/sharedbuffer/LockableTypeSerializerTest.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/sharedbuffer/LockableTypeSerializerTest.java
@@ -21,32 +21,34 @@ package org.apache.flink.cep.nfa.sharedbuffer;
import org.apache.flink.api.common.typeutils.base.IntSerializer;
import org.apache.flink.runtime.state.heap.TestDuplicateSerializer;
-import org.junit.Assert;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests for {@link org.apache.flink.cep.nfa.sharedbuffer.Lockable.LockableTypeSerializer}. */
-public class LockableTypeSerializerTest {
+class LockableTypeSerializerTest {
/** This tests that {@link Lockable.LockableTypeSerializer#duplicate()} works as expected. */
@Test
- public void testDuplicate() {
+ void testDuplicate() {
IntSerializer nonDuplicatingInnerSerializer = IntSerializer.INSTANCE;
- Assert.assertSame(nonDuplicatingInnerSerializer, nonDuplicatingInnerSerializer.duplicate());
+ assertThat(nonDuplicatingInnerSerializer.duplicate())
+ .isSameAs(nonDuplicatingInnerSerializer);
Lockable.LockableTypeSerializer<Integer> candidateTestShallowDuplicate =
new Lockable.LockableTypeSerializer<>(nonDuplicatingInnerSerializer);
- Assert.assertSame(candidateTestShallowDuplicate, candidateTestShallowDuplicate.duplicate());
+ assertThat(candidateTestShallowDuplicate.duplicate())
+ .isSameAs(candidateTestShallowDuplicate);
TestDuplicateSerializer duplicatingInnerSerializer = new TestDuplicateSerializer();
- Assert.assertNotSame(duplicatingInnerSerializer, duplicatingInnerSerializer.duplicate());
+ assertThat(duplicatingInnerSerializer.duplicate()).isNotSameAs(duplicatingInnerSerializer);
Lockable.LockableTypeSerializer<Integer> candidateTestDeepDuplicate =
new Lockable.LockableTypeSerializer<>(duplicatingInnerSerializer);
Lockable.LockableTypeSerializer<Integer> deepDuplicate =
candidateTestDeepDuplicate.duplicate();
- Assert.assertNotSame(candidateTestDeepDuplicate, deepDuplicate);
- Assert.assertNotSame(
- candidateTestDeepDuplicate.getElementSerializer(),
- deepDuplicate.getElementSerializer());
+ assertThat(deepDuplicate).isNotSameAs(candidateTestDeepDuplicate);
+ assertThat(deepDuplicate.getElementSerializer())
+ .isNotSameAs(candidateTestDeepDuplicate.getElementSerializer());
}
}
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/sharedbuffer/SharedBufferTest.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/sharedbuffer/SharedBufferTest.java
index ecd548634ed..775b19b5d2e 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/sharedbuffer/SharedBufferTest.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/sharedbuffer/SharedBufferTest.java
@@ -22,15 +22,15 @@ import org.apache.flink.cep.Event;
import org.apache.flink.cep.configuration.SharedBufferCacheConfig;
import org.apache.flink.cep.nfa.DeweyNumber;
import org.apache.flink.cep.utils.TestSharedBuffer;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameter;
+import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
import java.time.Duration;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
@@ -38,25 +38,23 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests for {@link SharedBuffer}. */
-@RunWith(Parameterized.class)
-public class SharedBufferTest extends TestLogger {
+@ExtendWith(ParameterizedTestExtension.class)
+class SharedBufferTest {
- @Parameterized.Parameter public SharedBufferCacheConfig cacheConfig;
+ @Parameter public SharedBufferCacheConfig cacheConfig;
- @Parameterized.Parameters
+ @Parameters
public static Collection<SharedBufferCacheConfig> prepareSharedBufferCacheConfig() {
- return Arrays.asList(
+ return List.of(
new SharedBufferCacheConfig(1, 1, Duration.ofSeconds(1L)),
new SharedBufferCacheConfig(10240, 10240, Duration.ofSeconds(1L)));
}
- @Test
- public void testSharedBuffer() throws Exception {
+ @TestTemplate
+ void testSharedBuffer() throws Exception {
SharedBuffer<Event> sharedBuffer =
TestSharedBuffer.createTestBuffer(Event.createTypeSerializer(), cacheConfig);
int numberEvents = 8;
@@ -145,25 +143,27 @@ public class SharedBufferTest extends TestLogger {
List<Map<String, List<EventId>>> patterns3 =
sharedBufferAccessor.extractPatterns(b3, b3Version);
- assertEquals(1L, patterns3.size());
- assertEquals(expectedPattern3, sharedBufferAccessor.materializeMatch(patterns3.get(0)));
+ assertThat(patterns3).hasSize(1);
+ assertThat(sharedBufferAccessor.materializeMatch(patterns3.get(0)))
+ .isEqualTo(expectedPattern3);
sharedBufferAccessor.releaseNode(b3, b3Version);
List<Map<String, List<EventId>>> patterns4 =
sharedBufferAccessor.extractPatterns(b3, b3Version);
- assertEquals(0L, patterns4.size());
- assertTrue(patterns4.isEmpty());
+ assertThat(patterns4).isEmpty();
List<Map<String, List<EventId>>> patterns1 =
sharedBufferAccessor.extractPatterns(b1, b1Version);
- assertEquals(1L, patterns1.size());
- assertEquals(expectedPattern1, sharedBufferAccessor.materializeMatch(patterns1.get(0)));
+ assertThat(patterns1).hasSize(1);
+ assertThat(sharedBufferAccessor.materializeMatch(patterns1.get(0)))
+ .isEqualTo(expectedPattern1);
sharedBufferAccessor.releaseNode(b1, b1Version);
List<Map<String, List<EventId>>> patterns2 =
sharedBufferAccessor.extractPatterns(b0, b0Version);
- assertEquals(1L, patterns2.size());
- assertEquals(expectedPattern2, sharedBufferAccessor.materializeMatch(patterns2.get(0)));
+ assertThat(patterns2).hasSize(1);
+ assertThat(sharedBufferAccessor.materializeMatch(patterns2.get(0)))
+ .isEqualTo(expectedPattern2);
sharedBufferAccessor.releaseNode(b0, b0Version);
for (EventId eventId : eventIds) {
@@ -171,11 +171,11 @@ public class SharedBufferTest extends TestLogger {
}
}
- assertTrue(sharedBuffer.isEmpty());
+ assertThat(sharedBuffer.isEmpty()).isTrue();
}
- @Test
- public void testClearingSharedBufferWithMultipleEdgesBetweenEntries() throws Exception {
+ @TestTemplate
+ void testClearingSharedBufferWithMultipleEdgesBetweenEntries() throws Exception {
SharedBuffer<Event> sharedBuffer =
TestSharedBuffer.createTestBuffer(Event.createTypeSerializer(), cacheConfig);
int numberEvents = 8;
@@ -218,11 +218,11 @@ public class SharedBufferTest extends TestLogger {
}
// There should be still events[1] and events[2] in the buffer
- assertFalse(sharedBuffer.isEmpty());
+ assertThat(sharedBuffer.isEmpty()).isFalse();
}
- @Test
- public void testSharedBufferExtractOrder() throws Exception {
+ @TestTemplate
+ void testSharedBufferExtractOrder() throws Exception {
SharedBuffer<Event> sharedBuffer =
TestSharedBuffer.createTestBuffer(Event.createTypeSerializer(), cacheConfig);
int numberEvents = 5;
@@ -277,12 +277,12 @@ public class SharedBufferTest extends TestLogger {
sharedBufferAccessor.releaseEvent(eventId);
}
List<String> resultOrder = new ArrayList<>(patternsResult.keySet());
- assertEquals(expectedOrder, resultOrder);
+ assertThat(resultOrder).isEqualTo(expectedOrder);
}
}
- @Test
- public void testSharedBufferCountersClearing() throws Exception {
+ @TestTemplate
+ void testSharedBufferCountersClearing() throws Exception {
SharedBuffer<Event> sharedBuffer =
TestSharedBuffer.createTestBuffer(Event.createTypeSerializer(), cacheConfig);
int numberEvents = 4;
@@ -297,13 +297,13 @@ public class SharedBufferTest extends TestLogger {
Iterator<Map.Entry<Long, Integer>> counters = sharedBuffer.getEventCounters();
Map.Entry<Long, Integer> entry = counters.next();
- assertEquals(3, entry.getKey().longValue());
- assertEquals(1, entry.getValue().intValue());
- assertFalse(counters.hasNext());
+ assertThat(entry.getKey().longValue()).isEqualTo(3);
+ assertThat(entry.getValue().intValue()).isOne();
+ assertThat(counters).isExhausted();
}
- @Test
- public void testSharedBufferAccessor() throws Exception {
+ @TestTemplate
+ void testSharedBufferAccessor() throws Exception {
TestSharedBuffer<Event> sharedBuffer =
TestSharedBuffer.createTestBuffer(Event.createTypeSerializer(), cacheConfig);
int numberEvents = 8;
@@ -317,8 +317,8 @@ public class SharedBufferTest extends TestLogger {
eventIds[i] = sharedBufferAccessor.registerEvent(events[i], timestamp);
}
int expectedEvents = Math.min(cacheConfig.getEventsBufferCacheSlots(), numberEvents);
- assertEquals(expectedEvents, sharedBuffer.getEventsBufferCacheSize());
- assertEquals(0, sharedBuffer.getSharedBufferNodeCacheSize());
+ assertThat(sharedBuffer.getEventsBufferCacheSize()).isEqualTo(expectedEvents);
+ assertThat(sharedBuffer.getSharedBufferNodeCacheSize()).isZero();
NodeId start =
sharedBufferAccessor.put(
@@ -336,9 +336,9 @@ public class SharedBufferTest extends TestLogger {
"branching", eventIds[4], b00, DeweyNumber.fromString("1.0.0.0"));
int expectedEntries = Math.min(cacheConfig.getEntryCacheSlots(), 4);
- assertEquals(expectedEntries, sharedBuffer.getSharedBufferNodeCacheSize());
+ assertThat(sharedBuffer.getSharedBufferNodeCacheSize()).isEqualTo(expectedEntries);
System.out.println(expectedEntries);
- assertEquals(4 - expectedEntries, sharedBuffer.getSharedBufferNodeSize());
+ assertThat(sharedBuffer.getSharedBufferNodeSize()).isEqualTo(4 - expectedEntries);
sharedBufferAccessor.lockNode(b0, DeweyNumber.fromString("1.0.0"));
@@ -346,10 +346,10 @@ public class SharedBufferTest extends TestLogger {
sharedBufferAccessor.releaseEvent(eventId);
}
}
- assertEquals(0, sharedBuffer.getEventsBufferCacheSize());
- assertEquals(4, sharedBuffer.getEventsBufferSize());
- assertEquals(0, sharedBuffer.getSharedBufferNodeCacheSize());
- assertEquals(4, sharedBuffer.getSharedBufferNodeSize());
+ assertThat(sharedBuffer.getEventsBufferCacheSize()).isZero();
+ assertThat(sharedBuffer.getEventsBufferSize()).isEqualTo(4);
+ assertThat(sharedBuffer.getSharedBufferNodeCacheSize()).isZero();
+ assertThat(sharedBuffer.getSharedBufferNodeSize()).isEqualTo(4);
}
/**
@@ -358,8 +358,8 @@ public class SharedBufferTest extends TestLogger {
*
* @throws Exception if creating the shared buffer accessor fails.
*/
- @Test
- public void testReleaseNodesWithLongPath() throws Exception {
+ @TestTemplate
+ void testReleaseNodesWithLongPath() throws Exception {
SharedBuffer<Event> sharedBuffer =
TestSharedBuffer.createTestBuffer(Event.createTypeSerializer(), cacheConfig);
@@ -392,6 +392,6 @@ public class SharedBufferTest extends TestLogger {
}
}
- assertTrue(sharedBuffer.isEmpty());
+ assertThat(sharedBuffer.isEmpty()).isTrue();
}
}
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPMigrationTest.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPMigrationTest.java
index 20011b2428e..df37356e3d0 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPMigrationTest.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPMigrationTest.java
@@ -35,10 +35,12 @@ import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.flink.streaming.util.OperatorSnapshotUtil;
import org.apache.flink.test.util.MigrationTest;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameter;
+import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
import java.time.Duration;
import java.util.Collection;
@@ -46,8 +48,7 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentLinkedQueue;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
/**
* Tests for checking whether CEP operator can restore from snapshots that were done using previous
@@ -56,21 +57,17 @@ import static org.junit.Assert.assertTrue;
* <p>For regenerating the binary snapshot file of previous versions you have to run the {@code
* write*()} method on the corresponding Flink release-* branch.
*/
-@RunWith(Parameterized.class)
-public class CEPMigrationTest implements MigrationTest {
+@ExtendWith(ParameterizedTestExtension.class)
+class CEPMigrationTest implements MigrationTest {
- private final FlinkVersion migrateVersion;
+ @Parameter public FlinkVersion migrateVersion;
- @Parameterized.Parameters(name = "Migration Savepoint: {0}")
+ @Parameters(name = "Migration Savepoint: {0}")
public static Collection<FlinkVersion> parameters() {
return FlinkVersion.rangeOf(
FlinkVersion.v2_0, MigrationTest.getMostRecentlyPublishedVersion());
}
- public CEPMigrationTest(FlinkVersion migrateVersion) {
- this.migrateVersion = migrateVersion;
- }
-
@SnapshotsGenerator
public void writeAfterBranchingPatternSnapshot(FlinkVersion flinkGenerateSavepointVersion)
throws Exception {
@@ -120,8 +117,8 @@ public class CEPMigrationTest implements MigrationTest {
}
}
- @Test
- public void testRestoreAfterBranchingPattern() throws Exception {
+ @TestTemplate
+ void testRestoreAfterBranchingPattern() throws Exception {
KeySelector<Event, Integer> keySelector =
new KeySelector<Event, Integer>() {
@@ -161,33 +158,33 @@ public class CEPMigrationTest implements MigrationTest {
ConcurrentLinkedQueue<Object> result = harness.getOutput();
// watermark and 2 results
- assertEquals(3, result.size());
+ assertThat(result).hasSize(3);
Object resultObject1 = result.poll();
- assertTrue(resultObject1 instanceof StreamRecord);
+ assertThat(resultObject1).isInstanceOf(StreamRecord.class);
StreamRecord<?> resultRecord1 = (StreamRecord<?>) resultObject1;
- assertTrue(resultRecord1.getValue() instanceof Map);
+ assertThat(resultRecord1.getValue()).isInstanceOf(Map.class);
Object resultObject2 = result.poll();
- assertTrue(resultObject2 instanceof StreamRecord);
+ assertThat(resultObject2).isInstanceOf(StreamRecord.class);
StreamRecord<?> resultRecord2 = (StreamRecord<?>) resultObject2;
- assertTrue(resultRecord2.getValue() instanceof Map);
+ assertThat(resultRecord2.getValue()).isInstanceOf(Map.class);
@SuppressWarnings("unchecked")
Map<String, List<Event>> patternMap1 =
(Map<String, List<Event>>) resultRecord1.getValue();
- assertEquals(startEvent, patternMap1.get("start").get(0));
- assertEquals(middleEvent1, patternMap1.get("middle").get(0));
- assertEquals(endEvent, patternMap1.get("end").get(0));
+ assertThat(patternMap1.get("start").get(0)).isEqualTo(startEvent);
+ assertThat(patternMap1.get("middle").get(0)).isEqualTo(middleEvent1);
+ assertThat(patternMap1.get("end").get(0)).isEqualTo(endEvent);
@SuppressWarnings("unchecked")
Map<String, List<Event>> patternMap2 =
(Map<String, List<Event>>) resultRecord2.getValue();
- assertEquals(startEvent, patternMap2.get("start").get(0));
- assertEquals(middleEvent2, patternMap2.get("middle").get(0));
- assertEquals(endEvent, patternMap2.get("end").get(0));
+ assertThat(patternMap2.get("start").get(0)).isEqualTo(startEvent);
+ assertThat(patternMap2.get("middle").get(0)).isEqualTo(middleEvent2);
+ assertThat(patternMap2.get("end").get(0)).isEqualTo(endEvent);
// and now go for a checkpoint with the new serializers
@@ -219,20 +216,20 @@ public class CEPMigrationTest implements MigrationTest {
result = harness.getOutput();
// watermark and the result
- assertEquals(2, result.size());
+ assertThat(result).hasSize(2);
Object resultObject3 = result.poll();
- assertTrue(resultObject3 instanceof StreamRecord);
+ assertThat(resultObject3).isInstanceOf(StreamRecord.class);
StreamRecord<?> resultRecord3 = (StreamRecord<?>) resultObject3;
- assertTrue(resultRecord3.getValue() instanceof Map);
+ assertThat(resultRecord3.getValue()).isInstanceOf(Map.class);
@SuppressWarnings("unchecked")
Map<String, List<Event>> patternMap3 =
(Map<String, List<Event>>) resultRecord3.getValue();
- assertEquals(startEvent1, patternMap3.get("start").get(0));
- assertEquals(middleEvent3, patternMap3.get("middle").get(0));
- assertEquals(endEvent1, patternMap3.get("end").get(0));
+ assertThat(patternMap3.get("start").get(0)).isEqualTo(startEvent1);
+ assertThat(patternMap3.get("middle").get(0)).isEqualTo(middleEvent3);
+ assertThat(patternMap3.get("end").get(0)).isEqualTo(endEvent1);
} finally {
harness.close();
}
@@ -283,8 +280,8 @@ public class CEPMigrationTest implements MigrationTest {
}
}
- @Test
- public void testRestoreStartingNewPatternAfterMigration() throws Exception {
+ @TestTemplate
+ void testRestoreStartingNewPatternAfterMigration() throws Exception {
KeySelector<Event, Integer> keySelector =
new KeySelector<Event, Integer>() {
@@ -328,46 +325,46 @@ public class CEPMigrationTest implements MigrationTest {
ConcurrentLinkedQueue<Object> result = harness.getOutput();
// watermark and 3 results
- assertEquals(4, result.size());
+ assertThat(result).hasSize(4);
Object resultObject1 = result.poll();
- assertTrue(resultObject1 instanceof StreamRecord);
+ assertThat(resultObject1).isInstanceOf(StreamRecord.class);
StreamRecord<?> resultRecord1 = (StreamRecord<?>) resultObject1;
- assertTrue(resultRecord1.getValue() instanceof Map);
+ assertThat(resultRecord1.getValue()).isInstanceOf(Map.class);
Object resultObject2 = result.poll();
- assertTrue(resultObject2 instanceof StreamRecord);
+ assertThat(resultObject2).isInstanceOf(StreamRecord.class);
StreamRecord<?> resultRecord2 = (StreamRecord<?>) resultObject2;
- assertTrue(resultRecord2.getValue() instanceof Map);
+ assertThat(resultRecord2.getValue()).isInstanceOf(Map.class);
Object resultObject3 = result.poll();
- assertTrue(resultObject3 instanceof StreamRecord);
+ assertThat(resultObject3).isInstanceOf(StreamRecord.class);
StreamRecord<?> resultRecord3 = (StreamRecord<?>) resultObject3;
- assertTrue(resultRecord3.getValue() instanceof Map);
+ assertThat(resultRecord3.getValue()).isInstanceOf(Map.class);
@SuppressWarnings("unchecked")
Map<String, List<Event>> patternMap1 =
(Map<String, List<Event>>) resultRecord1.getValue();
- assertEquals(startEvent1, patternMap1.get("start").get(0));
- assertEquals(middleEvent1, patternMap1.get("middle").get(0));
- assertEquals(endEvent, patternMap1.get("end").get(0));
+ assertThat(patternMap1.get("start").get(0)).isEqualTo(startEvent1);
+ assertThat(patternMap1.get("middle").get(0)).isEqualTo(middleEvent1);
+ assertThat(patternMap1.get("end").get(0)).isEqualTo(endEvent);
@SuppressWarnings("unchecked")
Map<String, List<Event>> patternMap2 =
(Map<String, List<Event>>) resultRecord2.getValue();
- assertEquals(startEvent1, patternMap2.get("start").get(0));
- assertEquals(middleEvent2, patternMap2.get("middle").get(0));
- assertEquals(endEvent, patternMap2.get("end").get(0));
+ assertThat(patternMap2.get("start").get(0)).isEqualTo(startEvent1);
+ assertThat(patternMap2.get("middle").get(0)).isEqualTo(middleEvent2);
+ assertThat(patternMap2.get("end").get(0)).isEqualTo(endEvent);
@SuppressWarnings("unchecked")
Map<String, List<Event>> patternMap3 =
(Map<String, List<Event>>) resultRecord3.getValue();
- assertEquals(startEvent2, patternMap3.get("start").get(0));
- assertEquals(middleEvent2, patternMap3.get("middle").get(0));
- assertEquals(endEvent, patternMap3.get("end").get(0));
+ assertThat(patternMap3.get("start").get(0)).isEqualTo(startEvent2);
+ assertThat(patternMap3.get("middle").get(0)).isEqualTo(middleEvent2);
+ assertThat(patternMap3.get("end").get(0)).isEqualTo(endEvent);
// and now go for a checkpoint with the new serializers
@@ -399,20 +396,20 @@ public class CEPMigrationTest implements MigrationTest {
result = harness.getOutput();
// watermark and the result
- assertEquals(2, result.size());
+ assertThat(result).hasSize(2);
Object resultObject4 = result.poll();
- assertTrue(resultObject4 instanceof StreamRecord);
+ assertThat(resultObject4).isInstanceOf(StreamRecord.class);
StreamRecord<?> resultRecord4 = (StreamRecord<?>) resultObject4;
- assertTrue(resultRecord4.getValue() instanceof Map);
+ assertThat(resultRecord4.getValue()).isInstanceOf(Map.class);
@SuppressWarnings("unchecked")
Map<String, List<Event>> patternMap4 =
(Map<String, List<Event>>) resultRecord4.getValue();
- assertEquals(startEvent3, patternMap4.get("start").get(0));
- assertEquals(middleEvent3, patternMap4.get("middle").get(0));
- assertEquals(endEvent1, patternMap4.get("end").get(0));
+ assertThat(patternMap4.get("start").get(0)).isEqualTo(startEvent3);
+ assertThat(patternMap4.get("middle").get(0)).isEqualTo(middleEvent3);
+ assertThat(patternMap4.get("end").get(0)).isEqualTo(endEvent1);
} finally {
harness.close();
}
@@ -458,8 +455,8 @@ public class CEPMigrationTest implements MigrationTest {
}
}
- @Test
- public void testSinglePatternAfterMigration() throws Exception {
+ @TestTemplate
+ void testSinglePatternAfterMigration() throws Exception {
KeySelector<Event, Integer> keySelector =
new KeySelector<Event, Integer>() {
@@ -498,18 +495,18 @@ public class CEPMigrationTest implements MigrationTest {
ConcurrentLinkedQueue<Object> result = harness.getOutput();
// watermark and the result
- assertEquals(2, result.size());
+ assertThat(result).hasSize(2);
Object resultObject = result.poll();
- assertTrue(resultObject instanceof StreamRecord);
+ assertThat(resultObject).isInstanceOf(StreamRecord.class);
StreamRecord<?> resultRecord = (StreamRecord<?>) resultObject;
- assertTrue(resultRecord.getValue() instanceof Map);
+ assertThat(resultRecord.getValue()).isInstanceOf(Map.class);
@SuppressWarnings("unchecked")
Map<String, List<Event>> patternMap =
(Map<String, List<Event>>) resultRecord.getValue();
- assertEquals(startEvent1, patternMap.get("start").get(0));
+ assertThat(patternMap.get("start").get(0)).isEqualTo(startEvent1);
} finally {
harness.close();
}
@@ -556,8 +553,8 @@ public class CEPMigrationTest implements MigrationTest {
}
}
- @Test
- public void testAndOrSubtypeConditionsAfterMigration() throws Exception {
+ @TestTemplate
+ void testAndOrSubtypeConditionsAfterMigration() throws Exception {
KeySelector<Event, Integer> keySelector =
new KeySelector<Event, Integer>() {
@@ -594,19 +591,19 @@ public class CEPMigrationTest implements MigrationTest {
ConcurrentLinkedQueue<Object> result = harness.getOutput();
// watermark and the result
- assertEquals(2, result.size());
+ assertThat(result).hasSize(2);
Object resultObject = result.poll();
- assertTrue(resultObject instanceof StreamRecord);
+ assertThat(resultObject).isInstanceOf(StreamRecord.class);
StreamRecord<?> resultRecord = (StreamRecord<?>) resultObject;
- assertTrue(resultRecord.getValue() instanceof Map);
+ assertThat(resultRecord.getValue()).isInstanceOf(Map.class);
@SuppressWarnings("unchecked")
Map<String, List<Event>> patternMap =
(Map<String, List<Event>>) resultRecord.getValue();
- assertEquals(startEvent1, patternMap.get("start").get(0));
- assertEquals(endEvent, patternMap.get("start").get(1));
+ assertThat(patternMap.get("start").get(0)).isEqualTo(startEvent1);
+ assertThat(patternMap.get("start").get(1)).isEqualTo(endEvent);
} finally {
harness.close();
}
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java
index df0da998e9b..b77cb345425 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java
@@ -50,17 +50,15 @@ import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.apache.flink.util.TernaryBoolean;
-import org.apache.flink.util.TestLogger;
import org.apache.flink.shaded.guava33.com.google.common.collect.Lists;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
import org.mockito.Mockito;
+import java.io.File;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
@@ -72,9 +70,7 @@ import java.util.Queue;
import static org.apache.flink.cep.utils.CepOperatorBuilder.createOperatorForNFA;
import static org.apache.flink.cep.utils.EventBuilder.event;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.spy;
@@ -82,17 +78,17 @@ import static org.mockito.Mockito.validateMockitoUsage;
import static org.mockito.Mockito.verify;
/** Tests for {@link CepOperator}. */
-public class CEPOperatorTest extends TestLogger {
+class CEPOperatorTest {
- @Rule public TemporaryFolder tempFolder = new TemporaryFolder();
+ @TempDir File tempFolder;
- @After
- public void validate() {
+ @AfterEach
+ void validate() {
validateMockitoUsage();
}
@Test
- public void testKeyedCEPOperatorWatermarkForwarding() throws Exception {
+ void testKeyedCEPOperatorWatermarkForwarding() throws Exception {
OneInputStreamOperatorTestHarness<Event, Map<String, List<Event>>> harness =
getCepTestHarness(false);
@@ -111,7 +107,7 @@ public class CEPOperatorTest extends TestLogger {
}
@Test
- public void testProcessingTimestampisPassedToNFA() throws Exception {
+ void testProcessingTimestampisPassedToNFA() throws Exception {
final NFA<Event> nfa =
NFACompiler.compileFactory(Pattern.<Event>begin("begin"), true).createNFA();
@@ -137,7 +133,7 @@ public class CEPOperatorTest extends TestLogger {
}
@Test
- public void testKeyedCEPOperatorCheckpointing() throws Exception {
+ void testKeyedCEPOperatorCheckpointing() throws Exception {
OneInputStreamOperatorTestHarness<Event, Map<String, List<Event>>> harness =
getCepTestHarness(false);
@@ -191,7 +187,7 @@ public class CEPOperatorTest extends TestLogger {
Queue<Object> result = harness.getOutput();
- assertEquals(2, result.size());
+ assertThat(result).hasSize(2);
verifyPattern(result.poll(), startEvent, middleEvent, endEvent);
verifyWatermark(result.poll(), Long.MAX_VALUE);
@@ -201,9 +197,9 @@ public class CEPOperatorTest extends TestLogger {
}
@Test
- public void testKeyedCEPOperatorCheckpointingWithRocksDB() throws Exception {
+ void testKeyedCEPOperatorCheckpointingWithRocksDB() throws Exception {
- String rocksDbPath = tempFolder.newFolder().getAbsolutePath();
+ String rocksDbPath = tempFolder.getAbsolutePath();
EmbeddedRocksDBStateBackend rocksDBStateBackend =
new EmbeddedRocksDBStateBackend(TernaryBoolean.FALSE);
rocksDBStateBackend.setDbStoragePath(rocksDbPath);
@@ -270,7 +266,7 @@ public class CEPOperatorTest extends TestLogger {
Queue<Object> result = harness.getOutput();
- assertEquals(2, result.size());
+ assertThat(result).hasSize(2);
verifyPattern(result.poll(), startEvent, middleEvent, endEvent);
verifyWatermark(result.poll(), Long.MAX_VALUE);
@@ -283,7 +279,7 @@ public class CEPOperatorTest extends TestLogger {
* Tests that the internal time of a CEP operator advances only given watermarks. See FLINK-5033
*/
@Test
- public void testKeyedAdvancingTimeWithoutElements() throws Exception {
+ void testKeyedAdvancingTimeWithoutElements() throws Exception {
final Event startEvent = new Event(42, "start", 1.0);
final long watermarkTimestamp1 = 5L;
final long watermarkTimestamp2 = 13L;
@@ -316,7 +312,7 @@ public class CEPOperatorTest extends TestLogger {
BasicTypeInfo.INT_TYPE_INFO);
try {
- String rocksDbPath = tempFolder.newFolder().getAbsolutePath();
+ String rocksDbPath = tempFolder.getAbsolutePath();
EmbeddedRocksDBStateBackend rocksDBStateBackend = new EmbeddedRocksDBStateBackend();
rocksDBStateBackend.setDbStoragePath(rocksDbPath);
@@ -336,32 +332,32 @@ public class CEPOperatorTest extends TestLogger {
Queue<StreamRecord<Tuple2<Map<String, List<Event>>, Long>>> sideOutput =
harness.getSideOutput(timedOut);
- assertEquals(2L, result.size());
- assertEquals(1L, sideOutput.size());
+ assertThat(result).hasSize(2);
+ assertThat(sideOutput).hasSize(1);
Object watermark1 = result.poll();
- assertTrue(watermark1 instanceof Watermark);
+ assertThat(watermark1).isInstanceOf(Watermark.class);
- assertEquals(watermarkTimestamp1, ((Watermark) watermark1).getTimestamp());
+ assertThat(((Watermark) watermark1).getTimestamp()).isEqualTo(watermarkTimestamp1);
Tuple2<Map<String, List<Event>>, Long> leftResult = sideOutput.poll().getValue();
- assertEquals(watermarkTimestamp2, (long) leftResult.f1);
- assertEquals(expectedSequence, leftResult.f0);
+ assertThat((long) leftResult.f1).isEqualTo(watermarkTimestamp2);
+ assertThat(leftResult.f0).isEqualTo(expectedSequence);
Object watermark2 = result.poll();
- assertTrue(watermark2 instanceof Watermark);
+ assertThat(watermark2).isInstanceOf(Watermark.class);
- assertEquals(watermarkTimestamp2, ((Watermark) watermark2).getTimestamp());
+ assertThat(((Watermark) watermark2).getTimestamp()).isEqualTo(watermarkTimestamp2);
} finally {
harness.close();
}
}
@Test
- public void testKeyedCEPOperatorNFAUpdate() throws Exception {
+ void testKeyedCEPOperatorNFAUpdate() throws Exception {
CepOperator<Event, Integer, Map<String, List<Event>>> operator =
CepOperatorTestUtilities.getKeyedCepOperator(true, new SimpleNFAFactory());
@@ -406,7 +402,7 @@ public class CEPOperatorTest extends TestLogger {
Queue<Object> result = harness.getOutput();
- assertEquals(1, result.size());
+ assertThat(result).hasSize(1);
verifyPattern(result.poll(), startEvent, middleEvent, endEvent);
} finally {
@@ -415,9 +411,9 @@ public class CEPOperatorTest extends TestLogger {
}
@Test
- public void testKeyedCEPOperatorNFAUpdateWithRocksDB() throws Exception {
+ void testKeyedCEPOperatorNFAUpdateWithRocksDB() throws Exception {
- String rocksDbPath = tempFolder.newFolder().getAbsolutePath();
+ String rocksDbPath = tempFolder.getAbsolutePath();
EmbeddedRocksDBStateBackend rocksDBStateBackend =
new EmbeddedRocksDBStateBackend(TernaryBoolean.FALSE);
rocksDBStateBackend.setDbStoragePath(rocksDbPath);
@@ -474,7 +470,7 @@ public class CEPOperatorTest extends TestLogger {
Queue<Object> result = harness.getOutput();
- assertEquals(1, result.size());
+ assertThat(result).hasSize(1);
verifyPattern(result.poll(), startEvent, middleEvent, endEvent);
} finally {
@@ -483,7 +479,7 @@ public class CEPOperatorTest extends TestLogger {
}
@Test
- public void testKeyedCEPOperatorNFAUpdateTimes() throws Exception {
+ void testKeyedCEPOperatorNFAUpdateTimes() throws Exception {
CepOperator<Event, Integer, Map<String, List<Event>>> operator =
CepOperatorTestUtilities.getKeyedCepOperator(true, new SimpleNFAFactory());
OneInputStreamOperatorTestHarness<Event, Map<String, List<Event>>> harness =
@@ -512,7 +508,7 @@ public class CEPOperatorTest extends TestLogger {
// get and verify the output
Queue<Object> result = harness.getOutput();
- assertEquals(1, result.size());
+ assertThat(result).hasSize(1);
verifyPattern(result.poll(), startEvent, middleEvent, endEvent);
} finally {
@@ -521,9 +517,9 @@ public class CEPOperatorTest extends TestLogger {
}
@Test
- public void testKeyedCEPOperatorNFAUpdateTimesWithRocksDB() throws Exception {
+ void testKeyedCEPOperatorNFAUpdateTimesWithRocksDB() throws Exception {
- String rocksDbPath = tempFolder.newFolder().getAbsolutePath();
+ String rocksDbPath = tempFolder.getAbsolutePath();
EmbeddedRocksDBStateBackend rocksDBStateBackend = new EmbeddedRocksDBStateBackend();
rocksDBStateBackend.setDbStoragePath(rocksDbPath);
@@ -558,7 +554,7 @@ public class CEPOperatorTest extends TestLogger {
// get and verify the output
Queue<Object> result = harness.getOutput();
- assertEquals(1, result.size());
+ assertThat(result).hasSize(1);
verifyPattern(result.poll(), startEvent, middleEvent, endEvent);
} finally {
@@ -567,7 +563,7 @@ public class CEPOperatorTest extends TestLogger {
}
@Test
- public void testCEPOperatorCleanupEventTime() throws Exception {
+ void testCEPOperatorCleanupEventTime() throws Exception {
Event startEvent1 = new Event(42, "start", 1.0);
Event startEvent2 = new Event(42, "start", 2.0);
@@ -598,11 +594,11 @@ public class CEPOperatorTest extends TestLogger {
// there must be 2 keys 42, 43 registered for the watermark callback
// all the seen elements must be in the priority queues but no NFA yet.
- assertEquals(4L, harness.numEventTimeTimers());
- assertEquals(4L, operator.getPQSize(42));
- assertEquals(1L, operator.getPQSize(43));
- assertTrue(!operator.hasNonEmptySharedBuffer(42));
- assertTrue(!operator.hasNonEmptySharedBuffer(43));
+ assertThat(harness.numEventTimeTimers()).isEqualTo(4L);
+ assertThat(operator.getPQSize(42)).isEqualTo(4L);
+ assertThat(operator.getPQSize(43)).isOne();
+ assertThat(operator.hasNonEmptySharedBuffer(42)).isFalse();
+ assertThat(operator.hasNonEmptySharedBuffer(43)).isFalse();
harness.processWatermark(new Watermark(2L));
@@ -613,11 +609,11 @@ public class CEPOperatorTest extends TestLogger {
// one element in PQ for 42 (the barfoo) as it arrived early
// for 43 the element entered the NFA and the PQ is empty
- assertEquals(4L, harness.numEventTimeTimers());
- assertTrue(operator.hasNonEmptySharedBuffer(42));
- assertEquals(1L, operator.getPQSize(42));
- assertTrue(operator.hasNonEmptySharedBuffer(43));
- assertTrue(!operator.hasNonEmptyPQ(43));
+ assertThat(harness.numEventTimeTimers()).isEqualTo(4L);
+ assertThat(operator.hasNonEmptySharedBuffer(42)).isTrue();
+ assertThat(operator.getPQSize(42)).isOne();
+ assertThat(operator.hasNonEmptySharedBuffer(43)).isTrue();
+ assertThat(operator.hasNonEmptyPQ(43)).isFalse();
harness.processElement(new StreamRecord<>(startEvent2, 4L));
harness.processElement(new StreamRecord<Event>(middleEvent2, 5L));
@@ -638,11 +634,11 @@ public class CEPOperatorTest extends TestLogger {
// now we have 1 key because the 43 expired and was removed.
// 42 is still there due to startEvent2
- assertEquals(3L, harness.numEventTimeTimers());
- assertTrue(operator2.hasNonEmptySharedBuffer(42));
- assertTrue(!operator2.hasNonEmptyPQ(42));
- assertTrue(!operator2.hasNonEmptySharedBuffer(43));
- assertTrue(!operator2.hasNonEmptyPQ(43));
+ assertThat(harness.numEventTimeTimers()).isEqualTo(3L);
+ assertThat(operator2.hasNonEmptySharedBuffer(42)).isTrue();
+ assertThat(operator2.hasNonEmptyPQ(42)).isFalse();
+ assertThat(operator2.hasNonEmptySharedBuffer(43)).isFalse();
+ assertThat(operator2.hasNonEmptyPQ(43)).isFalse();
verifyPattern(harness.getOutput().poll(), startEvent1, middleEvent1, endEvent1);
verifyPattern(harness.getOutput().poll(), startEvent1, middleEvent2, endEvent1);
@@ -656,11 +652,11 @@ public class CEPOperatorTest extends TestLogger {
harness.processWatermark(20L);
harness.processWatermark(21L);
- assertTrue(!operator2.hasNonEmptySharedBuffer(42));
- assertTrue(!operator2.hasNonEmptyPQ(42));
- assertEquals(0L, harness.numEventTimeTimers());
+ assertThat(operator2.hasNonEmptySharedBuffer(42)).isFalse();
+ assertThat(operator2.hasNonEmptyPQ(42)).isFalse();
+ assertThat(harness.numEventTimeTimers()).isZero();
- assertEquals(3, harness.getOutput().size());
+ assertThat(harness.getOutput()).hasSize(3);
verifyPattern(harness.getOutput().poll(), startEvent2, middleEvent2, endEvent2);
verifyWatermark(harness.getOutput().poll(), 20L);
@@ -671,7 +667,7 @@ public class CEPOperatorTest extends TestLogger {
}
@Test
- public void testCEPOperatorCleanupEventTimeWithSameElements() throws Exception {
+ void testCEPOperatorCleanupEventTimeWithSameElements() throws Exception {
Event startEvent = new Event(41, "c", 1.0);
Event middle1Event1 = new Event(41, "a", 2.0);
@@ -697,18 +693,19 @@ public class CEPOperatorTest extends TestLogger {
harness.processElement(new StreamRecord<>(middle1Event1, 3));
harness.processElement(new StreamRecord<>(new Event(41, "d", 6.0), 5));
- assertEquals(5L, harness.numEventTimeTimers());
- assertEquals(7L, operator.getPQSize(41));
- assertTrue(!operator.hasNonEmptySharedBuffer(41));
+ assertThat(harness.numEventTimeTimers()).isEqualTo(5L);
+ assertThat(operator.getPQSize(41)).isEqualTo(7L);
+ assertThat(operator.hasNonEmptySharedBuffer(41)).isFalse();
harness.processWatermark(new Watermark(2L));
verifyWatermark(harness.getOutput().poll(), Long.MIN_VALUE);
verifyWatermark(harness.getOutput().poll(), 2L);
- assertEquals(5L, harness.numEventTimeTimers());
- assertEquals(6L, operator.getPQSize(41));
- assertTrue(operator.hasNonEmptySharedBuffer(41)); // processed the first element
+ assertThat(harness.numEventTimeTimers()).isEqualTo(5L);
+ assertThat(operator.getPQSize(41)).isEqualTo(6L);
+ assertThat(operator.hasNonEmptySharedBuffer(41))
+ .isTrue(); // processed the first element
harness.processWatermark(new Watermark(8L));
@@ -758,23 +755,23 @@ public class CEPOperatorTest extends TestLogger {
middle2Event1,
middle1Event3)));
- assertEquals(1L, harness.numEventTimeTimers());
- assertEquals(0L, operator.getPQSize(41));
- assertTrue(operator.hasNonEmptySharedBuffer(41));
+ assertThat(harness.numEventTimeTimers()).isOne();
+ assertThat(operator.getPQSize(41)).isZero();
+ assertThat(operator.hasNonEmptySharedBuffer(41)).isTrue();
harness.processWatermark(new Watermark(17L));
verifyWatermark(harness.getOutput().poll(), 17L);
- assertTrue(!operator.hasNonEmptySharedBuffer(41));
- assertTrue(!operator.hasNonEmptyPQ(41));
- assertEquals(0L, harness.numEventTimeTimers());
+ assertThat(operator.hasNonEmptySharedBuffer(41)).isFalse();
+ assertThat(operator.hasNonEmptyPQ(41)).isFalse();
+ assertThat(harness.numEventTimeTimers()).isZero();
} finally {
harness.close();
}
}
@Test
- public void testCEPOperatorSideOutputLateElementsEventTime() throws Exception {
+ void testCEPOperatorSideOutputLateElementsEventTime() throws Exception {
Event startEvent = new Event(41, "c", 1.0);
Event middle1Event1 = new Event(41, "a", 2.0);
@@ -810,13 +807,12 @@ public class CEPOperatorTest extends TestLogger {
late.add(eventStreamRecord.getValue());
}
- List<Event> expected = Lists.newArrayList(middle1Event1, middle1Event2);
- Assert.assertArrayEquals(expected.toArray(), late.toArray());
+ assertThat(late).containsExactly(middle1Event1, middle1Event2);
}
}
@Test
- public void testCEPOperatorLateRecordsMetric() throws Exception {
+ void testCEPOperatorLateRecordsMetric() throws Exception {
Event startEvent = new Event(41, "c", 1.0);
Event middle1Event1 = new Event(41, "a", 2.0);
Event middle1Event2 = new Event(41, "a", 3.0);
@@ -835,14 +831,14 @@ public class CEPOperatorTest extends TestLogger {
harness.processWatermark(4L);
harness.processElement(middle1Event3, 3L);
- assertEquals(2L, operator.getLateRecordsNumber());
+ assertThat(operator.getLateRecordsNumber()).isEqualTo(2L);
} finally {
harness.close();
}
}
@Test
- public void testCEPOperatorCleanupProcessingTime() throws Exception {
+ void testCEPOperatorCleanupProcessingTime() throws Exception {
Event startEvent1 = new Event(42, "start", 1.0);
Event startEvent2 = new Event(42, "start", 2.0);
@@ -870,10 +866,10 @@ public class CEPOperatorTest extends TestLogger {
harness.processElement(
new StreamRecord<Event>(new SubEvent(42, "barfoo", 1.0, 5.0), 3L));
- assertTrue(!operator.hasNonEmptyPQ(42));
- assertTrue(!operator.hasNonEmptyPQ(43));
- assertTrue(operator.hasNonEmptySharedBuffer(42));
- assertTrue(operator.hasNonEmptySharedBuffer(43));
+ assertThat(operator.hasNonEmptyPQ(42)).isFalse();
+ assertThat(operator.hasNonEmptyPQ(43)).isFalse();
+ assertThat(operator.hasNonEmptySharedBuffer(42)).isTrue();
+ assertThat(operator.hasNonEmptySharedBuffer(43)).isTrue();
harness.setProcessingTime(3L);
@@ -907,10 +903,10 @@ public class CEPOperatorTest extends TestLogger {
harness.setProcessingTime(21L);
- assertTrue(!operator2.hasNonEmptySharedBuffer(42));
+ assertThat(operator2.hasNonEmptySharedBuffer(42)).isFalse();
harness.processElement(new StreamRecord<>(startEvent1, 21L));
- assertTrue(operator2.hasNonEmptySharedBuffer(42));
+ assertThat(operator2.hasNonEmptySharedBuffer(42)).isTrue();
harness.setProcessingTime(49L);
@@ -918,19 +914,19 @@ public class CEPOperatorTest extends TestLogger {
harness.processElement(new StreamRecord<>(new Event(42, "foobar", 1.0), 2L));
// the pattern expired
- assertTrue(!operator2.hasNonEmptySharedBuffer(42));
+ assertThat(operator2.hasNonEmptySharedBuffer(42)).isFalse();
- assertEquals(0L, harness.numEventTimeTimers());
- assertTrue(!operator2.hasNonEmptyPQ(42));
- assertTrue(!operator2.hasNonEmptyPQ(43));
+ assertThat(harness.numEventTimeTimers()).isZero();
+ assertThat(operator2.hasNonEmptyPQ(42)).isFalse();
+ assertThat(operator2.hasNonEmptyPQ(43)).isFalse();
} finally {
harness.close();
}
}
@Test
- public void testCEPOperatorSerializationWRocksDB() throws Exception {
- String rocksDbPath = tempFolder.newFolder().getAbsolutePath();
+ void testCEPOperatorSerializationWRocksDB() throws Exception {
+ String rocksDbPath = tempFolder.getAbsolutePath();
EmbeddedRocksDBStateBackend rocksDBStateBackend = new EmbeddedRocksDBStateBackend();
rocksDBStateBackend.setDbStoragePath(rocksDbPath);
@@ -1042,7 +1038,7 @@ public class CEPOperatorTest extends TestLogger {
}
@Test
- public void testCEPOperatorComparatorProcessTime() throws Exception {
+ void testCEPOperatorComparatorProcessTime() throws Exception {
Event startEvent1 = new Event(42, "start", 1.0);
Event startEvent2 = new Event(42, "start", 2.0);
SubEvent middleEvent1 = new SubEvent(42, "foo1", 3.0, 10.0);
@@ -1066,12 +1062,12 @@ public class CEPOperatorTest extends TestLogger {
harness.processElement(new StreamRecord<>(new Event(42, "foobar", 1.0), 0L));
harness.processElement(new StreamRecord<>(new SubEvent(42, "barfoo", 1.0, 5.0), 0L));
- assertTrue(!operator.hasNonEmptySharedBuffer(42));
- assertTrue(!operator.hasNonEmptySharedBuffer(43));
+ assertThat(operator.hasNonEmptySharedBuffer(42)).isFalse();
+ assertThat(operator.hasNonEmptySharedBuffer(43)).isFalse();
harness.setProcessingTime(3L);
- assertTrue(operator.hasNonEmptySharedBuffer(42));
- assertTrue(operator.hasNonEmptySharedBuffer(43));
+ assertThat(operator.hasNonEmptySharedBuffer(42)).isTrue();
+ assertThat(operator.hasNonEmptySharedBuffer(43)).isTrue();
harness.processElement(new StreamRecord<>(middleEvent2, 3L));
harness.processElement(new StreamRecord<>(middleEvent1, 3L));
@@ -1101,7 +1097,7 @@ public class CEPOperatorTest extends TestLogger {
}
@Test
- public void testCEPOperatorComparatorEventTime() throws Exception {
+ void testCEPOperatorComparatorEventTime() throws Exception {
Event startEvent1 = new Event(42, "start", 1.0);
Event startEvent2 = new Event(42, "start", 2.0);
SubEvent middleEvent1 = new SubEvent(42, "foo1", 1.0, 10.0);
@@ -1126,16 +1122,16 @@ public class CEPOperatorTest extends TestLogger {
harness.processElement(
new StreamRecord<Event>(new SubEvent(42, "barfoo", 1.0, 5.0), 3L));
- assertTrue(operator.hasNonEmptyPQ(42));
- assertTrue(operator.hasNonEmptyPQ(43));
- assertFalse(operator.hasNonEmptySharedBuffer(42));
- assertFalse(operator.hasNonEmptySharedBuffer(43));
+ assertThat(operator.hasNonEmptyPQ(42)).isTrue();
+ assertThat(operator.hasNonEmptyPQ(43)).isTrue();
+ assertThat(operator.hasNonEmptySharedBuffer(42)).isFalse();
+ assertThat(operator.hasNonEmptySharedBuffer(43)).isFalse();
harness.processWatermark(3L);
- assertFalse(operator.hasNonEmptyPQ(42));
- assertFalse(operator.hasNonEmptyPQ(43));
- assertTrue(operator.hasNonEmptySharedBuffer(42));
- assertTrue(operator.hasNonEmptySharedBuffer(43));
+ assertThat(operator.hasNonEmptyPQ(42)).isFalse();
+ assertThat(operator.hasNonEmptyPQ(43)).isFalse();
+ assertThat(operator.hasNonEmptySharedBuffer(42)).isTrue();
+ assertThat(operator.hasNonEmptySharedBuffer(43)).isTrue();
harness.processElement(new StreamRecord<>(startEvent2, 4L));
harness.processElement(new StreamRecord<Event>(middleEvent2, 5L));
@@ -1165,21 +1161,21 @@ public class CEPOperatorTest extends TestLogger {
}
private void verifyWatermark(Object outputObject, long timestamp) {
- assertTrue(outputObject instanceof Watermark);
- assertEquals(timestamp, ((Watermark) outputObject).getTimestamp());
+ assertThat(outputObject).isInstanceOf(Watermark.class);
+ assertThat(((Watermark) outputObject).getTimestamp()).isEqualTo(timestamp);
}
private void verifyPattern(Object outputObject, Event start, SubEvent middle, Event end) {
- assertTrue(outputObject instanceof StreamRecord);
+ assertThat(outputObject).isInstanceOf(StreamRecord.class);
StreamRecord<?> resultRecord = (StreamRecord<?>) outputObject;
- assertTrue(resultRecord.getValue() instanceof Map);
+ assertThat(resultRecord.getValue()).isInstanceOf(Map.class);
@SuppressWarnings("unchecked")
Map<String, List<Event>> patternMap = (Map<String, List<Event>>) resultRecord.getValue();
- assertEquals(start, patternMap.get("start").get(0));
- assertEquals(middle, patternMap.get("middle").get(0));
- assertEquals(end, patternMap.get("end").get(0));
+ assertThat(patternMap.get("start").get(0)).isEqualTo(start);
+ assertThat(patternMap.get("middle").get(0)).isEqualTo(middle);
+ assertThat(patternMap.get("end").get(0)).isEqualTo(end);
}
private CepOperator<Event, Integer, Map<String, List<Event>>> getKeyedCepOperator(
@@ -1202,39 +1198,17 @@ public class CEPOperatorTest extends TestLogger {
}
private void compareMaps(List<List<Event>> actual, List<List<Event>> expected) {
- Assert.assertEquals(expected.size(), actual.size());
-
+ // the order of events within a single match is not deterministic, so normalize each match
+ // before comparing the matches themselves without regard to their order
for (List<Event> p : actual) {
- Collections.sort(p, new EventComparator());
+ p.sort(new EventComparator());
}
for (List<Event> p : expected) {
- Collections.sort(p, new EventComparator());
+ p.sort(new EventComparator());
}
- Collections.sort(actual, new ListEventComparator());
- Collections.sort(expected, new ListEventComparator());
- Assert.assertArrayEquals(expected.toArray(), actual.toArray());
- }
-
- private class ListEventComparator implements Comparator<List<Event>> {
-
- @Override
- public int compare(List<Event> o1, List<Event> o2) {
- int sizeComp = Integer.compare(o1.size(), o2.size());
- if (sizeComp == 0) {
- EventComparator comp = new EventComparator();
- for (int i = 0; i < o1.size(); i++) {
- int eventComp = comp.compare(o1.get(i), o2.get(i));
- if (eventComp != 0) {
- return eventComp;
- }
- }
- return 0;
- } else {
- return sizeComp;
- }
- }
+ assertThat(actual).containsExactlyInAnyOrderElementsOf(expected);
}
private class EventComparator implements Comparator<Event> {
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPRescalingTest.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPRescalingTest.java
index 11d589fce04..9fda35ee227 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPRescalingTest.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPRescalingTest.java
@@ -37,21 +37,20 @@ import org.apache.flink.streaming.util.AbstractStreamOperatorTestHarness;
import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.Queue;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests for rescaling of CEP operators. */
-public class CEPRescalingTest {
+class CEPRescalingTest {
@Test
- public void testCEPFunctionScalingUp() throws Exception {
+ void testCEPFunctionScalingUp() throws Exception {
int maxParallelism = 10;
KeySelector<Event, Integer> keySelector =
@@ -74,11 +73,11 @@ public class CEPRescalingTest {
int keygroup =
KeyGroupRangeAssignment.assignToKeyGroup(
keySelector.getKey(startEvent1), maxParallelism);
- assertEquals(1, keygroup);
- assertEquals(
- 0,
- KeyGroupRangeAssignment.computeOperatorIndexForKeyGroup(
- maxParallelism, 2, keygroup));
+ assertThat(keygroup).isOne();
+ assertThat(
+ KeyGroupRangeAssignment.computeOperatorIndexForKeyGroup(
+ maxParallelism, 2, keygroup))
+ .isZero();
Event startEvent2 = new Event(10, "start", 1.0); // this will go to task index 2
SubEvent middleEvent2 = new SubEvent(10, "foo", 1.0, 10.0);
@@ -87,11 +86,11 @@ public class CEPRescalingTest {
keygroup =
KeyGroupRangeAssignment.assignToKeyGroup(
keySelector.getKey(startEvent2), maxParallelism);
- assertEquals(9, keygroup);
- assertEquals(
- 1,
- KeyGroupRangeAssignment.computeOperatorIndexForKeyGroup(
- maxParallelism, 2, keygroup));
+ assertThat(keygroup).isEqualTo(9);
+ assertThat(
+ KeyGroupRangeAssignment.computeOperatorIndexForKeyGroup(
+ maxParallelism, 2, keygroup))
+ .isOne();
// now we start the test, we go from parallelism 1 to 2.
OneInputStreamOperatorTestHarness<Event, Map<String, List<Event>>> harness = null;
@@ -142,7 +141,7 @@ public class CEPRescalingTest {
harness1.processWatermark(new Watermark(Long.MAX_VALUE));
// watermarks and the result
- assertEquals(3, harness1.getOutput().size());
+ assertThat(harness1.getOutput()).hasSize(3);
verifyWatermark(harness1.getOutput().poll(), 2);
verifyPattern(harness1.getOutput().poll(), startEvent1, middleEvent1, endEvent1);
@@ -160,7 +159,7 @@ public class CEPRescalingTest {
harness2.processWatermark(new Watermark(Long.MAX_VALUE));
- assertEquals(3, harness2.getOutput().size());
+ assertThat(harness2.getOutput()).hasSize(3);
verifyWatermark(harness2.getOutput().poll(), 2);
verifyPattern(harness2.getOutput().poll(), startEvent2, middleEvent2, endEvent2);
} finally {
@@ -180,7 +179,7 @@ public class CEPRescalingTest {
}
@Test
- public void testCEPFunctionScalingDown() throws Exception {
+ void testCEPFunctionScalingDown() throws Exception {
int maxParallelism = 10;
KeySelector<Event, Integer> keySelector =
@@ -203,15 +202,15 @@ public class CEPRescalingTest {
int keygroup =
KeyGroupRangeAssignment.assignToKeyGroup(
keySelector.getKey(startEvent1), maxParallelism);
- assertEquals(1, keygroup);
- assertEquals(
- 0,
- KeyGroupRangeAssignment.computeOperatorIndexForKeyGroup(
- maxParallelism, 3, keygroup));
- assertEquals(
- 0,
- KeyGroupRangeAssignment.computeOperatorIndexForKeyGroup(
- maxParallelism, 2, keygroup));
+ assertThat(keygroup).isOne();
+ assertThat(
+ KeyGroupRangeAssignment.computeOperatorIndexForKeyGroup(
+ maxParallelism, 3, keygroup))
+ .isZero();
+ assertThat(
+ KeyGroupRangeAssignment.computeOperatorIndexForKeyGroup(
+ maxParallelism, 2, keygroup))
+ .isZero();
Event startEvent2 = new Event(45, "start", 1.0); // this will go to task index 1
SubEvent middleEvent2 = new SubEvent(45, "foo", 1.0, 10.0);
@@ -220,15 +219,15 @@ public class CEPRescalingTest {
keygroup =
KeyGroupRangeAssignment.assignToKeyGroup(
keySelector.getKey(startEvent2), maxParallelism);
- assertEquals(6, keygroup);
- assertEquals(
- 1,
- KeyGroupRangeAssignment.computeOperatorIndexForKeyGroup(
- maxParallelism, 3, keygroup));
- assertEquals(
- 1,
- KeyGroupRangeAssignment.computeOperatorIndexForKeyGroup(
- maxParallelism, 2, keygroup));
+ assertThat(keygroup).isEqualTo(6);
+ assertThat(
+ KeyGroupRangeAssignment.computeOperatorIndexForKeyGroup(
+ maxParallelism, 3, keygroup))
+ .isOne();
+ assertThat(
+ KeyGroupRangeAssignment.computeOperatorIndexForKeyGroup(
+ maxParallelism, 2, keygroup))
+ .isOne();
Event startEvent3 = new Event(90, "start", 1.0); // this will go to task index 0
SubEvent middleEvent3 = new SubEvent(90, "foo", 1.0, 10.0);
@@ -237,15 +236,15 @@ public class CEPRescalingTest {
keygroup =
KeyGroupRangeAssignment.assignToKeyGroup(
keySelector.getKey(startEvent3), maxParallelism);
- assertEquals(2, keygroup);
- assertEquals(
- 0,
- KeyGroupRangeAssignment.computeOperatorIndexForKeyGroup(
- maxParallelism, 3, keygroup));
- assertEquals(
- 0,
- KeyGroupRangeAssignment.computeOperatorIndexForKeyGroup(
- maxParallelism, 2, keygroup));
+ assertThat(keygroup).isEqualTo(2);
+ assertThat(
+ KeyGroupRangeAssignment.computeOperatorIndexForKeyGroup(
+ maxParallelism, 3, keygroup))
+ .isZero();
+ assertThat(
+ KeyGroupRangeAssignment.computeOperatorIndexForKeyGroup(
+ maxParallelism, 2, keygroup))
+ .isZero();
Event startEvent4 = new Event(10, "start", 1.0); // this will go to task index 2
SubEvent middleEvent4 = new SubEvent(10, "foo", 1.0, 10.0);
@@ -254,15 +253,15 @@ public class CEPRescalingTest {
keygroup =
KeyGroupRangeAssignment.assignToKeyGroup(
keySelector.getKey(startEvent4), maxParallelism);
- assertEquals(9, keygroup);
- assertEquals(
- 2,
- KeyGroupRangeAssignment.computeOperatorIndexForKeyGroup(
- maxParallelism, 3, keygroup));
- assertEquals(
- 1,
- KeyGroupRangeAssignment.computeOperatorIndexForKeyGroup(
- maxParallelism, 2, keygroup));
+ assertThat(keygroup).isEqualTo(9);
+ assertThat(
+ KeyGroupRangeAssignment.computeOperatorIndexForKeyGroup(
+ maxParallelism, 3, keygroup))
+ .isEqualTo(2);
+ assertThat(
+ KeyGroupRangeAssignment.computeOperatorIndexForKeyGroup(
+ maxParallelism, 2, keygroup))
+ .isOne();
// starting the test, we will go from parallelism of 3 to parallelism of 2
@@ -306,13 +305,13 @@ public class CEPRescalingTest {
harness3.processElement(new StreamRecord<>(endEvent4, 17));
// so far we only have the initial watermark
- assertEquals(1, harness1.getOutput().size());
+ assertThat(harness1.getOutput()).hasSize(1);
verifyWatermark(harness1.getOutput().poll(), Long.MIN_VALUE);
- assertEquals(1, harness2.getOutput().size());
+ assertThat(harness2.getOutput()).hasSize(1);
verifyWatermark(harness2.getOutput().poll(), Long.MIN_VALUE);
- assertEquals(1, harness3.getOutput().size());
+ assertThat(harness3.getOutput()).hasSize(1);
verifyWatermark(harness3.getOutput().poll(), Long.MIN_VALUE);
// we take a snapshot and make it look as a single operator
@@ -351,7 +350,7 @@ public class CEPRescalingTest {
// a pruning time underflow exception in NFA
harness4.processWatermark(new Watermark(12));
- assertEquals(2, harness4.getOutput().size());
+ assertThat(harness4.getOutput()).hasSize(2);
verifyPattern(harness4.getOutput().poll(), startEvent1, middleEvent1, endEvent1);
verifyWatermark(harness4.getOutput().poll(), 12);
@@ -365,12 +364,12 @@ public class CEPRescalingTest {
harness5.processWatermark(new Watermark(Long.MAX_VALUE));
// verify result
- assertEquals(3, harness4.getOutput().size());
+ assertThat(harness4.getOutput()).hasSize(3);
// check the order of the events in the output
Queue<Object> output = harness4.getOutput();
StreamRecord<?> resultRecord = (StreamRecord<?>) output.peek();
- assertTrue(resultRecord.getValue() instanceof Map);
+ assertThat(resultRecord.getValue()).isInstanceOf(Map.class);
@SuppressWarnings("unchecked")
Map<String, List<Event>> patternMap =
@@ -384,7 +383,7 @@ public class CEPRescalingTest {
}
// after scaling down this should end up here
- assertEquals(2, harness5.getOutput().size());
+ assertThat(harness5.getOutput()).hasSize(2);
verifyPattern(harness5.getOutput().poll(), startEvent4, middleEvent4, endEvent4);
} finally {
closeSilently(harness1);
@@ -396,21 +395,21 @@ public class CEPRescalingTest {
}
private void verifyWatermark(Object outputObject, long timestamp) {
- assertTrue(outputObject instanceof Watermark);
- assertEquals(timestamp, ((Watermark) outputObject).getTimestamp());
+ assertThat(outputObject).isInstanceOf(Watermark.class);
+ assertThat(((Watermark) outputObject).getTimestamp()).isEqualTo(timestamp);
}
private void verifyPattern(Object outputObject, Event start, SubEvent middle, Event end) {
- assertTrue(outputObject instanceof StreamRecord);
+ assertThat(outputObject).isInstanceOf(StreamRecord.class);
StreamRecord<?> resultRecord = (StreamRecord<?>) outputObject;
- assertTrue(resultRecord.getValue() instanceof Map);
+ assertThat(resultRecord.getValue()).isInstanceOf(Map.class);
@SuppressWarnings("unchecked")
Map<String, List<Event>> patternMap = (Map<String, List<Event>>) resultRecord.getValue();
- assertEquals(start, patternMap.get("start").get(0));
- assertEquals(middle, patternMap.get("middle").get(0));
- assertEquals(end, patternMap.get("end").get(0));
+ assertThat(patternMap.get("start").get(0)).isEqualTo(start);
+ assertThat(patternMap.get("middle").get(0)).isEqualTo(middle);
+ assertThat(patternMap.get("end").get(0)).isEqualTo(end);
}
private KeyedOneInputStreamOperatorTestHarness<Integer, Event, Map<String, List<Event>>>
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CepProcessFunctionContextTest.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CepProcessFunctionContextTest.java
index cce6541744b..21d7935d70a 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CepProcessFunctionContextTest.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CepProcessFunctionContextTest.java
@@ -27,9 +27,8 @@ import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
-import org.apache.flink.util.TestLogger;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.time.Duration;
import java.util.List;
@@ -43,13 +42,13 @@ import static org.apache.flink.cep.utils.OutputAsserter.assertOutput;
/**
* Tests for {@link CepOperator} which check proper setting {@link PatternProcessFunction.Context}.
*/
-public class CepProcessFunctionContextTest extends TestLogger {
+class CepProcessFunctionContextTest {
private static final boolean PROCESSING_TIME = true;
private static final boolean EVENT_TIME = false;
@Test
- public void testTimestampPassingInEventTime() throws Exception {
+ void testTimestampPassingInEventTime() throws Exception {
try (OneInputStreamOperatorTestHarness<Event, String> harness =
getCepTestHarness(
@@ -74,7 +73,7 @@ public class CepProcessFunctionContextTest extends TestLogger {
}
@Test
- public void testTimestampPassingInProcessingTime() throws Exception {
+ void testTimestampPassingInProcessingTime() throws Exception {
try (OneInputStreamOperatorTestHarness<Event, String> harness =
getCepTestHarness(
@@ -98,7 +97,7 @@ public class CepProcessFunctionContextTest extends TestLogger {
}
@Test
- public void testCurrentProcessingTimeInProcessingTime() throws Exception {
+ void testCurrentProcessingTimeInProcessingTime() throws Exception {
try (OneInputStreamOperatorTestHarness<Event, String> harness =
getCepTestHarness(
@@ -121,7 +120,7 @@ public class CepProcessFunctionContextTest extends TestLogger {
}
@Test
- public void testCurrentProcessingTimeInEventTime() throws Exception {
+ void testCurrentProcessingTimeInEventTime() throws Exception {
try (OneInputStreamOperatorTestHarness<Event, String> harness =
getCepTestHarness(
@@ -144,7 +143,7 @@ public class CepProcessFunctionContextTest extends TestLogger {
}
@Test
- public void testTimestampPassingForTimedOutInEventTime() throws Exception {
+ void testTimestampPassingForTimedOutInEventTime() throws Exception {
OutputTag<String> timedOut = new OutputTag<String>("timedOut") {};
@@ -175,7 +174,7 @@ public class CepProcessFunctionContextTest extends TestLogger {
}
@Test
- public void testTimestampPassingForTimedOutInProcessingTime() throws Exception {
+ void testTimestampPassingForTimedOutInProcessingTime() throws Exception {
OutputTag<String> timedOut = new OutputTag<String>("timedOut") {};
@@ -203,7 +202,7 @@ public class CepProcessFunctionContextTest extends TestLogger {
}
@Test
- public void testCurrentProcessingTimeForTimedOutInEventTime() throws Exception {
+ void testCurrentProcessingTimeForTimedOutInEventTime() throws Exception {
OutputTag<String> sideOutputTag = new OutputTag<String>("timedOut") {};
@@ -235,7 +234,7 @@ public class CepProcessFunctionContextTest extends TestLogger {
}
@Test
- public void testCurrentProcessingTimeForTimedOutInProcessingTime() throws Exception {
+ void testCurrentProcessingTimeForTimedOutInProcessingTime() throws Exception {
OutputTag<String> sideOutputTag = new OutputTag<String>("timedOut") {};
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CepRuntimeContextTest.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CepRuntimeContextTest.java
index 85312f857b2..c70f1d4bd97 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CepRuntimeContextTest.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CepRuntimeContextTest.java
@@ -41,9 +41,8 @@ import org.apache.flink.metrics.groups.OperatorMetricGroup;
import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.flink.util.Collector;
-import org.apache.flink.util.TestLogger;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.util.HashMap;
import java.util.List;
@@ -53,20 +52,19 @@ import static org.apache.flink.cep.operator.CepRuntimeContextTest.MockProcessFun
import static org.apache.flink.cep.utils.CepOperatorBuilder.createOperatorForNFA;
import static org.apache.flink.cep.utils.CepOperatorTestUtilities.getCepTestHarness;
import static org.apache.flink.cep.utils.EventBuilder.event;
-import static org.hamcrest.core.Is.is;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.fail;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assertions.fail;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/** Test cases for {@link CepRuntimeContext}. */
-public class CepRuntimeContextTest extends TestLogger {
+class CepRuntimeContextTest {
@Test
- public void testCepRuntimeContextIsSetInNFA() throws Exception {
+ void testCepRuntimeContextIsSetInNFA() throws Exception {
@SuppressWarnings("unchecked")
final NFA<Event> mockNFA = mock(NFA.class);
@@ -80,7 +78,7 @@ public class CepRuntimeContextTest extends TestLogger {
}
@Test
- public void testCepRuntimeContextIsSetInProcessFunction() throws Exception {
+ void testCepRuntimeContextIsSetInProcessFunction() throws Exception {
final VerifyRuntimeContextProcessFunction processFunction =
new VerifyRuntimeContextProcessFunction();
@@ -107,7 +105,7 @@ public class CepRuntimeContextTest extends TestLogger {
}
@Test
- public void testCepRuntimeContext() {
+ void testCepRuntimeContext() {
final String taskName = "foobarTask";
final OperatorMetricGroup metricGroup =
UnregisteredMetricsGroup.createOperatorMetricGroup();
@@ -138,168 +136,127 @@ public class CepRuntimeContextTest extends TestLogger {
RuntimeContext runtimeContext = new CepRuntimeContext(mockedRuntimeContext);
- assertEquals(taskName, runtimeContext.getTaskInfo().getTaskName());
- assertEquals(metricGroup, runtimeContext.getMetricGroup());
- assertEquals(
- numberOfParallelSubtasks,
- runtimeContext.getTaskInfo().getNumberOfParallelSubtasks());
- assertEquals(indexOfSubtask, runtimeContext.getTaskInfo().getIndexOfThisSubtask());
- assertEquals(attemptNumber, runtimeContext.getTaskInfo().getAttemptNumber());
- assertEquals(taskNameWithSubtask, runtimeContext.getTaskInfo().getTaskNameWithSubtasks());
- assertEquals(globalJobParameters, runtimeContext.getGlobalJobParameters());
- assertEquals(isObjectReused, runtimeContext.isObjectReuseEnabled());
- assertEquals(userCodeClassLoader, runtimeContext.getUserCodeClassLoader());
- assertEquals(distributedCache, runtimeContext.getDistributedCache());
-
- try {
- runtimeContext.getState(new ValueStateDescriptor<>("foobar", Integer.class, 42));
- fail("Expected getState to fail with unsupported operation exception.");
- } catch (UnsupportedOperationException e) {
- // expected
- }
-
- try {
- runtimeContext.getListState(new ListStateDescriptor<>("foobar", Integer.class));
- fail("Expected getListState to fail with unsupported operation exception.");
- } catch (UnsupportedOperationException e) {
- // expected
- }
-
- try {
- runtimeContext.getReducingState(
- new ReducingStateDescriptor<>(
- "foobar", mock(ReduceFunction.class), Integer.class));
- fail("Expected getReducingState to fail with unsupported operation exception.");
- } catch (UnsupportedOperationException e) {
- // expected
- }
-
- try {
- runtimeContext.getAggregatingState(
- new AggregatingStateDescriptor<>(
- "foobar", mock(AggregateFunction.class), Integer.class));
- fail("Expected getAggregatingState to fail with unsupported operation exception.");
- } catch (UnsupportedOperationException e) {
- // expected
- }
-
- try {
- runtimeContext.getMapState(
- new MapStateDescriptor<>("foobar", Integer.class, String.class));
- fail("Expected getMapState to fail with unsupported operation exception.");
- } catch (UnsupportedOperationException e) {
- // expected
- }
-
- try {
- runtimeContext.getState(
- new org.apache.flink.api.common.state.v2.ValueStateDescriptor<>(
- "foobar", Integer.class));
- fail("Expected getState to fail with unsupported operation exception.");
- } catch (UnsupportedOperationException e) {
- // expected
- }
-
- try {
- runtimeContext.getListState(
- new org.apache.flink.api.common.state.v2.ListStateDescriptor<>(
- "foobar", Integer.class));
- fail("Expected getListState to fail with unsupported operation exception.");
- } catch (UnsupportedOperationException e) {
- // expected
- }
-
- try {
- runtimeContext.getReducingState(
- new org.apache.flink.api.common.state.v2.ReducingStateDescriptor<>(
- "foobar", mock(ReduceFunction.class), Integer.class));
- fail("Expected getReducingState to fail with unsupported operation exception.");
- } catch (UnsupportedOperationException e) {
- // expected
- }
-
- try {
- runtimeContext.getAggregatingState(
- new org.apache.flink.api.common.state.v2.AggregatingStateDescriptor<>(
- "foobar", mock(AggregateFunction.class), Integer.class));
- fail("Expected getAggregatingState to fail with unsupported operation exception.");
- } catch (UnsupportedOperationException e) {
- // expected
- }
-
- try {
- runtimeContext.getMapState(
- new org.apache.flink.api.common.state.v2.MapStateDescriptor<>(
- "foobar", Integer.class, String.class));
- fail("Expected getMapState to fail with unsupported operation exception.");
- } catch (UnsupportedOperationException e) {
- // expected
- }
-
- try {
- runtimeContext.addAccumulator("foobar", mock(Accumulator.class));
- fail("Expected addAccumulator to fail with unsupported operation exception.");
- } catch (UnsupportedOperationException e) {
- // expected
- }
-
- try {
- runtimeContext.getAccumulator("foobar");
- fail("Expected getAccumulator to fail with unsupported operation exception.");
- } catch (UnsupportedOperationException e) {
- // expected
- }
-
- try {
- runtimeContext.getIntCounter("foobar");
- fail("Expected getIntCounter to fail with unsupported operation exception.");
- } catch (UnsupportedOperationException e) {
- // expected
- }
-
- try {
- runtimeContext.getLongCounter("foobar");
- fail("Expected getLongCounter to fail with unsupported operation exception.");
- } catch (UnsupportedOperationException e) {
- // expected
- }
-
- try {
- runtimeContext.getDoubleCounter("foobar");
- fail("Expected getDoubleCounter to fail with unsupported operation exception.");
- } catch (UnsupportedOperationException e) {
- // expected
- }
-
- try {
- runtimeContext.getHistogram("foobar");
- fail("Expected getHistogram to fail with unsupported operation exception.");
- } catch (UnsupportedOperationException e) {
- // expected
- }
-
- try {
- runtimeContext.hasBroadcastVariable("foobar");
- fail("Expected hasBroadcastVariable to fail with unsupported operation exception.");
- } catch (UnsupportedOperationException e) {
- // expected
- }
-
- try {
- runtimeContext.getBroadcastVariable("foobar");
- fail("Expected getBroadcastVariable to fail with unsupported operation exception.");
- } catch (UnsupportedOperationException e) {
- // expected
- }
-
- try {
- runtimeContext.getBroadcastVariableWithInitializer(
- "foobar", mock(BroadcastVariableInitializer.class));
- fail(
- "Expected getBroadcastVariableWithInitializer to fail with unsupported operation exception.");
- } catch (UnsupportedOperationException e) {
- // expected
- }
+ assertThat(runtimeContext.getTaskInfo().getTaskName()).isEqualTo(taskName);
+ assertThat(runtimeContext.getMetricGroup()).isEqualTo(metricGroup);
+ assertThat(runtimeContext.getTaskInfo().getNumberOfParallelSubtasks())
+ .isEqualTo(numberOfParallelSubtasks);
+ assertThat(runtimeContext.getTaskInfo().getIndexOfThisSubtask()).isEqualTo(indexOfSubtask);
+ assertThat(runtimeContext.getTaskInfo().getAttemptNumber()).isEqualTo(attemptNumber);
+ assertThat(runtimeContext.getTaskInfo().getTaskNameWithSubtasks())
+ .isEqualTo(taskNameWithSubtask);
+ assertThat(runtimeContext.getGlobalJobParameters()).isEqualTo(globalJobParameters);
+ assertThat(runtimeContext.isObjectReuseEnabled()).isEqualTo(isObjectReused);
+ assertThat(runtimeContext.getUserCodeClassLoader()).isEqualTo(userCodeClassLoader);
+ assertThat(runtimeContext.getDistributedCache()).isEqualTo(distributedCache);
+
+ assertThatThrownBy(
+ () ->
+ runtimeContext.getState(
+ new ValueStateDescriptor<>("foobar", Integer.class, 42)))
+ .isInstanceOf(UnsupportedOperationException.class);
+
+ assertThatThrownBy(
+ () ->
+ runtimeContext.getListState(
+ new ListStateDescriptor<>("foobar", Integer.class)))
+ .isInstanceOf(UnsupportedOperationException.class);
+
+ assertThatThrownBy(
+ () ->
+ runtimeContext.getReducingState(
+ new ReducingStateDescriptor<>(
+ "foobar",
+ mock(ReduceFunction.class),
+ Integer.class)))
+ .isInstanceOf(UnsupportedOperationException.class);
+
+ assertThatThrownBy(
+ () ->
+ runtimeContext.getAggregatingState(
+ new AggregatingStateDescriptor<>(
+ "foobar",
+ mock(AggregateFunction.class),
+ Integer.class)))
+ .isInstanceOf(UnsupportedOperationException.class);
+
+ assertThatThrownBy(
+ () ->
+ runtimeContext.getMapState(
+ new MapStateDescriptor<>(
+ "foobar", Integer.class, String.class)))
+ .isInstanceOf(UnsupportedOperationException.class);
+
+ assertThatThrownBy(
+ () ->
+ runtimeContext.getState(
+ new org.apache.flink.api.common.state.v2
+ .ValueStateDescriptor<>("foobar", Integer.class)))
+ .isInstanceOf(UnsupportedOperationException.class);
+
+ assertThatThrownBy(
+ () ->
+ runtimeContext.getListState(
+ new org.apache.flink.api.common.state.v2
+ .ListStateDescriptor<>("foobar", Integer.class)))
+ .isInstanceOf(UnsupportedOperationException.class);
+
+ assertThatThrownBy(
+ () ->
+ runtimeContext.getReducingState(
+ new org.apache.flink.api.common.state.v2
+ .ReducingStateDescriptor<>(
+ "foobar",
+ mock(ReduceFunction.class),
+ Integer.class)))
+ .isInstanceOf(UnsupportedOperationException.class);
+
+ assertThatThrownBy(
+ () ->
+ runtimeContext.getAggregatingState(
+ new org.apache.flink.api.common.state.v2
+ .AggregatingStateDescriptor<>(
+ "foobar",
+ mock(AggregateFunction.class),
+ Integer.class)))
+ .isInstanceOf(UnsupportedOperationException.class);
+
+ assertThatThrownBy(
+ () ->
+ runtimeContext.getMapState(
+ new org.apache.flink.api.common.state.v2
+ .MapStateDescriptor<>(
+ "foobar", Integer.class, String.class)))
+ .isInstanceOf(UnsupportedOperationException.class);
+
+ assertThatThrownBy(() -> runtimeContext.addAccumulator("foobar", mock(Accumulator.class)))
+ .isInstanceOf(UnsupportedOperationException.class);
+
+ assertThatThrownBy(() -> runtimeContext.getAccumulator("foobar"))
+ .isInstanceOf(UnsupportedOperationException.class);
+
+ assertThatThrownBy(() -> runtimeContext.getIntCounter("foobar"))
+ .isInstanceOf(UnsupportedOperationException.class);
+
+ assertThatThrownBy(() -> runtimeContext.getLongCounter("foobar"))
+ .isInstanceOf(UnsupportedOperationException.class);
+
+ assertThatThrownBy(() -> runtimeContext.getDoubleCounter("foobar"))
+ .isInstanceOf(UnsupportedOperationException.class);
+
+ assertThatThrownBy(() -> runtimeContext.getHistogram("foobar"))
+ .isInstanceOf(UnsupportedOperationException.class);
+
+ assertThatThrownBy(() -> runtimeContext.hasBroadcastVariable("foobar"))
+ .isInstanceOf(UnsupportedOperationException.class);
+
+ assertThatThrownBy(() -> runtimeContext.getBroadcastVariable("foobar"))
+ .isInstanceOf(UnsupportedOperationException.class);
+
+ assertThatThrownBy(
+ () ->
+ runtimeContext.getBroadcastVariableWithInitializer(
+ "foobar", mock(BroadcastVariableInitializer.class)))
+ .isInstanceOf(UnsupportedOperationException.class);
}
/* Test Utils */
@@ -316,17 +273,17 @@ public class CepRuntimeContextTest extends TestLogger {
}
MockProcessFunctionAsserter checkOpenCalled() {
- assertThat(function.openCalled, is(true));
+ assertThat(function.openCalled).isTrue();
return this;
}
MockProcessFunctionAsserter checkCloseCalled() {
- assertThat(function.openCalled, is(true));
+ assertThat(function.openCalled).isTrue();
return this;
}
MockProcessFunctionAsserter checkProcessMatchCalled() {
- assertThat(function.processMatchCalled, is(true));
+ assertThat(function.processMatchCalled).isTrue();
return this;
}
}
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/pattern/PatternTest.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/pattern/PatternTest.java
index 54b021c242b..e8fdd3a3bab 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/pattern/PatternTest.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/pattern/PatternTest.java
@@ -26,58 +26,54 @@ import org.apache.flink.cep.pattern.conditions.RichAndCondition;
import org.apache.flink.cep.pattern.conditions.RichOrCondition;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.cep.pattern.conditions.SubtypeCondition;
-import org.apache.flink.util.TestLogger;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.Mockito.mock;
/** Tests for constructing {@link Pattern}. */
-public class PatternTest extends TestLogger {
+class PatternTest {
/** These test simply test that the pattern construction completes without failure. */
@Test
- public void testStrictContiguity() {
+ void testStrictContiguity() {
Pattern<Object, ?> pattern = Pattern.begin("start").next("next").next("end");
Pattern<Object, ?> previous;
Pattern<Object, ?> previous2;
- assertNotNull(previous = pattern.getPrevious());
- assertNotNull(previous2 = previous.getPrevious());
- assertNull(previous2.getPrevious());
+ assertThat(previous = pattern.getPrevious()).isNotNull();
+ assertThat(previous2 = previous.getPrevious()).isNotNull();
+ assertThat(previous2.getPrevious()).isNull();
- assertEquals(pattern.getName(), "end");
- assertEquals(previous.getName(), "next");
- assertEquals(previous2.getName(), "start");
+ assertThat(pattern.getName()).isEqualTo("end");
+ assertThat(previous.getName()).isEqualTo("next");
+ assertThat(previous2.getName()).isEqualTo("start");
}
@Test
- public void testNonStrictContiguity() {
+ void testNonStrictContiguity() {
Pattern<Object, ?> pattern = Pattern.begin("start").followedBy("next").followedBy("end");
Pattern<Object, ?> previous;
Pattern<Object, ?> previous2;
- assertNotNull(previous = pattern.getPrevious());
- assertNotNull(previous2 = previous.getPrevious());
- assertNull(previous2.getPrevious());
+ assertThat(previous = pattern.getPrevious()).isNotNull();
+ assertThat(previous2 = previous.getPrevious()).isNotNull();
+ assertThat(previous2.getPrevious()).isNull();
- assertEquals(
- ConsumingStrategy.SKIP_TILL_NEXT, pattern.getQuantifier().getConsumingStrategy());
- assertEquals(
- ConsumingStrategy.SKIP_TILL_NEXT, previous.getQuantifier().getConsumingStrategy());
+ assertThat(pattern.getQuantifier().getConsumingStrategy())
+ .isEqualTo(ConsumingStrategy.SKIP_TILL_NEXT);
+ assertThat(previous.getQuantifier().getConsumingStrategy())
+ .isEqualTo(ConsumingStrategy.SKIP_TILL_NEXT);
- assertEquals(pattern.getName(), "end");
- assertEquals(previous.getName(), "next");
- assertEquals(previous2.getName(), "start");
+ assertThat(pattern.getName()).isEqualTo("end");
+ assertThat(previous.getName()).isEqualTo("next");
+ assertThat(previous2.getName()).isEqualTo("start");
}
@Test
- public void testStrictContiguityWithCondition() {
+ void testStrictContiguityWithCondition() {
Pattern<Event, ?> pattern =
Pattern.<Event>begin("start")
.next("next")
@@ -88,21 +84,21 @@ public class PatternTest extends TestLogger {
Pattern<Event, ?> previous;
Pattern<Event, ?> previous2;
- assertNotNull(previous = pattern.getPrevious());
- assertNotNull(previous2 = previous.getPrevious());
- assertNull(previous2.getPrevious());
+ assertThat(previous = pattern.getPrevious()).isNotNull();
+ assertThat(previous2 = previous.getPrevious()).isNotNull();
+ assertThat(previous2.getPrevious()).isNull();
- assertNotNull(pattern.getCondition());
- assertNotNull(previous.getCondition());
- assertNotNull(previous2.getCondition());
+ assertThat(pattern.getCondition()).isNotNull();
+ assertThat(previous.getCondition()).isNotNull();
+ assertThat(previous2.getCondition()).isNotNull();
- assertEquals(pattern.getName(), "end");
- assertEquals(previous.getName(), "next");
- assertEquals(previous2.getName(), "start");
+ assertThat(pattern.getName()).isEqualTo("end");
+ assertThat(previous.getName()).isEqualTo("next");
+ assertThat(previous2.getName()).isEqualTo("start");
}
@Test
- public void testPatternWithSubtyping() {
+ void testPatternWithSubtyping() {
Pattern<Event, ?> pattern =
Pattern.<Event>begin("start")
.next("subevent")
@@ -112,20 +108,20 @@ public class PatternTest extends TestLogger {
Pattern<Event, ?> previous;
Pattern<Event, ?> previous2;
- assertNotNull(previous = pattern.getPrevious());
- assertNotNull(previous2 = previous.getPrevious());
- assertNull(previous2.getPrevious());
+ assertThat(previous = pattern.getPrevious()).isNotNull();
+ assertThat(previous2 = previous.getPrevious()).isNotNull();
+ assertThat(previous2.getPrevious()).isNull();
- assertNotNull(previous.getCondition());
- assertTrue(previous.getCondition() instanceof SubtypeCondition);
+ assertThat(previous.getCondition()).isNotNull();
+ assertThat(previous.getCondition()).isInstanceOf(SubtypeCondition.class);
- assertEquals(pattern.getName(), "end");
- assertEquals(previous.getName(), "subevent");
- assertEquals(previous2.getName(), "start");
+ assertThat(pattern.getName()).isEqualTo("end");
+ assertThat(previous.getName()).isEqualTo("subevent");
+ assertThat(previous2.getName()).isEqualTo("start");
}
@Test
- public void testPatternWithSubtypingAndFilter() {
+ void testPatternWithSubtypingAndFilter() {
Pattern<Event, Event> pattern =
Pattern.<Event>begin("start")
.next("subevent")
@@ -136,21 +132,21 @@ public class PatternTest extends TestLogger {
Pattern<Event, ?> previous;
Pattern<Event, ?> previous2;
- assertNotNull(previous = pattern.getPrevious());
- assertNotNull(previous2 = previous.getPrevious());
- assertNull(previous2.getPrevious());
+ assertThat(previous = pattern.getPrevious()).isNotNull();
+ assertThat(previous2 = previous.getPrevious()).isNotNull();
+ assertThat(previous2.getPrevious()).isNull();
- assertEquals(
- ConsumingStrategy.SKIP_TILL_NEXT, pattern.getQuantifier().getConsumingStrategy());
- assertNotNull(previous.getCondition());
+ assertThat(pattern.getQuantifier().getConsumingStrategy())
+ .isEqualTo(ConsumingStrategy.SKIP_TILL_NEXT);
+ assertThat(previous.getCondition()).isNotNull();
- assertEquals(pattern.getName(), "end");
- assertEquals(previous.getName(), "subevent");
- assertEquals(previous2.getName(), "start");
+ assertThat(pattern.getName()).isEqualTo("end");
+ assertThat(previous.getName()).isEqualTo("subevent");
+ assertThat(previous2.getName()).isEqualTo("start");
}
@Test
- public void testPatternWithOrFilter() {
+ void testPatternWithOrFilter() {
Pattern<Event, Event> pattern =
Pattern.<Event>begin("start")
.where(SimpleCondition.of(value -> false))
@@ -162,22 +158,22 @@ public class PatternTest extends TestLogger {
Pattern<Event, ?> previous;
Pattern<Event, ?> previous2;
- assertNotNull(previous = pattern.getPrevious());
- assertNotNull(previous2 = previous.getPrevious());
- assertNull(previous2.getPrevious());
+ assertThat(previous = pattern.getPrevious()).isNotNull();
+ assertThat(previous2 = previous.getPrevious()).isNotNull();
+ assertThat(previous2.getPrevious()).isNull();
- assertEquals(
- ConsumingStrategy.SKIP_TILL_NEXT, pattern.getQuantifier().getConsumingStrategy());
- assertFalse(previous.getCondition() instanceof RichOrCondition);
- assertTrue(previous2.getCondition() instanceof RichOrCondition);
+ assertThat(pattern.getQuantifier().getConsumingStrategy())
+ .isEqualTo(ConsumingStrategy.SKIP_TILL_NEXT);
+ assertThat(previous.getCondition()).isNotInstanceOf(RichOrCondition.class);
+ assertThat(previous2.getCondition()).isInstanceOf(RichOrCondition.class);
- assertEquals(pattern.getName(), "end");
- assertEquals(previous.getName(), "or");
- assertEquals(previous2.getName(), "start");
+ assertThat(pattern.getName()).isEqualTo("end");
+ assertThat(previous.getName()).isEqualTo("or");
+ assertThat(previous2.getName()).isEqualTo("start");
}
@Test
- public void testRichCondition() {
+ void testRichCondition() {
Pattern<Event, Event> pattern =
Pattern.<Event>begin("start")
.where(mock(IterativeCondition.class))
@@ -185,125 +181,178 @@ public class PatternTest extends TestLogger {
.followedBy("end")
.where(mock(IterativeCondition.class))
.or(mock(IterativeCondition.class));
- assertTrue(pattern.getCondition() instanceof RichOrCondition);
- assertTrue(pattern.getPrevious().getCondition() instanceof RichAndCondition);
+ assertThat(pattern.getCondition()).isInstanceOf(RichOrCondition.class);
+ assertThat(pattern.getPrevious().getCondition()).isInstanceOf(RichAndCondition.class);
}
- @Test(expected = IllegalArgumentException.class)
- public void testPatternTimesNegativeTimes() throws Exception {
- Pattern.begin("start").where(dummyCondition()).times(-1);
+ @Test
+ void testPatternTimesNegativeTimes() {
+ assertThatThrownBy(() -> Pattern.begin("start").where(dummyCondition()).times(-1))
+ .isInstanceOf(IllegalArgumentException.class);
}
- @Test(expected = IllegalArgumentException.class)
- public void testPatternTimesNegativeFrom() throws Exception {
- Pattern.begin("start").where(dummyCondition()).times(-1, 2);
+ @Test
+ void testPatternTimesNegativeFrom() {
+ assertThatThrownBy(() -> Pattern.begin("start").where(dummyCondition()).times(-1, 2))
+ .isInstanceOf(IllegalArgumentException.class);
}
- @Test(expected = MalformedPatternException.class)
- public void testPatternCanHaveQuantifierSpecifiedOnce1() throws Exception {
-
- Pattern.begin("start").where(dummyCondition()).oneOrMore().oneOrMore().optional();
+ @Test
+ void testPatternCanHaveQuantifierSpecifiedOnce1() {
+ assertThatThrownBy(
+ () ->
+ Pattern.begin("start")
+ .where(dummyCondition())
+ .oneOrMore()
+ .oneOrMore()
+ .optional())
+ .isInstanceOf(MalformedPatternException.class);
}
- @Test(expected = MalformedPatternException.class)
- public void testPatternCanHaveQuantifierSpecifiedOnce2() throws Exception {
-
- Pattern.begin("start").where(dummyCondition()).oneOrMore().optional().times(1);
+ @Test
+ void testPatternCanHaveQuantifierSpecifiedOnce2() {
+ assertThatThrownBy(
+ () ->
+ Pattern.begin("start")
+ .where(dummyCondition())
+ .oneOrMore()
+ .optional()
+ .times(1))
+ .isInstanceOf(MalformedPatternException.class);
}
- @Test(expected = MalformedPatternException.class)
- public void testPatternCanHaveQuantifierSpecifiedOnce3() throws Exception {
-
- Pattern.begin("start").where(dummyCondition()).times(1).oneOrMore();
+ @Test
+ void testPatternCanHaveQuantifierSpecifiedOnce3() {
+ assertThatThrownBy(
+ () -> Pattern.begin("start").where(dummyCondition()).times(1).oneOrMore())
+ .isInstanceOf(MalformedPatternException.class);
}
- @Test(expected = MalformedPatternException.class)
- public void testPatternCanHaveQuantifierSpecifiedOnce4() throws Exception {
-
- Pattern.begin("start").where(dummyCondition()).oneOrMore().oneOrMore();
+ @Test
+ void testPatternCanHaveQuantifierSpecifiedOnce4() {
+ assertThatThrownBy(
+ () ->
+ Pattern.begin("start")
+ .where(dummyCondition())
+ .oneOrMore()
+ .oneOrMore())
+ .isInstanceOf(MalformedPatternException.class);
}
- @Test(expected = MalformedPatternException.class)
- public void testPatternCanHaveQuantifierSpecifiedOnce5() throws Exception {
-
- Pattern.begin("start").where(dummyCondition()).oneOrMore().oneOrMore().optional();
+ @Test
+ void testPatternCanHaveQuantifierSpecifiedOnce5() {
+ assertThatThrownBy(
+ () ->
+ Pattern.begin("start")
+ .where(dummyCondition())
+ .oneOrMore()
+ .oneOrMore()
+ .optional())
+ .isInstanceOf(MalformedPatternException.class);
}
- @Test(expected = MalformedPatternException.class)
- public void testNotNextCannotBeOneOrMore() throws Exception {
- Pattern.begin("start")
- .where(dummyCondition())
- .notNext("not")
- .where(dummyCondition())
- .oneOrMore();
+ @Test
+ void testNotNextCannotBeOneOrMore() {
+ assertThatThrownBy(
+ () ->
+ Pattern.begin("start")
+ .where(dummyCondition())
+ .notNext("not")
+ .where(dummyCondition())
+ .oneOrMore())
+ .isInstanceOf(MalformedPatternException.class);
}
- @Test(expected = MalformedPatternException.class)
- public void testNotNextCannotBeTimes() throws Exception {
- Pattern.begin("start")
- .where(dummyCondition())
- .notNext("not")
- .where(dummyCondition())
- .times(3);
+ @Test
+ void testNotNextCannotBeTimes() {
+ assertThatThrownBy(
+ () ->
+ Pattern.begin("start")
+ .where(dummyCondition())
+ .notNext("not")
+ .where(dummyCondition())
+ .times(3))
+ .isInstanceOf(MalformedPatternException.class);
}
- @Test(expected = MalformedPatternException.class)
- public void testNotNextCannotBeOptional() throws Exception {
-
- Pattern.begin("start")
- .where(dummyCondition())
- .notNext("not")
- .where(dummyCondition())
- .optional();
+ @Test
+ void testNotNextCannotBeOptional() {
+ assertThatThrownBy(
+ () ->
+ Pattern.begin("start")
+ .where(dummyCondition())
+ .notNext("not")
+ .where(dummyCondition())
+ .optional())
+ .isInstanceOf(MalformedPatternException.class);
}
- @Test(expected = MalformedPatternException.class)
- public void testNotFollowedCannotBeOneOrMore() throws Exception {
- Pattern.begin("start")
- .where(dummyCondition())
- .notFollowedBy("not")
- .where(dummyCondition())
- .oneOrMore();
+ @Test
+ void testNotFollowedCannotBeOneOrMore() {
+ assertThatThrownBy(
+ () ->
+ Pattern.begin("start")
+ .where(dummyCondition())
+ .notFollowedBy("not")
+ .where(dummyCondition())
+ .oneOrMore())
+ .isInstanceOf(MalformedPatternException.class);
}
- @Test(expected = MalformedPatternException.class)
- public void testNotFollowedCannotBeTimes() throws Exception {
- Pattern.begin("start")
- .where(dummyCondition())
- .notFollowedBy("not")
- .where(dummyCondition())
- .times(3);
+ @Test
+ void testNotFollowedCannotBeTimes() {
+ assertThatThrownBy(
+ () ->
+ Pattern.begin("start")
+ .where(dummyCondition())
+ .notFollowedBy("not")
+ .where(dummyCondition())
+ .times(3))
+ .isInstanceOf(MalformedPatternException.class);
}
- @Test(expected = MalformedPatternException.class)
- public void testNotFollowedCannotBeOptional() throws Exception {
-
- Pattern.begin("start")
- .where(dummyCondition())
- .notFollowedBy("not")
- .where(dummyCondition())
- .optional();
+ @Test
+ void testNotFollowedCannotBeOptional() {
+ assertThatThrownBy(
+ () ->
+ Pattern.begin("start")
+ .where(dummyCondition())
+ .notFollowedBy("not")
+ .where(dummyCondition())
+ .optional())
+ .isInstanceOf(MalformedPatternException.class);
}
- @Test(expected = MalformedPatternException.class)
- public void testUntilCannotBeAppliedToTimes() throws Exception {
-
- Pattern.begin("start").where(dummyCondition()).times(1).until(dummyCondition());
+ @Test
+ void testUntilCannotBeAppliedToTimes() {
+ assertThatThrownBy(
+ () ->
+ Pattern.begin("start")
+ .where(dummyCondition())
+ .times(1)
+ .until(dummyCondition()))
+ .isInstanceOf(MalformedPatternException.class);
}
- @Test(expected = MalformedPatternException.class)
- public void testUntilCannotBeAppliedToSingleton() throws Exception {
-
- Pattern.begin("start").where(dummyCondition()).until(dummyCondition());
+ @Test
+ void testUntilCannotBeAppliedToSingleton() {
+ assertThatThrownBy(
+ () ->
+ Pattern.begin("start")
+ .where(dummyCondition())
+ .until(dummyCondition()))
+ .isInstanceOf(MalformedPatternException.class);
}
- @Test(expected = MalformedPatternException.class)
- public void testUntilCannotBeAppliedTwice() throws Exception {
-
- Pattern.begin("start")
- .where(dummyCondition())
- .until(dummyCondition())
- .until(dummyCondition());
+ @Test
+ void testUntilCannotBeAppliedTwice() {
+ assertThatThrownBy(
+ () ->
+ Pattern.begin("start")
+ .where(dummyCondition())
+ .until(dummyCondition())
+ .until(dummyCondition()))
+ .isInstanceOf(MalformedPatternException.class);
}
private SimpleCondition<Object> dummyCondition() {
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/utils/NFATestUtilities.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/utils/NFATestUtilities.java
index 644cd481706..7aa741ce151 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/utils/NFATestUtilities.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/utils/NFATestUtilities.java
@@ -22,12 +22,11 @@ import org.apache.flink.cep.Event;
import org.apache.flink.cep.nfa.NFA;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-import org.junit.Assert;
-
-import java.util.Collections;
import java.util.Comparator;
import java.util.List;
+import static org.assertj.core.api.Assertions.assertThat;
+
/** Base method for IT tests of {@link NFA}. It provides utility methods. */
public class NFATestUtilities {
@@ -39,39 +38,17 @@ public class NFATestUtilities {
}
public static void comparePatterns(List<List<Event>> actual, List<List<Event>> expected) {
- Assert.assertEquals(expected.size(), actual.size());
-
+ // the order of events within a single match is not deterministic, so normalize each match
+ // before comparing the matches themselves without regard to their order
for (List<Event> p : actual) {
- Collections.sort(p, new EventComparator());
+ p.sort(new EventComparator());
}
for (List<Event> p : expected) {
- Collections.sort(p, new EventComparator());
+ p.sort(new EventComparator());
}
- Collections.sort(actual, new ListEventComparator());
- Collections.sort(expected, new ListEventComparator());
- Assert.assertArrayEquals(expected.toArray(), actual.toArray());
- }
-
- private static class ListEventComparator implements Comparator<List<Event>> {
-
- @Override
- public int compare(List<Event> o1, List<Event> o2) {
- int sizeComp = Integer.compare(o1.size(), o2.size());
- if (sizeComp == 0) {
- EventComparator comp = new EventComparator();
- for (int i = 0; i < o1.size(); i++) {
- int eventComp = comp.compare(o1.get(i), o2.get(i));
- if (eventComp != 0) {
- return eventComp;
- }
- }
- return 0;
- } else {
- return sizeComp;
- }
- }
+ assertThat(actual).containsExactlyInAnyOrderElementsOf(expected);
}
private static class EventComparator implements Comparator<Event> {
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/utils/OutputAsserter.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/utils/OutputAsserter.java
index 173cec2438e..c1355839e64 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/utils/OutputAsserter.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/utils/OutputAsserter.java
@@ -24,9 +24,7 @@ import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import java.util.Queue;
-import static org.hamcrest.core.Is.is;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
/** Asserter for output from {@link OneInputStreamOperatorTestHarness}. */
public class OutputAsserter {
@@ -55,19 +53,19 @@ public class OutputAsserter {
// This is in case we assert the main output
actual = record;
}
- assertThat(actual, is(expected));
+ assertThat(actual).isEqualTo(expected);
return this;
}
public void hasNoMoreElements() {
- assertTrue(output.isEmpty());
+ assertThat(output).isEmpty();
}
public OutputAsserter watermarkEquals(long timestamp) {
Object record = output.poll();
if (record instanceof Watermark) {
Watermark watermark = (Watermark) record;
- assertThat(watermark.getTimestamp(), is(timestamp));
+ assertThat(watermark.getTimestamp()).isEqualTo(timestamp);
} else {
throw fail(record);
}
diff --git a/flink-libraries/flink-cep/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension b/flink-libraries/flink-cep/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension
new file mode 100644
index 00000000000..75ae4ae1db3
--- /dev/null
+++ b/flink-libraries/flink-cep/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension
@@ -0,0 +1,17 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+org.apache.flink.util.TestLoggerExtension
+org.apache.flink.util.TestNameProviderExtension