This is an automated email from the ASF dual-hosted git repository.

jiabaosun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git

commit 395928901cb99c019d8885d1c39839c33e5ed587
Author: Jiabao Sun <[email protected]>
AuthorDate: Mon Mar 4 18:46:47 2024 +0800

    [FLINK-25544][streaming][JUnit5 Migration] The util package of module 
flink-stream-java
---
 .../AbstractStreamOperatorTestHarnessTest.java     |  31 +--
 .../flink/streaming/util/LatencyStatsTest.java     |  88 +++----
 .../streaming/util/MockDeserializationSchema.java  |  10 +-
 .../streaming/util/MockSerializationSchema.java    |   8 +-
 .../util/ProcessFunctionTestHarnessesTest.java     |  34 ++-
 .../flink/streaming/util/StreamTaskUtil.java       |   2 +-
 .../flink/streaming/util/TestHarnessUtil.java      |  16 +-
 .../streaming/util/keys/ArrayKeySelectorTest.java  |  71 +++--
 .../util/retryable/AsyncRetryStrategiesTest.java   |  18 +-
 .../util/typeutils/FieldAccessorTest.java          | 290 +++++++++++----------
 10 files changed, 275 insertions(+), 293 deletions(-)

diff --git 
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/AbstractStreamOperatorTestHarnessTest.java
 
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/AbstractStreamOperatorTestHarnessTest.java
index b2b34e01dc1..9decf7aaff5 100644
--- 
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/AbstractStreamOperatorTestHarnessTest.java
+++ 
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/AbstractStreamOperatorTestHarnessTest.java
@@ -36,14 +36,11 @@ import 
org.apache.flink.streaming.api.operators.AbstractStreamOperator;
 import org.apache.flink.streaming.api.operators.ProcessOperator;
 import org.apache.flink.util.Collector;
 import org.apache.flink.util.OutputTag;
-import org.apache.flink.util.TestLogger;
 
-import org.junit.Assert;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.ExpectedException;
+import org.junit.jupiter.api.Test;
 
-import static org.hamcrest.CoreMatchers.containsString;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.spy;
@@ -52,26 +49,24 @@ import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 /** Tests for {@link AbstractStreamOperatorTestHarness}. */
-public class AbstractStreamOperatorTestHarnessTest extends TestLogger {
-    @Rule public ExpectedException expectedException = 
ExpectedException.none();
+class AbstractStreamOperatorTestHarnessTest {
 
     @Test
-    public void testInitializeAfterOpenning() throws Throwable {
-        expectedException.expect(IllegalStateException.class);
-        expectedException.expectMessage(
-                containsString("TestHarness has already been initialized."));
-
+    void testInitializeAfterOpenning() throws Throwable {
         AbstractStreamOperatorTestHarness<Integer> result;
         result =
                 new AbstractStreamOperatorTestHarness<>(
                         new AbstractStreamOperator<Integer>() {}, 1, 1, 0);
         result.setup();
         result.open();
-        result.initializeState(OperatorSubtaskState.builder().build());
+
+        assertThatThrownBy(() -> 
result.initializeState(OperatorSubtaskState.builder().build()))
+                .isInstanceOf(IllegalStateException.class)
+                .hasMessageContaining("TestHarness has already been 
initialized.");
     }
 
     @Test
-    public void testSetTtlTimeProvider() throws Exception {
+    void testSetTtlTimeProvider() throws Exception {
         AbstractStreamOperator<Integer> operator = new 
AbstractStreamOperator<Integer>() {};
         try (AbstractStreamOperatorTestHarness<Integer> result =
                 new AbstractStreamOperatorTestHarness<>(operator, 1, 1, 0)) {
@@ -97,14 +92,14 @@ public class AbstractStreamOperatorTestHarnessTest extends 
TestLogger {
             keyedStateBackend.setCurrentKey(1);
             result.setStateTtlProcessingTime(0L);
             state.update(expectedValue);
-            Assert.assertEquals(expectedValue, (int) state.value());
+            assertThat(state.value()).isEqualTo(expectedValue);
             result.setStateTtlProcessingTime(timeToLive.toMilliseconds() + 1);
-            Assert.assertNull(state.value());
+            assertThat(state.value()).isNull();
         }
     }
 
     @Test
-    public void testSideOutputTypeInformation() throws Throwable {
+    void testSideOutputTypeInformation() throws Throwable {
         final int probe = 12;
         final TypeSerializer<Integer> typeSerializer = 
spy(TypeSerializer.class);
 
diff --git 
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/LatencyStatsTest.java
 
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/LatencyStatsTest.java
index df5a43df126..5e423bda462 100644
--- 
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/LatencyStatsTest.java
+++ 
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/LatencyStatsTest.java
@@ -28,17 +28,17 @@ import 
org.apache.flink.runtime.metrics.groups.GenericMetricGroup;
 import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
 import org.apache.flink.runtime.metrics.util.TestingMetricRegistry;
 import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
-import org.apache.flink.util.TestLogger;
 
-import org.junit.Assert;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
 
 import java.util.ArrayList;
 import java.util.List;
 import java.util.function.Consumer;
 
+import static org.assertj.core.api.Assertions.assertThat;
+
 /** Tests for the {@link LatencyStats}. */
-public class LatencyStatsTest extends TestLogger {
+class LatencyStatsTest {
 
     private static final OperatorID OPERATOR_ID = new OperatorID();
     private static final OperatorID SOURCE_ID_1 = new OperatorID();
@@ -49,70 +49,70 @@ public class LatencyStatsTest extends TestLogger {
     private static final String PARENT_GROUP_NAME = "parent";
 
     @Test
-    public void testLatencyStatsSingle() {
+    void testLatencyStatsSingle() {
         testLatencyStats(
                 LatencyStats.Granularity.SINGLE,
                 registrations -> {
-                    Assert.assertEquals(1, registrations.size());
+                    assertThat(registrations).hasSize(1);
 
                     {
                         final Tuple2<String, Histogram> registration = 
registrations.get(0);
                         assertName(registration.f0);
-                        Assert.assertEquals(5, registration.f1.getCount());
+                        assertThat(registration.f1.getCount()).isEqualTo(5);
                     }
                 });
     }
 
     @Test
-    public void testLatencyStatsOperator() {
+    void testLatencyStatsOperator() {
         testLatencyStats(
                 LatencyStats.Granularity.OPERATOR,
                 registrations -> {
-                    Assert.assertEquals(2, registrations.size());
+                    assertThat(registrations).hasSize(2);
 
                     {
                         final Tuple2<String, Histogram> registration = 
registrations.get(0);
                         assertName(registration.f0, SOURCE_ID_1);
-                        Assert.assertEquals(3, registration.f1.getCount());
+                        assertThat(registration.f1.getCount()).isEqualTo(3);
                     }
 
                     {
                         final Tuple2<String, Histogram> registration = 
registrations.get(1);
                         assertName(registration.f0, SOURCE_ID_2);
-                        Assert.assertEquals(2, registration.f1.getCount());
+                        assertThat(registration.f1.getCount()).isEqualTo(2);
                     }
                 });
     }
 
     @Test
-    public void testLatencyStatsSubtask() {
+    void testLatencyStatsSubtask() {
         testLatencyStats(
                 LatencyStats.Granularity.SUBTASK,
                 registrations -> {
-                    Assert.assertEquals(4, registrations.size());
+                    assertThat(registrations).hasSize(4);
 
                     {
                         final Tuple2<String, Histogram> registration = 
registrations.get(0);
                         assertName(registration.f0, SOURCE_ID_1, 0);
-                        Assert.assertEquals(2, registration.f1.getCount());
+                        assertThat(registration.f1.getCount()).isEqualTo(2);
                     }
 
                     {
                         final Tuple2<String, Histogram> registration = 
registrations.get(1);
                         assertName(registration.f0, SOURCE_ID_1, 1);
-                        Assert.assertEquals(1, registration.f1.getCount());
+                        assertThat(registration.f1.getCount()).isOne();
                     }
 
                     {
                         final Tuple2<String, Histogram> registration = 
registrations.get(2);
                         assertName(registration.f0, SOURCE_ID_2, 2);
-                        Assert.assertEquals(1, registration.f1.getCount());
+                        assertThat(registration.f1.getCount()).isOne();
                     }
 
                     {
                         final Tuple2<String, Histogram> registration = 
registrations.get(3);
                         assertName(registration.f0, SOURCE_ID_2, 3);
-                        Assert.assertEquals(1, registration.f1.getCount());
+                        assertThat(registration.f1.getCount()).isOne();
                     }
                 });
     }
@@ -165,41 +165,41 @@ public class LatencyStatsTest extends TestLogger {
 
     private static void assertName(final String registrationName) {
         final String sanitizedName = sanitizeName(registrationName);
-        Assert.assertEquals(
-                "operator_id."
-                        + OPERATOR_ID
-                        + ".operator_subtask_index."
-                        + OPERATOR_SUBTASK_INDEX
-                        + ".latency",
-                sanitizedName);
+        assertThat(sanitizedName)
+                .isEqualTo(
+                        "operator_id."
+                                + OPERATOR_ID
+                                + ".operator_subtask_index."
+                                + OPERATOR_SUBTASK_INDEX
+                                + ".latency");
     }
 
     private static void assertName(final String registrationName, final 
OperatorID sourceId) {
         final String sanitizedName = sanitizeName(registrationName);
-        Assert.assertEquals(
-                "source_id."
-                        + sourceId
-                        + ".operator_id."
-                        + OPERATOR_ID
-                        + ".operator_subtask_index."
-                        + OPERATOR_SUBTASK_INDEX
-                        + ".latency",
-                sanitizedName);
+        assertThat(sanitizedName)
+                .isEqualTo(
+                        "source_id."
+                                + sourceId
+                                + ".operator_id."
+                                + OPERATOR_ID
+                                + ".operator_subtask_index."
+                                + OPERATOR_SUBTASK_INDEX
+                                + ".latency");
     }
 
     private static void assertName(
             final String registrationName, final OperatorID sourceId, final 
int sourceIndex) {
         final String sanitizedName = sanitizeName(registrationName);
-        Assert.assertEquals(
-                "source_id."
-                        + sourceId
-                        + ".source_subtask_index."
-                        + sourceIndex
-                        + ".operator_id."
-                        + OPERATOR_ID
-                        + ".operator_subtask_index."
-                        + OPERATOR_SUBTASK_INDEX
-                        + ".latency",
-                sanitizedName);
+        assertThat(sanitizedName)
+                .isEqualTo(
+                        "source_id."
+                                + sourceId
+                                + ".source_subtask_index."
+                                + sourceIndex
+                                + ".operator_id."
+                                + OPERATOR_ID
+                                + ".operator_subtask_index."
+                                + OPERATOR_SUBTASK_INDEX
+                                + ".latency");
     }
 }
diff --git 
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/MockDeserializationSchema.java
 
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/MockDeserializationSchema.java
index 27c90e4213c..5ab3116cd10 100644
--- 
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/MockDeserializationSchema.java
+++ 
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/MockDeserializationSchema.java
@@ -24,9 +24,7 @@ import org.apache.flink.metrics.MetricGroup;
 
 import java.io.IOException;
 
-import static org.hamcrest.Matchers.is;
-import static org.hamcrest.Matchers.notNullValue;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
 
 /**
  * A mocked {@link DeserializationSchema} that verifies that {@link
@@ -35,14 +33,14 @@ import static org.junit.Assert.assertThat;
  *
  * <p>It does not implement any of the deserialization methods.
  */
-public class MockDeserializationSchema<T> implements DeserializationSchema<T> {
+class MockDeserializationSchema<T> implements DeserializationSchema<T> {
 
     private boolean openCalled = false;
 
     @Override
     public void open(InitializationContext context) throws Exception {
-        assertThat("Open was called multiple times", openCalled, is(false));
-        assertThat(context.getMetricGroup(), notNullValue(MetricGroup.class));
+        assertThat(openCalled).as("Open was called multiple times").isFalse();
+        assertThat(context.getMetricGroup()).isNotNull();
         this.openCalled = true;
     }
 
diff --git 
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/MockSerializationSchema.java
 
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/MockSerializationSchema.java
index 4442fddf7b1..cfad115f375 100644
--- 
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/MockSerializationSchema.java
+++ 
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/MockSerializationSchema.java
@@ -21,9 +21,7 @@ package org.apache.flink.streaming.util;
 import org.apache.flink.api.common.serialization.SerializationSchema;
 import org.apache.flink.metrics.MetricGroup;
 
-import static org.hamcrest.Matchers.is;
-import static org.hamcrest.Matchers.notNullValue;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
 
 /**
  * A mocked {@link SerializationSchema} that verifies that {@link
@@ -38,8 +36,8 @@ public class MockSerializationSchema<T> implements 
SerializationSchema<T> {
 
     @Override
     public void open(SerializationSchema.InitializationContext context) throws 
Exception {
-        assertThat("Open was called multiple times", openCalled, is(false));
-        assertThat(context.getMetricGroup(), notNullValue(MetricGroup.class));
+        assertThat(openCalled).as("Open was called multiple times").isFalse();
+        assertThat(context.getMetricGroup()).isNotNull();
         this.openCalled = true;
     }
 
diff --git 
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/ProcessFunctionTestHarnessesTest.java
 
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/ProcessFunctionTestHarnessesTest.java
index 40ad6eddec4..d30199f6b0f 100644
--- 
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/ProcessFunctionTestHarnessesTest.java
+++ 
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/ProcessFunctionTestHarnessesTest.java
@@ -28,20 +28,16 @@ import 
org.apache.flink.streaming.api.functions.co.CoProcessFunction;
 import 
org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
 import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
 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.Arrays;
-import java.util.Collections;
-
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
 
 /** Tests for {@link ProcessFunctionTestHarnesses}. */
-public class ProcessFunctionTestHarnessesTest extends TestLogger {
+class ProcessFunctionTestHarnessesTest {
 
     @Test
-    public void testHarnessForProcessFunction() throws Exception {
+    void testHarnessForProcessFunction() throws Exception {
         ProcessFunction<Integer, Integer> function =
                 new ProcessFunction<Integer, Integer>() {
 
@@ -56,11 +52,11 @@ public class ProcessFunctionTestHarnessesTest extends 
TestLogger {
 
         harness.processElement(1, 10);
 
-        assertEquals(harness.extractOutputValues(), 
Collections.singletonList(1));
+        assertThat(harness.extractOutputValues()).containsExactly(1);
     }
 
     @Test
-    public void testHarnessForKeyedProcessFunction() throws Exception {
+    void testHarnessForKeyedProcessFunction() throws Exception {
         KeyedProcessFunction<Integer, Integer, Integer> function =
                 new KeyedProcessFunction<Integer, Integer, Integer>() {
                     @Override
@@ -75,11 +71,11 @@ public class ProcessFunctionTestHarnessesTest extends 
TestLogger {
 
         harness.processElement(1, 10);
 
-        assertEquals(harness.extractOutputValues(), 
Collections.singletonList(1));
+        assertThat(harness.extractOutputValues()).containsExactly(1);
     }
 
     @Test
-    public void testHarnessForCoProcessFunction() throws Exception {
+    void testHarnessForCoProcessFunction() throws Exception {
         CoProcessFunction<Integer, String, Integer> function =
                 new CoProcessFunction<Integer, String, Integer>() {
 
@@ -101,11 +97,11 @@ public class ProcessFunctionTestHarnessesTest extends 
TestLogger {
         harness.processElement2("0", 1);
         harness.processElement1(1, 10);
 
-        assertEquals(harness.extractOutputValues(), Arrays.asList(0, 1));
+        assertThat(harness.extractOutputValues()).containsExactly(0, 1);
     }
 
     @Test
-    public void testHarnessForKeyedCoProcessFunction() throws Exception {
+    void testHarnessForKeyedCoProcessFunction() throws Exception {
         KeyedCoProcessFunction<Integer, Integer, Integer, Integer> function =
                 new KeyedCoProcessFunction<Integer, Integer, Integer, 
Integer>() {
 
@@ -129,11 +125,11 @@ public class ProcessFunctionTestHarnessesTest extends 
TestLogger {
         harness.processElement1(0, 1);
         harness.processElement2(1, 10);
 
-        assertEquals(harness.extractOutputValues(), Arrays.asList(0, 1));
+        assertThat(harness.extractOutputValues()).containsExactly(0, 1);
     }
 
     @Test
-    public void testHarnessForBroadcastProcessFunction() throws Exception {
+    void testHarnessForBroadcastProcessFunction() throws Exception {
         BroadcastProcessFunction<Integer, String, Integer> function =
                 new BroadcastProcessFunction<Integer, String, Integer>() {
 
@@ -156,11 +152,11 @@ public class ProcessFunctionTestHarnessesTest extends 
TestLogger {
         harness.processBroadcastElement("0", 1);
         harness.processElement(1, 10);
 
-        assertEquals(harness.extractOutputValues(), Arrays.asList(0, 1));
+        assertThat(harness.extractOutputValues()).containsExactly(0, 1);
     }
 
     @Test
-    public void testHarnessForKeyedBroadcastProcessFunction() throws Exception 
{
+    void testHarnessForKeyedBroadcastProcessFunction() throws Exception {
         KeyedBroadcastProcessFunction<Integer, Integer, String, Integer> 
function =
                 new KeyedBroadcastProcessFunction<Integer, Integer, String, 
Integer>() {
 
@@ -189,6 +185,6 @@ public class ProcessFunctionTestHarnessesTest extends 
TestLogger {
         harness.processBroadcastElement("0", 1);
         harness.processElement(1, 10);
 
-        assertEquals(harness.extractOutputValues(), Arrays.asList(0, 1));
+        assertThat(harness.extractOutputValues()).containsExactly(0, 1);
     }
 }
diff --git 
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/StreamTaskUtil.java
 
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/StreamTaskUtil.java
index b1ce51430b1..cc79ac204ef 100644
--- 
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/StreamTaskUtil.java
+++ 
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/StreamTaskUtil.java
@@ -23,7 +23,7 @@ import org.apache.flink.streaming.runtime.tasks.StreamTask;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 
-import static org.junit.Assert.fail;
+import static org.junit.jupiter.api.Assertions.fail;
 
 /** Utils for working with StreamTask. */
 public class StreamTaskUtil {
diff --git 
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/TestHarnessUtil.java
 
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/TestHarnessUtil.java
index e3a0f49da50..9605e0fa663 100644
--- 
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/TestHarnessUtil.java
+++ 
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/TestHarnessUtil.java
@@ -22,10 +22,6 @@ import 
org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
 import org.apache.flink.streaming.api.watermark.Watermark;
 import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
 
-import org.apache.flink.shaded.guava31.com.google.common.collect.Iterables;
-
-import org.junit.Assert;
-
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Comparator;
@@ -35,7 +31,7 @@ import java.util.List;
 import java.util.Queue;
 import java.util.stream.Collectors;
 
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
 
 /** Utils for working with the various test harnesses. */
 public class TestHarnessUtil {
@@ -56,7 +52,7 @@ public class TestHarnessUtil {
      * Compare the two queues containing operator/task output by converting 
them to an array first.
      */
     public static <T> void assertOutputEquals(String message, Queue<T> 
expected, Queue<T> actual) {
-        Assert.assertArrayEquals(message, expected.toArray(), 
actual.toArray());
+        assertThat(actual.toArray()).as(message).isEqualTo(expected.toArray());
     }
 
     /**
@@ -67,7 +63,7 @@ public class TestHarnessUtil {
             Iterable<Object> expected,
             Iterable<Object> actual,
             Comparator<Object> comparator) {
-        assertEquals(Iterables.size(expected), Iterables.size(actual));
+        assertThat(actual).hasSameSizeAs(expected);
 
         // first, compare only watermarks, their position should be 
deterministic
         Iterator<Object> exIt = expected.iterator();
@@ -76,7 +72,7 @@ public class TestHarnessUtil {
             Object nextEx = exIt.next();
             Object nextAct = actIt.next();
             if (nextEx instanceof Watermark) {
-                assertEquals(nextEx, nextAct);
+                assertThat(nextAct).isEqualTo(nextEx);
             }
         }
 
@@ -101,7 +97,7 @@ public class TestHarnessUtil {
         Arrays.sort(sortedExpected, comparator);
         Arrays.sort(sortedActual, comparator);
 
-        Assert.assertArrayEquals(message, sortedExpected, sortedActual);
+        assertThat(sortedActual).as(message).isEqualTo(sortedExpected);
     }
 
     /**
@@ -119,7 +115,7 @@ public class TestHarnessUtil {
                 highestWatermark = ((Watermark) 
elem).asWatermark().getTimestamp();
             } else if (elem instanceof StreamRecord) {
                 boolean dataIsOnTime = highestWatermark < ((StreamRecord) 
elem).getTimestamp();
-                Assert.assertTrue("Late data was emitted after join", 
dataIsOnTime);
+                assertThat(dataIsOnTime).as("Late data was emitted after 
join").isTrue();
             }
         }
     }
diff --git 
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/keys/ArrayKeySelectorTest.java
 
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/keys/ArrayKeySelectorTest.java
index bd74866f892..ffd8f66951f 100644
--- 
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/keys/ArrayKeySelectorTest.java
+++ 
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/keys/ArrayKeySelectorTest.java
@@ -23,63 +23,50 @@ import 
org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
 import org.apache.flink.api.java.tuple.Tuple1;
 import org.apache.flink.api.java.tuple.Tuple2;
 
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.fail;
+import static org.assertj.core.api.Assertions.assertThat;
 
 /** Tests key selectors on arrays. */
-public class ArrayKeySelectorTest {
+class ArrayKeySelectorTest {
 
     @Test
-    public void testObjectArrays() {
-        try {
-            String[] array1 = {"a", "b", "c", "d", "e"};
-            String[] array2 = {"v", "w", "x", "y", "z"};
+    void testObjectArrays() {
+        String[] array1 = {"a", "b", "c", "d", "e"};
+        String[] array2 = {"v", "w", "x", "y", "z"};
 
-            KeySelectorUtil.ArrayKeySelector<String[]> singleFieldSelector =
-                    KeySelectorUtil.getSelectorForArray(
-                            new int[] {1}, 
BasicArrayTypeInfo.STRING_ARRAY_TYPE_INFO);
+        KeySelectorUtil.ArrayKeySelector<String[]> singleFieldSelector =
+                KeySelectorUtil.getSelectorForArray(
+                        new int[] {1}, 
BasicArrayTypeInfo.STRING_ARRAY_TYPE_INFO);
 
-            assertEquals(new Tuple1<>("b"), 
singleFieldSelector.getKey(array1));
-            assertEquals(new Tuple1<>("w"), 
singleFieldSelector.getKey(array2));
+        assertThat(singleFieldSelector.getKey(array1)).isEqualTo(new 
Tuple1<>("b"));
+        assertThat(singleFieldSelector.getKey(array2)).isEqualTo(new 
Tuple1<>("w"));
 
-            KeySelectorUtil.ArrayKeySelector<String[]> twoFieldsSelector =
-                    KeySelectorUtil.getSelectorForArray(
-                            new int[] {3, 0}, 
BasicArrayTypeInfo.STRING_ARRAY_TYPE_INFO);
+        KeySelectorUtil.ArrayKeySelector<String[]> twoFieldsSelector =
+                KeySelectorUtil.getSelectorForArray(
+                        new int[] {3, 0}, 
BasicArrayTypeInfo.STRING_ARRAY_TYPE_INFO);
 
-            assertEquals(new Tuple2<>("d", "a"), 
twoFieldsSelector.getKey(array1));
-            assertEquals(new Tuple2<>("y", "v"), 
twoFieldsSelector.getKey(array2));
-
-        } catch (Exception e) {
-            e.printStackTrace();
-            fail(e.getMessage());
-        }
+        assertThat(twoFieldsSelector.getKey(array1)).isEqualTo(new 
Tuple2<>("d", "a"));
+        assertThat(twoFieldsSelector.getKey(array2)).isEqualTo(new 
Tuple2<>("y", "v"));
     }
 
     @Test
-    public void testPrimitiveArrays() {
-        try {
-            int[] array1 = {1, 2, 3, 4, 5};
-            int[] array2 = {-5, -4, -3, -2, -1, 0};
-
-            KeySelectorUtil.ArrayKeySelector<int[]> singleFieldSelector =
-                    KeySelectorUtil.getSelectorForArray(
-                            new int[] {1}, 
PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO);
+    void testPrimitiveArrays() {
+        int[] array1 = {1, 2, 3, 4, 5};
+        int[] array2 = {-5, -4, -3, -2, -1, 0};
 
-            assertEquals(new Tuple1<>(2), singleFieldSelector.getKey(array1));
-            assertEquals(new Tuple1<>(-4), singleFieldSelector.getKey(array2));
+        KeySelectorUtil.ArrayKeySelector<int[]> singleFieldSelector =
+                KeySelectorUtil.getSelectorForArray(
+                        new int[] {1}, 
PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO);
 
-            KeySelectorUtil.ArrayKeySelector<int[]> twoFieldsSelector =
-                    KeySelectorUtil.getSelectorForArray(
-                            new int[] {3, 0}, 
PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO);
+        assertThat(singleFieldSelector.getKey(array1)).isEqualTo(new 
Tuple1<>(2));
+        assertThat(singleFieldSelector.getKey(array2)).isEqualTo(new 
Tuple1<>(-4));
 
-            assertEquals(new Tuple2<>(4, 1), twoFieldsSelector.getKey(array1));
-            assertEquals(new Tuple2<>(-2, -5), 
twoFieldsSelector.getKey(array2));
+        KeySelectorUtil.ArrayKeySelector<int[]> twoFieldsSelector =
+                KeySelectorUtil.getSelectorForArray(
+                        new int[] {3, 0}, 
PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO);
 
-        } catch (Exception e) {
-            e.printStackTrace();
-            fail(e.getMessage());
-        }
+        assertThat(twoFieldsSelector.getKey(array1)).isEqualTo(new Tuple2<>(4, 
1));
+        assertThat(twoFieldsSelector.getKey(array2)).isEqualTo(new 
Tuple2<>(-2, -5));
     }
 }
diff --git 
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/retryable/AsyncRetryStrategiesTest.java
 
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/retryable/AsyncRetryStrategiesTest.java
index 606119912ce..df1525c8e19 100644
--- 
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/retryable/AsyncRetryStrategiesTest.java
+++ 
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/retryable/AsyncRetryStrategiesTest.java
@@ -19,16 +19,16 @@
 package org.apache.flink.streaming.util.retryable;
 
 import org.apache.flink.streaming.api.functions.async.AsyncRetryStrategy;
-import org.apache.flink.util.TestLogger;
 
-import org.junit.Assert;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
 
 /** Tests for the {@link AsyncRetryStrategies}. */
-public class AsyncRetryStrategiesTest extends TestLogger {
+class AsyncRetryStrategiesTest {
 
     @Test
-    public void testExponentialBackoffDelayRetryStrategy() {
+    void testExponentialBackoffDelayRetryStrategy() {
         int maxAttempts = 10;
         long initialDelay = 100L;
         long maxRetryDelay = 2000L;
@@ -39,16 +39,16 @@ public class AsyncRetryStrategiesTest extends TestLogger {
                                 maxAttempts, initialDelay, maxRetryDelay, 
multiplier)
                         .build();
 
-        
Assert.assertTrue(exponentialBackoffDelayRetryStrategy.canRetry(maxAttempts));
-        
Assert.assertFalse(exponentialBackoffDelayRetryStrategy.canRetry(maxAttempts + 
1));
+        
assertThat(exponentialBackoffDelayRetryStrategy.canRetry(maxAttempts)).isTrue();
+        assertThat(exponentialBackoffDelayRetryStrategy.canRetry(maxAttempts + 
1)).isFalse();
 
         // test if this strategy can be reused.
         for (int j = 1; j <= 5; j++) {
             long currentDelay = initialDelay;
 
             for (int i = 1; i <= maxAttempts; i++) {
-                Assert.assertEquals(
-                        currentDelay, 
exponentialBackoffDelayRetryStrategy.getBackoffTimeMillis(i));
+                
assertThat(exponentialBackoffDelayRetryStrategy.getBackoffTimeMillis(i))
+                        .isEqualTo(currentDelay);
                 currentDelay = Math.min((long) (currentDelay * multiplier), 
maxRetryDelay);
             }
         }
diff --git 
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/typeutils/FieldAccessorTest.java
 
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/typeutils/FieldAccessorTest.java
index af1e04e8ec6..c92a02821dd 100644
--- 
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/typeutils/FieldAccessorTest.java
+++ 
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/typeutils/FieldAccessorTest.java
@@ -30,88 +30,90 @@ import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.api.java.typeutils.TupleTypeInfo;
 import org.apache.flink.api.java.typeutils.TypeExtractor;
 
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
 
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /** Tests for field accessors. */
-public class FieldAccessorTest {
+class FieldAccessorTest {
 
     // Note, that AggregationFunctionTest indirectly also tests FieldAccessors.
     // ProductFieldAccessors are tested in CaseClassFieldAccessorTest.
 
     @Test
-    public void testFlatTuple() {
+    void testFlatTuple() {
         Tuple2<String, Integer> t = Tuple2.of("aa", 5);
         TupleTypeInfo<Tuple2<String, Integer>> tpeInfo =
                 (TupleTypeInfo<Tuple2<String, Integer>>) 
TypeExtractor.getForObject(t);
 
         FieldAccessor<Tuple2<String, Integer>, String> f0 =
                 FieldAccessorFactory.getAccessor(tpeInfo, "f0", null);
-        assertEquals(String.class, f0.getFieldType().getTypeClass());
-        assertEquals("aa", f0.get(t));
-        assertEquals("aa", t.f0);
+        assertThat(f0.getFieldType().getTypeClass()).isEqualTo(String.class);
+        assertThat(f0.get(t)).isEqualTo("aa");
+        assertThat(t.f0).isEqualTo("aa");
         t = f0.set(t, "b");
-        assertEquals("b", f0.get(t));
-        assertEquals("b", t.f0);
+        assertThat(f0.get(t)).isEqualTo("b");
+        assertThat(t.f0).isEqualTo("b");
 
         FieldAccessor<Tuple2<String, Integer>, Integer> f1 =
                 FieldAccessorFactory.getAccessor(tpeInfo, "f1", null);
-        assertEquals(Integer.class, f1.getFieldType().getTypeClass());
-        assertEquals(5, (int) f1.get(t));
-        assertEquals(5, (int) t.f1);
+        assertThat(f1.getFieldType().getTypeClass()).isEqualTo(Integer.class);
+        assertThat(f1.get(t)).isEqualTo(5);
+        assertThat(t.f1).isEqualTo(5);
         t = f1.set(t, 7);
-        assertEquals(7, (int) f1.get(t));
-        assertEquals(7, (int) t.f1);
-        assertEquals("b", f0.get(t));
-        assertEquals("b", t.f0);
+        assertThat(f1.get(t)).isEqualTo(7);
+        assertThat(t.f1).isEqualTo(7);
+        assertThat(f0.get(t)).isEqualTo("b");
+        assertThat(t.f0).isEqualTo("b");
 
         FieldAccessor<Tuple2<String, Integer>, Integer> f1n =
                 FieldAccessorFactory.getAccessor(tpeInfo, 1, null);
-        assertEquals(Integer.class, f1n.getFieldType().getTypeClass());
-        assertEquals(7, (int) f1n.get(t));
-        assertEquals(7, (int) t.f1);
+        assertThat(f1n.getFieldType().getTypeClass()).isEqualTo(Integer.class);
+        assertThat(f1n.get(t)).isEqualTo(7);
+        assertThat(t.f1).isEqualTo(7);
         t = f1n.set(t, 10);
-        assertEquals(10, (int) f1n.get(t));
-        assertEquals(10, (int) f1.get(t));
-        assertEquals(10, (int) t.f1);
-        assertEquals("b", f0.get(t));
-        assertEquals("b", t.f0);
+        assertThat(f1n.get(t)).isEqualTo(10);
+        assertThat(f1.get(t)).isEqualTo(10);
+        assertThat(t.f1).isEqualTo(10);
+        assertThat(f0.get(t)).isEqualTo("b");
+        assertThat(t.f0).isEqualTo("b");
 
         FieldAccessor<Tuple2<String, Integer>, Integer> f1ns =
                 FieldAccessorFactory.getAccessor(tpeInfo, "1", null);
-        assertEquals(Integer.class, f1ns.getFieldType().getTypeClass());
-        assertEquals(10, (int) f1ns.get(t));
-        assertEquals(10, (int) t.f1);
+        
assertThat(f1ns.getFieldType().getTypeClass()).isEqualTo(Integer.class);
+        assertThat(f1ns.get(t)).isEqualTo(10);
+        assertThat(t.f1).isEqualTo(10);
         t = f1ns.set(t, 11);
-        assertEquals(11, (int) f1ns.get(t));
-        assertEquals(11, (int) f1.get(t));
-        assertEquals(11, (int) t.f1);
-        assertEquals("b", f0.get(t));
-        assertEquals("b", t.f0);
+        assertThat(f1ns.get(t)).isEqualTo(11);
+        assertThat(f1.get(t)).isEqualTo(11);
+        assertThat(t.f1).isEqualTo(11);
+        assertThat(f0.get(t)).isEqualTo("b");
+        assertThat(t.f0).isEqualTo("b");
 
         // This is technically valid (the ".0" is selecting the 0th field of a 
basic type).
         FieldAccessor<Tuple2<String, Integer>, String> f0f0 =
                 FieldAccessorFactory.getAccessor(tpeInfo, "f0.0", null);
-        assertEquals(String.class, f0f0.getFieldType().getTypeClass());
-        assertEquals("b", f0f0.get(t));
-        assertEquals("b", t.f0);
+        assertThat(f0f0.getFieldType().getTypeClass()).isEqualTo(String.class);
+        assertThat(f0f0.get(t)).isEqualTo("b");
+        assertThat(t.f0).isEqualTo("b");
         t = f0f0.set(t, "cc");
-        assertEquals("cc", f0f0.get(t));
-        assertEquals("cc", t.f0);
+        assertThat(f0f0.get(t)).isEqualTo("cc");
+        assertThat(t.f0).isEqualTo("cc");
     }
 
-    @Test(expected = CompositeType.InvalidFieldReferenceException.class)
-    public void testIllegalFlatTuple() {
+    @Test
+    void testIllegalFlatTuple() {
         Tuple2<String, Integer> t = Tuple2.of("aa", 5);
         TupleTypeInfo<Tuple2<String, Integer>> tpeInfo =
                 (TupleTypeInfo<Tuple2<String, Integer>>) 
TypeExtractor.getForObject(t);
 
-        FieldAccessorFactory.getAccessor(tpeInfo, "illegal", null);
+        assertThatThrownBy(() -> FieldAccessorFactory.getAccessor(tpeInfo, 
"illegal", null))
+                
.isInstanceOf(CompositeType.InvalidFieldReferenceException.class);
     }
 
     @Test
-    public void testTupleInTuple() {
+    void testTupleInTuple() {
         Tuple2<String, Tuple3<Integer, Long, Double>> t = Tuple2.of("aa", 
Tuple3.of(5, 9L, 2.0));
         TupleTypeInfo<Tuple2<String, Tuple3<Integer, Long, Double>>> tpeInfo =
                 (TupleTypeInfo<Tuple2<String, Tuple3<Integer, Long, Double>>>)
@@ -119,50 +121,55 @@ public class FieldAccessorTest {
 
         FieldAccessor<Tuple2<String, Tuple3<Integer, Long, Double>>, String> 
f0 =
                 FieldAccessorFactory.getAccessor(tpeInfo, "f0", null);
-        assertEquals(String.class, f0.getFieldType().getTypeClass());
-        assertEquals("aa", f0.get(t));
-        assertEquals("aa", t.f0);
+        assertThat(f0.getFieldType().getTypeClass()).isEqualTo(String.class);
+        assertThat(f0.get(t)).isEqualTo("aa");
+        assertThat(t.f0).isEqualTo("aa");
 
         FieldAccessor<Tuple2<String, Tuple3<Integer, Long, Double>>, Double> 
f1f2 =
                 FieldAccessorFactory.getAccessor(tpeInfo, "f1.f2", null);
-        assertEquals(Double.class, f1f2.getFieldType().getTypeClass());
-        assertEquals(2.0, f1f2.get(t), 0);
-        assertEquals(2.0, t.f1.f2, 0);
+        assertThat(f1f2.getFieldType().getTypeClass()).isEqualTo(Double.class);
+        assertThat(f1f2.get(t)).isEqualTo(2.0);
+        assertThat(t.f1.f2).isEqualTo(2.0);
         t = f1f2.set(t, 3.0);
-        assertEquals(3.0, f1f2.get(t), 0);
-        assertEquals(3.0, t.f1.f2, 0);
-        assertEquals("aa", f0.get(t));
-        assertEquals("aa", t.f0);
+        assertThat(f1f2.get(t)).isEqualTo(3.0);
+        assertThat(t.f1.f2).isEqualTo(3.0);
+        assertThat(f0.get(t)).isEqualTo("aa");
+        assertThat(t.f0).isEqualTo("aa");
 
         FieldAccessor<Tuple2<String, Tuple3<Integer, Long, Double>>, 
Tuple3<Integer, Long, Double>>
                 f1 = FieldAccessorFactory.getAccessor(tpeInfo, "f1", null);
-        assertEquals(Tuple3.class, f1.getFieldType().getTypeClass());
-        assertEquals(Tuple3.of(5, 9L, 3.0), f1.get(t));
-        assertEquals(Tuple3.of(5, 9L, 3.0), t.f1);
+        assertThat(f1.getFieldType().getTypeClass()).isEqualTo(Tuple3.class);
+        assertThat(f1.get(t)).isEqualTo(Tuple3.of(5, 9L, 3.0));
+        assertThat(t.f1).isEqualTo(Tuple3.of(5, 9L, 3.0));
         t = f1.set(t, Tuple3.of(8, 12L, 4.0));
-        assertEquals(Tuple3.of(8, 12L, 4.0), f1.get(t));
-        assertEquals(Tuple3.of(8, 12L, 4.0), t.f1);
-        assertEquals("aa", f0.get(t));
-        assertEquals("aa", t.f0);
+        assertThat(f1.get(t)).isEqualTo(Tuple3.of(8, 12L, 4.0));
+        assertThat(t.f1).isEqualTo(Tuple3.of(8, 12L, 4.0));
+        assertThat(f0.get(t)).isEqualTo("aa");
+        assertThat(t.f0).isEqualTo("aa");
 
         FieldAccessor<Tuple2<String, Tuple3<Integer, Long, Double>>, 
Tuple3<Integer, Long, Double>>
                 f1n = FieldAccessorFactory.getAccessor(tpeInfo, 1, null);
-        assertEquals(Tuple3.class, f1n.getFieldType().getTypeClass());
-        assertEquals(Tuple3.of(8, 12L, 4.0), f1n.get(t));
-        assertEquals(Tuple3.of(8, 12L, 4.0), t.f1);
+        assertThat(f1n.getFieldType().getTypeClass()).isEqualTo(Tuple3.class);
+        assertThat(f1n.get(t)).isEqualTo(Tuple3.of(8, 12L, 4.0));
+        assertThat(t.f1).isEqualTo(Tuple3.of(8, 12L, 4.0));
         t = f1n.set(t, Tuple3.of(10, 13L, 5.0));
-        assertEquals(Tuple3.of(10, 13L, 5.0), f1n.get(t));
-        assertEquals(Tuple3.of(10, 13L, 5.0), f1.get(t));
-        assertEquals(Tuple3.of(10, 13L, 5.0), t.f1);
-        assertEquals("aa", f0.get(t));
-        assertEquals("aa", t.f0);
+        assertThat(f1n.get(t)).isEqualTo(Tuple3.of(10, 13L, 5.0));
+        assertThat(f1.get(t)).isEqualTo(Tuple3.of(10, 13L, 5.0));
+        assertThat(t.f1).isEqualTo(Tuple3.of(10, 13L, 5.0));
+        assertThat(f0.get(t)).isEqualTo("aa");
+        assertThat(t.f0).isEqualTo("aa");
     }
 
-    @Test(expected = CompositeType.InvalidFieldReferenceException.class)
-    @SuppressWarnings("unchecked")
-    public void testIllegalTupleField() {
-        FieldAccessorFactory.getAccessor(
-                TupleTypeInfo.getBasicTupleTypeInfo(Integer.class, 
Integer.class), 2, null);
+    @Test
+    void testIllegalTupleField() {
+        assertThatThrownBy(
+                        () ->
+                                FieldAccessorFactory.getAccessor(
+                                        TupleTypeInfo.getBasicTupleTypeInfo(
+                                                Integer.class, Integer.class),
+                                        2,
+                                        null))
+                
.isInstanceOf(CompositeType.InvalidFieldReferenceException.class);
     }
 
     /** POJO. */
@@ -181,52 +188,56 @@ public class FieldAccessorTest {
     }
 
     @Test
-    public void testTupleInPojoInTuple() {
+    void testTupleInPojoInTuple() {
         Tuple2<String, Foo> t = Tuple2.of("aa", new Foo(8, Tuple2.of("ddd", 
9L), (short) 2));
         TupleTypeInfo<Tuple2<String, Foo>> tpeInfo =
                 (TupleTypeInfo<Tuple2<String, Foo>>) 
TypeExtractor.getForObject(t);
 
         FieldAccessor<Tuple2<String, Foo>, Long> f1tf1 =
                 FieldAccessorFactory.getAccessor(tpeInfo, "f1.t.f1", null);
-        assertEquals(Long.class, f1tf1.getFieldType().getTypeClass());
-        assertEquals(9L, (long) f1tf1.get(t));
-        assertEquals(9L, (long) t.f1.t.f1);
+        assertThat(f1tf1.getFieldType().getTypeClass()).isEqualTo(Long.class);
+        assertThat(f1tf1.get(t)).isEqualTo(9L);
+        assertThat(t.f1.t.f1).isEqualTo(9L);
         t = f1tf1.set(t, 12L);
-        assertEquals(12L, (long) f1tf1.get(t));
-        assertEquals(12L, (long) t.f1.t.f1);
+        assertThat(f1tf1.get(t)).isEqualTo(12L);
+        assertThat(t.f1.t.f1).isEqualTo(12L);
 
         FieldAccessor<Tuple2<String, Foo>, String> f1tf0 =
                 FieldAccessorFactory.getAccessor(tpeInfo, "f1.t.f0", null);
-        assertEquals(String.class, f1tf0.getFieldType().getTypeClass());
-        assertEquals("ddd", f1tf0.get(t));
-        assertEquals("ddd", t.f1.t.f0);
+        
assertThat(f1tf0.getFieldType().getTypeClass()).isEqualTo(String.class);
+        assertThat(f1tf0.get(t)).isEqualTo("ddd");
+        assertThat(t.f1.t.f0).isEqualTo("ddd");
         t = f1tf0.set(t, "alma");
-        assertEquals("alma", f1tf0.get(t));
-        assertEquals("alma", t.f1.t.f0);
+        assertThat(f1tf0.get(t)).isEqualTo("alma");
+        assertThat(t.f1.t.f0).isEqualTo("alma");
 
         FieldAccessor<Tuple2<String, Foo>, Foo> f1 =
                 FieldAccessorFactory.getAccessor(tpeInfo, "f1", null);
         FieldAccessor<Tuple2<String, Foo>, Foo> f1n =
                 FieldAccessorFactory.getAccessor(tpeInfo, 1, null);
-        assertEquals(Foo.class, f1.getFieldType().getTypeClass());
-        assertEquals(Foo.class, f1n.getFieldType().getTypeClass());
-        assertEquals(Tuple2.of("alma", 12L), f1.get(t).t);
-        assertEquals(Tuple2.of("alma", 12L), f1n.get(t).t);
-        assertEquals(Tuple2.of("alma", 12L), t.f1.t);
+        assertThat(f1.getFieldType().getTypeClass()).isEqualTo(Foo.class);
+        assertThat(f1n.getFieldType().getTypeClass()).isEqualTo(Foo.class);
+        assertThat(f1.get(t).t).isEqualTo(Tuple2.of("alma", 12L));
+        assertThat(f1n.get(t).t).isEqualTo(Tuple2.of("alma", 12L));
+        assertThat(t.f1.t).isEqualTo(Tuple2.of("alma", 12L));
         Foo newFoo = new Foo(8, Tuple2.of("ddd", 9L), (short) 2);
         f1.set(t, newFoo);
-        assertEquals(newFoo, f1.get(t));
-        assertEquals(newFoo, f1n.get(t));
-        assertEquals(newFoo, t.f1);
+        assertThat(f1.get(t)).isEqualTo(newFoo);
+        assertThat(f1n.get(t)).isEqualTo(newFoo);
+        assertThat(t.f1).isEqualTo(newFoo);
     }
 
-    @Test(expected = CompositeType.InvalidFieldReferenceException.class)
-    public void testIllegalTupleInPojoInTuple() {
+    @Test
+    void testIllegalTupleInPojoInTuple() {
         Tuple2<String, Foo> t = Tuple2.of("aa", new Foo(8, Tuple2.of("ddd", 
9L), (short) 2));
         TupleTypeInfo<Tuple2<String, Foo>> tpeInfo =
                 (TupleTypeInfo<Tuple2<String, Foo>>) 
TypeExtractor.getForObject(t);
 
-        FieldAccessorFactory.getAccessor(tpeInfo, "illegal.illegal.illegal", 
null);
+        assertThatThrownBy(
+                        () ->
+                                FieldAccessorFactory.getAccessor(
+                                        tpeInfo, "illegal.illegal.illegal", 
null))
+                
.isInstanceOf(CompositeType.InvalidFieldReferenceException.class);
     }
 
     /** POJO for testing field access. */
@@ -272,32 +283,32 @@ public class FieldAccessorTest {
     }
 
     @Test
-    public void testPojoInPojo() {
+    void testPojoInPojo() {
         Outer o = new Outer(10, new Inner(4L), (short) 12);
         PojoTypeInfo<Outer> tpeInfo = (PojoTypeInfo<Outer>) 
TypeInformation.of(Outer.class);
 
         FieldAccessor<Outer, Long> fix = 
FieldAccessorFactory.getAccessor(tpeInfo, "i.x", null);
-        assertEquals(Long.class, fix.getFieldType().getTypeClass());
-        assertEquals(4L, (long) fix.get(o));
-        assertEquals(4L, o.i.x);
+        assertThat(fix.getFieldType().getTypeClass()).isEqualTo(Long.class);
+        assertThat(fix.get(o)).isEqualTo(4L);
+        assertThat(o.i.x).isEqualTo(4L);
         o = fix.set(o, 22L);
-        assertEquals(22L, (long) fix.get(o));
-        assertEquals(22L, o.i.x);
+        assertThat(fix.get(o)).isEqualTo(22L);
+        assertThat(o.i.x).isEqualTo(22L);
 
         FieldAccessor<Outer, Inner> fi = 
FieldAccessorFactory.getAccessor(tpeInfo, "i", null);
-        assertEquals(Inner.class, fi.getFieldType().getTypeClass());
-        assertEquals(22L, fi.get(o).x);
-        assertEquals(22L, (long) fix.get(o));
-        assertEquals(22L, o.i.x);
+        assertThat(fi.getFieldType().getTypeClass()).isEqualTo(Inner.class);
+        assertThat(fi.get(o).x).isEqualTo(22L);
+        assertThat(fix.get(o)).isEqualTo(22L);
+        assertThat(o.i.x).isEqualTo(22L);
         o = fi.set(o, new Inner(30L));
-        assertEquals(30L, fi.get(o).x);
-        assertEquals(30L, (long) fix.get(o));
-        assertEquals(30L, o.i.x);
+        assertThat(fi.get(o).x).isEqualTo(30L);
+        assertThat(fix.get(o)).isEqualTo(30L);
+        assertThat(o.i.x).isEqualTo(30L);
     }
 
     @Test
     @SuppressWarnings("unchecked")
-    public void testArray() {
+    void testArray() {
         int[] a = new int[] {3, 5};
         FieldAccessor<int[], Integer> fieldAccessor =
                 (FieldAccessor<int[], Integer>)
@@ -305,12 +316,12 @@ public class FieldAccessorTest {
                                 FieldAccessorFactory.getAccessor(
                                         
PrimitiveArrayTypeInfo.getInfoFor(a.getClass()), 1, null);
 
-        assertEquals(Integer.class, 
fieldAccessor.getFieldType().getTypeClass());
+        
assertThat(fieldAccessor.getFieldType().getTypeClass()).isEqualTo(Integer.class);
 
-        assertEquals((Integer) a[1], fieldAccessor.get(a));
+        assertThat(fieldAccessor.get(a)).isEqualTo(a[1]);
 
         a = fieldAccessor.set(a, 6);
-        assertEquals((Integer) a[1], fieldAccessor.get(a));
+        assertThat(fieldAccessor.get(a)).isEqualTo(a[1]);
 
         Integer[] b = new Integer[] {3, 5};
         FieldAccessor<Integer[], Integer> fieldAccessor2 =
@@ -319,12 +330,12 @@ public class FieldAccessorTest {
                                 FieldAccessorFactory.getAccessor(
                                         
BasicArrayTypeInfo.getInfoFor(b.getClass()), 1, null);
 
-        assertEquals(Integer.class, 
fieldAccessor2.getFieldType().getTypeClass());
+        
assertThat(fieldAccessor2.getFieldType().getTypeClass()).isEqualTo(Integer.class);
 
-        assertEquals(b[1], fieldAccessor2.get(b));
+        assertThat(fieldAccessor2.get(b)).isEqualTo(b[1]);
 
         b = fieldAccessor2.set(b, 6);
-        assertEquals(b[1], fieldAccessor2.get(b));
+        assertThat(fieldAccessor2.get(b)).isEqualTo(b[1]);
     }
 
     /** POJO with array. */
@@ -343,60 +354,60 @@ public class FieldAccessorTest {
     }
 
     @Test
-    public void testArrayInPojo() {
+    void testArrayInPojo() {
         ArrayInPojo o = new ArrayInPojo(10L, new int[] {3, 4, 5}, 12);
         PojoTypeInfo<ArrayInPojo> tpeInfo =
                 (PojoTypeInfo<ArrayInPojo>) 
TypeInformation.of(ArrayInPojo.class);
 
         FieldAccessor<ArrayInPojo, Integer> fix =
                 FieldAccessorFactory.getAccessor(tpeInfo, "arr.1", null);
-        assertEquals(Integer.class, fix.getFieldType().getTypeClass());
-        assertEquals(4, (int) fix.get(o));
-        assertEquals(4L, o.arr[1]);
+        assertThat(fix.getFieldType().getTypeClass()).isEqualTo(Integer.class);
+        assertThat(fix.get(o)).isEqualTo(4);
+        assertThat(o.arr[1]).isEqualTo(4L);
         o = fix.set(o, 8);
-        assertEquals(8, (int) fix.get(o));
-        assertEquals(8, o.arr[1]);
+        assertThat(fix.get(o)).isEqualTo(8);
+        assertThat(o.arr[1]).isEqualTo(8);
     }
 
     @Test
-    public void testBasicType() {
+    void testBasicType() {
         Long x = 7L;
         TypeInformation<Long> tpeInfo = BasicTypeInfo.LONG_TYPE_INFO;
 
         FieldAccessor<Long, Long> f = 
FieldAccessorFactory.getAccessor(tpeInfo, 0, null);
-        assertEquals(Long.class, f.getFieldType().getTypeClass());
-        assertEquals(7L, (long) f.get(x));
+        assertThat(f.getFieldType().getTypeClass()).isEqualTo(Long.class);
+        assertThat(f.get(x)).isEqualTo(7L);
         x = f.set(x, 12L);
-        assertEquals(12L, (long) f.get(x));
-        assertEquals(12L, (long) x);
+        assertThat(f.get(x)).isEqualTo(12L);
+        assertThat(x).isEqualTo(12L);
 
         FieldAccessor<Long, Long> f2 = 
FieldAccessorFactory.getAccessor(tpeInfo, "*", null);
-        assertEquals(Long.class, f2.getFieldType().getTypeClass());
-        assertEquals(12L, (long) f2.get(x));
+        assertThat(f2.getFieldType().getTypeClass()).isEqualTo(Long.class);
+        assertThat(f2.get(x)).isEqualTo(12L);
         x = f2.set(x, 14L);
-        assertEquals(14L, (long) f2.get(x));
-        assertEquals(14L, (long) x);
+        assertThat(f2.get(x)).isEqualTo(14L);
+        assertThat(x).isEqualTo(14L);
     }
 
-    @Test(expected = IllegalArgumentException.class)
-    public void testIllegalBasicType1() {
-        Long x = 7L;
+    @Test
+    void testIllegalBasicType1() {
         TypeInformation<Long> tpeInfo = BasicTypeInfo.LONG_TYPE_INFO;
 
-        FieldAccessor<Long, Long> f = 
FieldAccessorFactory.getAccessor(tpeInfo, 1, null);
+        assertThatThrownBy(() -> FieldAccessorFactory.getAccessor(tpeInfo, 1, 
null))
+                .isInstanceOf(IllegalArgumentException.class);
     }
 
-    @Test(expected = IllegalArgumentException.class)
-    public void testIllegalBasicType2() {
-        Long x = 7L;
+    @Test
+    void testIllegalBasicType2() {
         TypeInformation<Long> tpeInfo = BasicTypeInfo.LONG_TYPE_INFO;
 
-        FieldAccessor<Long, Long> f = 
FieldAccessorFactory.getAccessor(tpeInfo, "foo", null);
+        assertThatThrownBy(() -> FieldAccessorFactory.getAccessor(tpeInfo, 
"foo", null))
+                .isInstanceOf(IllegalArgumentException.class);
     }
 
     /** Validates that no ClassCastException happens should not fail e.g. like 
in FLINK-8255. */
-    @Test(expected = CompositeType.InvalidFieldReferenceException.class)
-    public void testRowTypeInfo() {
+    @Test
+    void testRowTypeInfo() {
         TypeInformation<?>[] typeList =
                 new TypeInformation<?>[] {
                     new RowTypeInfo(BasicTypeInfo.SHORT_TYPE_INFO, 
BasicTypeInfo.BIG_DEC_TYPE_INFO)
@@ -405,6 +416,7 @@ public class FieldAccessorTest {
         String[] fieldNames = new String[] {"row"};
         RowTypeInfo rowTypeInfo = new RowTypeInfo(typeList, fieldNames);
 
-        FieldAccessor f = FieldAccessorFactory.getAccessor(rowTypeInfo, 
"row.0", null);
+        assertThatThrownBy(() -> FieldAccessorFactory.getAccessor(rowTypeInfo, 
"row.0", null))
+                
.isInstanceOf(CompositeType.InvalidFieldReferenceException.class);
     }
 }

Reply via email to