diff --git a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogDelegateStateTest.java b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogDelegateStateTest.java index 40b07d8cf8877f..46ca34fac6b6e0 100644 --- a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogDelegateStateTest.java +++ b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogDelegateStateTest.java @@ -32,13 +32,15 @@ import org.apache.flink.runtime.state.VoidNamespaceSerializer; import org.apache.flink.runtime.state.hashmap.HashMapStateBackend; import org.apache.flink.state.rocksdb.EmbeddedRocksDBStateBackend; +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.IOUtils; -import org.junit.After; -import org.junit.Before; -import org.junit.Test; -import org.junit.runner.RunWith; -import org.junit.runners.Parameterized; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.TestTemplate; +import org.junit.jupiter.api.extension.ExtendWith; import java.util.Arrays; import java.util.List; @@ -46,60 +48,60 @@ import static org.apache.flink.state.changelog.ChangelogStateBackendTestUtils.DummyCheckpointingStorageAccess; import static org.apache.flink.state.changelog.ChangelogStateBackendTestUtils.createKeyedBackend; -import static org.junit.Assert.assertSame; +import static org.assertj.core.api.Assertions.assertThat; /** Tests for {@link ChangelogStateBackend} delegating state accesses. */ -@RunWith(Parameterized.class) -public class ChangelogDelegateStateTest { +@ExtendWith(ParameterizedTestExtension.class) +class ChangelogDelegateStateTest { private MockEnvironment env; - @Parameterized.Parameters + @Parameters public static List> delegatedStateBackend() { return Arrays.asList(HashMapStateBackend::new, EmbeddedRocksDBStateBackend::new); } - @Parameterized.Parameter public Supplier backend; + @Parameter public Supplier backend; - @Before - public void before() { + @BeforeEach + void before() { env = MockEnvironment.builder().build(); env.setCheckpointStorageAccess(new DummyCheckpointingStorageAccess()); } - @After - public void after() { + @AfterEach + void after() { IOUtils.closeQuietly(env); } - @Test - public void testDelegatingValueState() throws Exception { + @TestTemplate + void testDelegatingValueState() throws Exception { testDelegatingState( new ValueStateDescriptor<>("id", String.class), ChangelogValueState.class); } - @Test - public void testDelegatingListState() throws Exception { + @TestTemplate + void testDelegatingListState() throws Exception { testDelegatingState( new ListStateDescriptor<>("id", String.class), ChangelogListState.class); } - @Test - public void testDelegatingMapState() throws Exception { + @TestTemplate + void testDelegatingMapState() throws Exception { testDelegatingState( new MapStateDescriptor<>("id", Integer.class, String.class), ChangelogMapState.class); } - @Test - public void testDelegatingReducingState() throws Exception { + @TestTemplate + void testDelegatingReducingState() throws Exception { testDelegatingState( new ReducingStateDescriptor<>( "id", (value1, value2) -> value1 + "," + value2, String.class), ChangelogReducingState.class); } - @Test - public void testDelegatingAggregatingState() throws Exception { + @TestTemplate + void testDelegatingAggregatingState() throws Exception { testDelegatingState( new AggregatingStateDescriptor<>( "my-state", @@ -121,15 +123,18 @@ private void testDelegatingState(StateDescriptor descriptor, Class stateClass changelogBackend.getPartitionedState( VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, descriptor); - assertSame(state.getClass(), stateClass); - assertSame( - ((AbstractChangelogState) state).getDelegatedState().getClass(), - delegatedBackend - .getPartitionedState( - VoidNamespace.INSTANCE, - VoidNamespaceSerializer.INSTANCE, - descriptor) - .getClass()); + assertThat(state.getClass()).isSameAs(stateClass); + assertThat( + ((AbstractChangelogState) state) + .getDelegatedState() + .getClass()) + .isSameAs( + delegatedBackend + .getPartitionedState( + VoidNamespace.INSTANCE, + VoidNamespaceSerializer.INSTANCE, + descriptor) + .getClass()); } finally { if (delegatedBackend != null) { delegatedBackend.dispose(); diff --git a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogKeyedStateBackendTest.java b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogKeyedStateBackendTest.java index 10a13a9b3d4762..a768f149b45ccc 100644 --- a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogKeyedStateBackendTest.java +++ b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogKeyedStateBackendTest.java @@ -41,25 +41,25 @@ import org.apache.flink.state.changelog.ChangelogStateBackendTestUtils.DummyCheckpointingStorageAccess; import org.apache.flink.state.common.PeriodicMaterializationManager.MaterializationRunnable; -import org.junit.Test; -import org.junit.runner.RunWith; -import org.junit.runners.Parameterized; -import org.junit.runners.Parameterized.Parameter; +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.jupiter.api.TestTemplate; +import org.junit.jupiter.api.extension.ExtendWith; import java.io.IOException; import java.util.Optional; import java.util.concurrent.RunnableFuture; import static java.util.Collections.emptyList; -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; /** {@link ChangelogKeyedStateBackend} test. */ -@RunWith(Parameterized.class) -public class ChangelogKeyedStateBackendTest { +@ExtendWith(ParameterizedTestExtension.class) +class ChangelogKeyedStateBackendTest { - @Parameterized.Parameters(name = "checkpointID={0}, materializationId={1}") + @Parameters(name = "checkpointID={0}, materializationId={1}") public static Object[][] parameters() { return new Object[][] { {0L, 200L}, @@ -73,8 +73,8 @@ public static Object[][] parameters() { @Parameter(1) public long materializationId; - @Test - public void testCheckpointConfirmation() throws Exception { + @TestTemplate + void testCheckpointConfirmation() throws Exception { MockKeyedStateBackend mock = createMock(); ChangelogKeyedStateBackend changelog = createChangelog(mock); try { @@ -83,7 +83,7 @@ public void testCheckpointConfirmation() throws Exception { checkpoint(changelog, checkpointId).get().discardState(); changelog.notifyCheckpointComplete(checkpointId); - assertEquals(materializationId, mock.getLastCompletedCheckpointID()); + assertThat(mock.getLastCompletedCheckpointID()).isEqualTo(materializationId); } finally { changelog.close(); @@ -91,8 +91,8 @@ public void testCheckpointConfirmation() throws Exception { } } - @Test - public void testInitMaterialization() throws Exception { + @TestTemplate + void testInitMaterialization() throws Exception { MockKeyedStateBackend delegatedBackend = createMock(); ChangelogKeyedStateBackend backend = createChangelog(delegatedBackend); @@ -103,12 +103,12 @@ public void testInitMaterialization() throws Exception { runnable = backend.initMaterialization(); // 1. should trigger first materialization - assertTrue("first materialization should be trigger.", runnable.isPresent()); + assertThat(runnable).as("first materialization should be trigger.").isPresent(); appendMockStateChange(backend); // ensure there is non-materialized changelog // 2. should not trigger new one until the previous one has been confirmed or failed - assertFalse(backend.initMaterialization().isPresent()); + assertThat(backend.initMaterialization()).isNotPresent(); backend.handleMaterializationFailureOrCancellation( runnable.get().getMaterializationID(), @@ -116,12 +116,12 @@ public void testInitMaterialization() throws Exception { null); runnable = backend.initMaterialization(); // 3. should trigger new one after previous one failed - assertTrue(runnable.isPresent()); + assertThat(runnable).isPresent(); appendMockStateChange(backend); // ensure there is non-materialized changelog // 4. should not trigger new one until the previous one has been confirmed or failed - assertFalse(backend.initMaterialization().isPresent()); + assertThat(backend.initMaterialization()).isNotPresent(); backend.handleMaterializationResult( SnapshotResult.empty(), @@ -130,7 +130,7 @@ public void testInitMaterialization() throws Exception { checkpoint(backend, checkpointId).get().discardState(); backend.notifyCheckpointComplete(checkpointId); // 5. should trigger new one after previous one has been confirmed - assertTrue(backend.initMaterialization().isPresent()); + assertThat(backend.initMaterialization()).isPresent(); } finally { backend.close(); backend.dispose(); diff --git a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogListStateTest.java b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogListStateTest.java index fb09247435d7f9..bdb3ca7bd5526e 100644 --- a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogListStateTest.java +++ b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogListStateTest.java @@ -27,7 +27,7 @@ import org.apache.flink.util.function.FunctionWithException; import org.apache.flink.util.function.ThrowingConsumer; -import org.junit.Test; +import org.junit.jupiter.api.Test; import java.util.ArrayList; import java.util.Arrays; @@ -38,50 +38,48 @@ import static java.util.Collections.emptyList; import static java.util.Collections.singletonList; -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; /** ChangelogListState Test. */ @SuppressWarnings({"rawtypes", "unchecked"}) -public class ChangelogListStateTest { +class ChangelogListStateTest { @Test - public void testValuesIterator() throws Exception { + void testValuesIterator() throws Exception { testIterator(singletonList("value"), state -> state.get().iterator(), "value"); } @Test - public void testPutRecorded() throws Exception { + void testPutRecorded() throws Exception { testRecorded( emptyList(), state -> state.add("x"), - logger -> assertTrue(logger.stateElementAdded)); + logger -> assertThat(logger.stateElementAdded).isTrue()); } @Test - public void testAddAllRecorded() throws Exception { + void testAddAllRecorded() throws Exception { List list = Arrays.asList("a", "b", "c"); testRecorded( emptyList(), state -> state.addAll(list), - logger -> assertEquals(list, logger.state)); + logger -> assertThat(logger.state).isEqualTo(list)); } @Test - public void testGetNotRecorded() throws Exception { + void testGetNotRecorded() throws Exception { testRecorded( singletonList("x"), ChangelogListState::get, - logger -> assertFalse(logger.anythingChanged())); + logger -> assertThat(logger.anythingChanged()).isFalse()); } @Test - public void testClearRecorded() throws Exception { + void testClearRecorded() throws Exception { testRecorded( singletonList("x"), ChangelogListState::clear, - logger -> assertTrue(logger.stateCleared)); + logger -> assertThat(logger.stateCleared).isTrue()); } private void testIterator( @@ -94,15 +92,15 @@ private void testIterator( Iterator iterator = iteratorSupplier.apply(state); for (T el : elements) { - assertTrue(iterator.hasNext()); - assertEquals(el, iterator.next()); + assertThat(iterator.hasNext()).isTrue(); + assertThat(iterator.next()).isEqualTo(el); iterator.remove(); } - assertFalse(iterator.hasNext()); - assertTrue(state.getInternal().isEmpty()); + assertThat(iterator.hasNext()).isFalse(); + assertThat(state.getInternal().isEmpty()).isTrue(); // changes to the rocksdb list iterator are not propagated back - expect the same here - assertFalse(logger.stateElementRemoved); + assertThat(logger.stateElementRemoved).isFalse(); } private void testRecorded( diff --git a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogMapStateTest.java b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogMapStateTest.java index d17bd3df19367e..56f1f9e46c30c0 100644 --- a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogMapStateTest.java +++ b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogMapStateTest.java @@ -27,7 +27,7 @@ import org.apache.flink.util.function.FunctionWithException; import org.apache.flink.util.function.ThrowingConsumer; -import org.junit.Test; +import org.junit.jupiter.api.Test; import java.util.HashMap; import java.util.Iterator; @@ -36,33 +36,31 @@ import static java.util.Collections.emptyMap; import static java.util.Collections.singletonMap; -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; /** ChangelogMapState Test. */ @SuppressWarnings({"rawtypes", "unchecked"}) -public class ChangelogMapStateTest { +class ChangelogMapStateTest { @Test - public void testValuesIterator() throws Exception { + void testValuesIterator() throws Exception { testIterator(singletonMap("key", "value"), state -> state.values().iterator(), "value"); } @Test - public void testKeysIterator() throws Exception { + void testKeysIterator() throws Exception { testIterator(singletonMap("key", "value"), state -> state.keys().iterator(), "key"); } @Test - public void testEntriesIterator() throws Exception { + void testEntriesIterator() throws Exception { Map map = singletonMap("key", "value"); Map.Entry entry = map.entrySet().iterator().next(); testIterator(map, state -> state.entries().iterator(), entry); } @Test - public void testEntryUpdateRecorded() throws Exception { + void testEntryUpdateRecorded() throws Exception { testRecorded( singletonMap("x", "y"), state -> @@ -70,46 +68,48 @@ public void testEntryUpdateRecorded() throws Exception { .iterator() .next() .setValue("z"), - logger -> assertTrue(logger.stateElementChanged)); + logger -> assertThat(logger.stateElementChanged).isTrue()); } @Test - public void testPutRecorded() throws Exception { + void testPutRecorded() throws Exception { testRecorded( emptyMap(), state -> state.put("x", "y"), - logger -> assertTrue(logger.stateElementChanged)); + logger -> assertThat(logger.stateElementChanged).isTrue()); } @Test - public void testPutAllRecorded() throws Exception { + void testPutAllRecorded() throws Exception { Map map = singletonMap("x", "y"); testRecorded( - emptyMap(), state -> state.putAll(map), logger -> assertEquals(map, logger.state)); + emptyMap(), + state -> state.putAll(map), + logger -> assertThat(logger.state).isEqualTo(map)); } @Test - public void testRemoveRecorded() throws Exception { + void testRemoveRecorded() throws Exception { testRecorded( singletonMap("x", "y"), state -> state.remove("x"), - logger -> assertTrue(logger.stateElementRemoved)); + logger -> assertThat(logger.stateElementRemoved).isTrue()); } @Test - public void testGetNotRecorded() throws Exception { + void testGetNotRecorded() throws Exception { testRecorded( singletonMap("x", "y"), state -> state.get("x"), - logger -> assertFalse(logger.anythingChanged())); + logger -> assertThat(logger.anythingChanged()).isFalse()); } @Test - public void testClearRecorded() throws Exception { + void testClearRecorded() throws Exception { testRecorded( singletonMap("x", "y"), ChangelogMapState::clear, - logger -> assertTrue(logger.stateCleared)); + logger -> assertThat(logger.stateCleared).isTrue()); } private void testIterator( @@ -122,14 +122,14 @@ private void testIterator( Iterator iterator = iteratorSupplier.apply(state); for (T el : elements) { - assertTrue(iterator.hasNext()); - assertEquals(el, iterator.next()); + assertThat(iterator.hasNext()).isTrue(); + assertThat(iterator.next()).isEqualTo(el); iterator.remove(); } - assertFalse(iterator.hasNext()); - assertTrue(state.isEmpty()); - assertTrue(logger.stateElementRemoved); + assertThat(iterator.hasNext()).isFalse(); + assertThat(state.isEmpty()).isTrue(); + assertThat(logger.stateElementRemoved).isTrue(); } private void testRecorded( diff --git a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogMetricGroupTest.java b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogMetricGroupTest.java index 6bfe146d701933..98661af0521c54 100644 --- a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogMetricGroupTest.java +++ b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogMetricGroupTest.java @@ -41,7 +41,7 @@ import org.apache.flink.state.common.PeriodicMaterializationManager; import org.apache.flink.util.Preconditions; -import org.junit.Test; +import org.junit.jupiter.api.Test; import java.io.IOException; import java.util.HashMap; @@ -58,14 +58,13 @@ import static org.apache.flink.state.common.ChangelogMaterializationMetricGroup.FAILED_MATERIALIZATION; import static org.apache.flink.state.common.ChangelogMaterializationMetricGroup.LAST_DURATION_OF_MATERIALIZATION; import static org.apache.flink.state.common.ChangelogMaterializationMetricGroup.STARTED_MATERIALIZATION; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotEquals; +import static org.assertj.core.api.Assertions.assertThat; /** * Test the {@link MetricGroup} Changelog used. e.g.{@link ChangelogStateBackendMetricGroup}, {@link * ChangelogMaterializationMetricGroup} */ -public class ChangelogMetricGroupTest { +class ChangelogMetricGroupTest { private ChangelogKeyedStateBackend changelogKeyedStateBackend; private PeriodicMaterializationManager periodicMaterializationManager; @@ -81,69 +80,71 @@ public class ChangelogMetricGroupTest { private Gauge lastIncSizeOfNonMaterializationGauge; @Test - public void testCompletedMaterialization() throws Exception { + void testCompletedMaterialization() throws Exception { setup(snapshotResult -> snapshotResult); // The materialization will be skipped if no data updated. - assertEquals(-1L, lastDurationOfMaterializationGauge.getValue().longValue()); + assertThat(lastDurationOfMaterializationGauge.getValue()).isEqualTo(-1L); periodicMaterializationManager.triggerMaterialization(); runSnapshot(1L); - assertEquals(1L, startedMaterializationCounter.getCount()); - assertEquals(1L, completedMaterializationCounter.getCount()); - assertNotEquals(-1L, lastDurationOfMaterializationGauge.getValue().longValue()); - assertEquals(0L, lastFullSizeOfMaterializationGauge.getValue().longValue()); - assertEquals(0L, lastIncSizeOfMaterializationGauge.getValue().longValue()); - assertEquals(0L, lastFullSizeOfNonMaterializationGauge.getValue().longValue()); - assertEquals(0L, lastIncSizeOfNonMaterializationGauge.getValue().longValue()); + assertThat(startedMaterializationCounter.getCount()).isEqualTo(1L); + assertThat(completedMaterializationCounter.getCount()).isEqualTo(1L); + assertThat(lastDurationOfMaterializationGauge.getValue()).isNotEqualTo(-1L); + assertThat(lastFullSizeOfMaterializationGauge.getValue()).isEqualTo(0L); + assertThat(lastIncSizeOfMaterializationGauge.getValue()).isEqualTo(0L); + assertThat(lastFullSizeOfNonMaterializationGauge.getValue()).isEqualTo(0L); + assertThat(lastIncSizeOfNonMaterializationGauge.getValue()).isEqualTo(0L); changelogKeyedStateBackend.setCurrentKey(1); state.update(1); periodicMaterializationManager.triggerMaterialization(); runSnapshot(2L); - assertEquals(2L, startedMaterializationCounter.getCount()); - assertEquals(2L, completedMaterializationCounter.getCount()); + assertThat(startedMaterializationCounter.getCount()).isEqualTo(2L); + assertThat(completedMaterializationCounter.getCount()).isEqualTo(2L); Long lastFullSizeOfMaterialization = lastFullSizeOfMaterializationGauge.getValue(); Long lastIncSizeOfMaterialization = lastIncSizeOfMaterializationGauge.getValue(); Long lastFullSizeOfNonMaterialization = lastFullSizeOfNonMaterializationGauge.getValue(); Long lastIncSizeOfNonMaterialization = lastIncSizeOfNonMaterializationGauge.getValue(); - assertNotEquals(0L, lastFullSizeOfMaterialization.longValue()); - assertNotEquals(0L, lastIncSizeOfMaterialization.longValue()); - assertNotEquals(-1L, lastDurationOfMaterializationGauge.getValue().longValue()); + assertThat(lastFullSizeOfMaterialization).isNotEqualTo(0L); + assertThat(lastIncSizeOfMaterialization).isNotEqualTo(0L); + assertThat(lastDurationOfMaterializationGauge.getValue()).isNotEqualTo(-1L); // The non-materialization size will be zero if no data updated between completed // materialization and checkpoint. - assertEquals(0L, lastFullSizeOfNonMaterialization.longValue()); - assertEquals(0L, lastIncSizeOfNonMaterialization.longValue()); + assertThat(lastFullSizeOfNonMaterialization).isEqualTo(0L); + assertThat(lastIncSizeOfNonMaterialization).isEqualTo(0L); changelogKeyedStateBackend.setCurrentKey(2); state.update(2); runSnapshot(3L); // The materialization size will not be updated if no materialization triggered. - assertEquals(lastFullSizeOfMaterialization, lastFullSizeOfMaterializationGauge.getValue()); - assertEquals(lastIncSizeOfMaterialization, lastIncSizeOfMaterializationGauge.getValue()); - assertNotEquals( - lastFullSizeOfNonMaterialization, lastFullSizeOfNonMaterializationGauge.getValue()); - assertNotEquals( - lastIncSizeOfNonMaterialization, lastIncSizeOfNonMaterializationGauge.getValue()); - - assertEquals(0L, failedMaterializationCounter.getCount()); + assertThat(lastFullSizeOfMaterializationGauge.getValue()) + .isEqualTo(lastFullSizeOfMaterialization); + assertThat(lastIncSizeOfMaterializationGauge.getValue()) + .isEqualTo(lastIncSizeOfMaterialization); + assertThat(lastFullSizeOfNonMaterializationGauge.getValue()) + .isNotEqualTo(lastFullSizeOfNonMaterialization); + assertThat(lastIncSizeOfNonMaterializationGauge.getValue()) + .isNotEqualTo(lastIncSizeOfNonMaterialization); + + assertThat(failedMaterializationCounter.getCount()).isEqualTo(0L); } @Test - public void testFailedMaterialization() throws Exception { + void testFailedMaterialization() throws Exception { setup(snapshotResult -> ExceptionallyDoneFuture.of(new RuntimeException())); changelogKeyedStateBackend.setCurrentKey(1); state.update(1); - assertEquals(-1L, lastDurationOfMaterializationGauge.getValue().longValue()); + assertThat(lastDurationOfMaterializationGauge.getValue()).isEqualTo(-1L); periodicMaterializationManager.triggerMaterialization(); runSnapshot(1L); - assertEquals(0L, completedMaterializationCounter.getCount()); - assertEquals(1L, failedMaterializationCounter.getCount()); - assertEquals(1L, startedMaterializationCounter.getCount()); - assertEquals(-1L, lastDurationOfMaterializationGauge.getValue().longValue()); - assertEquals(0L, lastFullSizeOfMaterializationGauge.getValue().longValue()); - assertEquals(0L, lastIncSizeOfMaterializationGauge.getValue().longValue()); - assertNotEquals(0L, lastFullSizeOfNonMaterializationGauge.getValue().longValue()); - assertNotEquals(0L, lastIncSizeOfNonMaterializationGauge.getValue().longValue()); + assertThat(completedMaterializationCounter.getCount()).isEqualTo(0L); + assertThat(failedMaterializationCounter.getCount()).isEqualTo(1L); + assertThat(startedMaterializationCounter.getCount()).isEqualTo(1L); + assertThat(lastDurationOfMaterializationGauge.getValue()).isEqualTo(-1L); + assertThat(lastFullSizeOfMaterializationGauge.getValue()).isEqualTo(0L); + assertThat(lastIncSizeOfMaterializationGauge.getValue()).isEqualTo(0L); + assertThat(lastFullSizeOfNonMaterializationGauge.getValue()).isNotEqualTo(0L); + assertThat(lastIncSizeOfNonMaterializationGauge.getValue()).isNotEqualTo(0L); } @SuppressWarnings("unchecked") diff --git a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogPqStateTest.java b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogPqStateTest.java index 96203e18915748..3e1ce162028d2a 100644 --- a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogPqStateTest.java +++ b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogPqStateTest.java @@ -24,7 +24,7 @@ import org.apache.flink.util.function.FunctionWithException; import org.apache.flink.util.function.ThrowingConsumer; -import org.junit.Test; +import org.junit.jupiter.api.Test; import javax.annotation.Nonnull; import javax.annotation.Nullable; @@ -40,57 +40,55 @@ import static java.util.Collections.emptyList; import static java.util.Collections.singletonList; -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; /** ChangelogKeyGroupedPriorityQueue Test. */ @SuppressWarnings({"rawtypes", "unchecked"}) -public class ChangelogPqStateTest { +class ChangelogPqStateTest { @Test - public void testValuesIterator() throws Exception { + void testValuesIterator() throws Exception { testIterator(singletonList("value"), ChangelogKeyGroupedPriorityQueue::iterator, "value"); } @Test - public void testPutRecorded() throws Exception { + void testPutRecorded() throws Exception { testRecorded( emptyList(), state -> state.add("x"), - logger -> assertTrue(logger.stateElementAdded)); + logger -> assertThat(logger.stateElementAdded).isTrue()); } @Test - public void testPollRecorded() throws Exception { + void testPollRecorded() throws Exception { testRecorded( singletonList("x"), ChangelogKeyGroupedPriorityQueue::poll, - logger -> assertTrue(logger.stateElementRemoved)); + logger -> assertThat(logger.stateElementRemoved).isTrue()); } @Test - public void testRemoveRecorded() throws Exception { + void testRemoveRecorded() throws Exception { testRecorded( singletonList("x"), state -> state.remove("x"), - logger -> assertTrue(logger.stateElementRemoved)); + logger -> assertThat(logger.stateElementRemoved).isTrue()); } @Test - public void testAddAllRecorded() throws Exception { + void testAddAllRecorded() throws Exception { testRecorded( emptyList(), state -> state.addAll(singletonList("x")), - logger -> assertTrue(logger.stateElementAdded)); + logger -> assertThat(logger.stateElementAdded).isTrue()); } @Test - public void testGetNotRecorded() throws Exception { + void testGetNotRecorded() throws Exception { testRecorded( singletonList("x"), ChangelogKeyGroupedPriorityQueue::peek, - logger -> assertFalse(logger.anythingChanged())); + logger -> assertThat(logger.anythingChanged()).isFalse()); } private void testIterator( @@ -106,14 +104,14 @@ private void testIterator( Iterator iterator = iteratorSupplier.apply(state); for (T el : elements) { - assertTrue(iterator.hasNext()); - assertEquals(el, iterator.next()); + assertThat(iterator.hasNext()).isTrue(); + assertThat(iterator.next()).isEqualTo(el); iterator.remove(); } - assertFalse(iterator.hasNext()); - assertTrue(state.isEmpty()); - assertTrue(logger.stateElementRemoved); + assertThat(iterator.hasNext()).isFalse(); + assertThat(state.isEmpty()).isTrue(); + assertThat(logger.stateElementRemoved).isTrue(); } private void testRecorded( diff --git a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogStateBackendLoadingTest.java b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogStateBackendLoadingTest.java index b7a49fcbaba776..a3d76ee78658bc 100644 --- a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogStateBackendLoadingTest.java +++ b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogStateBackendLoadingTest.java @@ -49,37 +49,36 @@ import org.apache.flink.util.Collector; import org.apache.flink.util.TernaryBoolean; -import org.junit.Rule; -import org.junit.Test; -import org.junit.rules.TemporaryFolder; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import java.nio.file.Path; import java.util.Collections; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertSame; -import static org.junit.Assert.assertTrue; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Verify Changelog StateBackend is properly loaded. */ -public class ChangelogStateBackendLoadingTest { - @Rule public final TemporaryFolder tmp = new TemporaryFolder(); +class ChangelogStateBackendLoadingTest { + @TempDir Path tmp; private final ClassLoader cl = getClass().getClassLoader(); private final String backendKey = StateBackendOptions.STATE_BACKEND.key(); @Test - public void testLoadingDefault() throws Exception { + void testLoadingDefault() throws Exception { final StateBackend backend = StateBackendLoader.fromApplicationOrConfigOrDefault( null, config(), config(), cl, null); final CheckpointStorage storage = CheckpointStorageLoader.load(null, backend, config(), config(), cl, null); - assertTrue(backend instanceof HashMapStateBackend); + assertThat(backend).isInstanceOf(HashMapStateBackend.class); } @Test - public void testApplicationDefinedHasPrecedence() throws Exception { + void testApplicationDefinedHasPrecedence() throws Exception { final StateBackend appBackend = new MockStateBackend(); // "rocksdb" should not take effect final StateBackend backend = @@ -90,13 +89,16 @@ public void testApplicationDefinedHasPrecedence() throws Exception { assertDelegateStateBackend( backend, MockStateBackend.class, storage, MockStateBackend.class); - assertTrue( - ((MockStateBackend) (((ChangelogStateBackend) backend).getDelegatedStateBackend())) - .isConfigUpdated()); + assertThat( + ((MockStateBackend) + (((ChangelogStateBackend) backend) + .getDelegatedStateBackend())) + .isConfigUpdated()) + .isTrue(); } @Test - public void testApplicationDefinedChangelogStateBackend() throws Exception { + void testApplicationDefinedChangelogStateBackend() throws Exception { final StateBackend appBackend = new MockStateBackend(); // "rocksdb" should not take effect final StateBackend backend = @@ -107,13 +109,16 @@ public void testApplicationDefinedChangelogStateBackend() throws Exception { assertDelegateStateBackend( backend, MockStateBackend.class, storage, MockStateBackend.class); - assertTrue( - ((MockStateBackend) (((ChangelogStateBackend) backend).getDelegatedStateBackend())) - .isConfigUpdated()); + assertThat( + ((MockStateBackend) + (((ChangelogStateBackend) backend) + .getDelegatedStateBackend())) + .isConfigUpdated()) + .isTrue(); } @Test - public void testApplicationEnableChangelogStateBackend() throws Exception { + void testApplicationEnableChangelogStateBackend() throws Exception { final StateBackend backend = StateBackendLoader.fromApplicationOrConfigOrDefault( null, config(true), config(false), cl, null); @@ -125,16 +130,16 @@ public void testApplicationEnableChangelogStateBackend() throws Exception { } @Test - public void testApplicationDisableChangelogStateBackend() throws Exception { + void testApplicationDisableChangelogStateBackend() throws Exception { final StateBackend backend = StateBackendLoader.fromApplicationOrConfigOrDefault( null, config(false), config(true), cl, null); - assertTrue(backend instanceof HashMapStateBackend); + assertThat(backend).isInstanceOf(HashMapStateBackend.class); } @Test - public void testLoadingChangelogForRecovery() throws Exception { + void testLoadingChangelogForRecovery() throws Exception { final StateBackend backend = StateBackendLoader.loadStateBackendFromKeyedStateHandles( new MockStateBackend(), @@ -142,32 +147,36 @@ public void testLoadingChangelogForRecovery() throws Exception { Collections.singletonList( ChangelogTestUtils.createChangelogStateBackendHandle())); - assertTrue(backend instanceof DeactivatedChangelogStateBackend); + assertThat(backend).isInstanceOf(DeactivatedChangelogStateBackend.class); } - @Test(expected = IllegalArgumentException.class) - public void testRecursiveDelegation() throws Exception { - final StateBackend appBackend = - new ChangelogStateBackend(new ChangelogStateBackend(new MockStateBackend())); - - StateBackendLoader.fromApplicationOrConfigOrDefault( - appBackend, config("rocksdb", true), config(), cl, null); + @Test + void testRecursiveDelegation() { + assertThatThrownBy( + () -> { + final StateBackend appBackend = + new ChangelogStateBackend( + new ChangelogStateBackend(new MockStateBackend())); + StateBackendLoader.fromApplicationOrConfigOrDefault( + appBackend, config("rocksdb", true), config(), cl, null); + }) + .isInstanceOf(IllegalArgumentException.class); } @Test - public void testLoadingHashMapStateBackendFromConfig() throws Exception { + void testLoadingHashMapStateBackendFromConfig() throws Exception { testLoadingStateBackend( "hashmap", HashMapStateBackend.class, JobManagerCheckpointStorage.class, true); } @Test - public void testLoadingHashMapStateBackend() throws Exception { + void testLoadingHashMapStateBackend() throws Exception { testLoadingStateBackend( "hashmap", HashMapStateBackend.class, JobManagerCheckpointStorage.class, false); } @Test - public void testLoadingRocksDBStateBackendFromConfig() throws Exception { + void testLoadingRocksDBStateBackendFromConfig() throws Exception { testLoadingStateBackend( "rocksdb", EmbeddedRocksDBStateBackend.class, @@ -176,7 +185,7 @@ public void testLoadingRocksDBStateBackendFromConfig() throws Exception { } @Test - public void testLoadingRocksDBStateBackend() throws Exception { + void testLoadingRocksDBStateBackend() throws Exception { testLoadingStateBackend( "rocksdb", EmbeddedRocksDBStateBackend.class, @@ -185,7 +194,7 @@ public void testLoadingRocksDBStateBackend() throws Exception { } @Test - public void testEnableChangelogStateBackendInStreamExecutionEnvironment() throws Exception { + void testEnableChangelogStateBackendInStreamExecutionEnvironment() throws Exception { StreamExecutionEnvironment env = getEnvironment(); assertStateBackendAndChangelogInStreamGraphAndJobGraph(env, TernaryBoolean.UNDEFINED); @@ -228,11 +237,13 @@ private void assertDelegateStateBackend( Class delegatedStateBackendClass, CheckpointStorage storage, Class storageClass) { - assertTrue(backend instanceof ChangelogStateBackend); - assertSame( - ((DelegatingStateBackend) backend).getDelegatedStateBackend().getClass(), - delegatedStateBackendClass); - assertSame(storage.getClass(), storageClass); + assertThat(backend).isInstanceOf(ChangelogStateBackend.class); + assertThat( + ((DelegatingStateBackend) backend) + .getDelegatedStateBackend() + .getClass()) + .isSameAs(delegatedStateBackendClass); + assertThat(storage.getClass()).isSameAs(storageClass); } private void testLoadingStateBackend( @@ -286,20 +297,21 @@ private static class EmptyTestSource extends AbstractTestSource {} private void assertStateBackendAndChangelogInStreamGraphAndJobGraph( StreamExecutionEnvironment env, TernaryBoolean isChangelogEnabled) throws Exception { - assertEquals(isChangelogEnabled, env.isChangelogStateBackendEnabled()); + assertThat(env.isChangelogStateBackendEnabled()).isEqualTo(isChangelogEnabled); StreamGraph streamGraph = env.getStreamGraph(false); - assertEquals( - isChangelogEnabled, - streamGraph - .getJobConfiguration() - .getOptional(StateChangelogOptions.ENABLE_STATE_CHANGE_LOG) - .map(TernaryBoolean::fromBoolean) - .orElse(TernaryBoolean.UNDEFINED)); + assertThat( + streamGraph + .getJobConfiguration() + .getOptional(StateChangelogOptions.ENABLE_STATE_CHANGE_LOG) + .map(TernaryBoolean::fromBoolean) + .orElse(TernaryBoolean.UNDEFINED)) + .isEqualTo(isChangelogEnabled); JobCheckpointingSettings checkpointingSettings = streamGraph.getJobGraph().getCheckpointingSettings(); - assertEquals(isChangelogEnabled, checkpointingSettings.isChangelogStateBackendEnabled()); + assertThat(checkpointingSettings.isChangelogStateBackendEnabled()) + .isEqualTo(isChangelogEnabled); } private static class MockStateBackend extends AbstractStateBackend diff --git a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogStateBackendTestUtils.java b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogStateBackendTestUtils.java index 15b53ef3d9c685..f62c8c5d642fca 100644 --- a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogStateBackendTestUtils.java +++ b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogStateBackendTestUtils.java @@ -78,11 +78,7 @@ import java.util.concurrent.CompletableFuture; import static org.apache.flink.runtime.state.StateBackendTestBase.runSnapshot; -import static org.hamcrest.MatcherAssert.assertThat; -import static org.hamcrest.Matchers.containsInAnyOrder; -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; /** Test Utilities for Changelog StateBackend. */ public class ChangelogStateBackendTestUtils { @@ -289,13 +285,13 @@ public static void testMaterializedRestore( VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, kvId); keyedBackend.setCurrentKey(1); - assertEquals(new StateBackendTestBase.TestPojo("u1", 1), state.value()); + assertThat(state.value()).isEqualTo(new StateBackendTestBase.TestPojo("u1", 1)); keyedBackend.setCurrentKey(2); - assertEquals(new StateBackendTestBase.TestPojo("u2", 222), state.value()); + assertThat(state.value()).isEqualTo(new StateBackendTestBase.TestPojo("u2", 222)); keyedBackend.setCurrentKey(3); - assertEquals(new StateBackendTestBase.TestPojo("u3", 3), state.value()); + assertThat(state.value()).isEqualTo(new StateBackendTestBase.TestPojo("u3", 3)); } finally { IOUtils.closeQuietly(keyedBackend); keyedBackend.dispose(); @@ -322,29 +318,30 @@ public static void testMaterializedRestoreForPriorityQueue( TestType elementA10 = new TestType("a", 10); TestType elementA20 = new TestType("a", 20); - assertTrue(priorityQueue.add(elementA100)); - assertTrue(priorityQueue.add(elementA10)); - assertFalse(priorityQueue.add(elementA20)); - assertFalse(priorityQueue.add(elementA10)); + assertThat(priorityQueue.add(elementA100)).isTrue(); + assertThat(priorityQueue.add(elementA10)).isTrue(); + assertThat(priorityQueue.add(elementA20)).isFalse(); + assertThat(priorityQueue.add(elementA10)).isFalse(); List actualList = new ArrayList<>(); try (CloseableIterator iterator = priorityQueue.iterator()) { iterator.forEachRemaining(actualList::add); } - assertThat(actualList, containsInAnyOrder(elementA100, elementA10, elementA20)); + assertThat(actualList) + .containsExactlyInAnyOrder(elementA100, elementA10, elementA20); periodicMaterializationManager.triggerMaterialization(); TestType elementB9 = new TestType("b", 9); - assertTrue(priorityQueue.add(elementB9)); + assertThat(priorityQueue.add(elementB9)).isTrue(); periodicMaterializationManager.triggerMaterialization(); TestType elementC9 = new TestType("c", 9); TestType elementC8 = new TestType("c", 8); - assertFalse(priorityQueue.add(elementC9)); - assertTrue(priorityQueue.add(elementC8)); + assertThat(priorityQueue.add(elementC9)).isFalse(); + assertThat(priorityQueue.add(elementC8)).isTrue(); KeyedStateHandle snapshot = runSnapshot( @@ -377,12 +374,11 @@ public static void testMaterializedRestoreForPriorityQueue( iterator.forEachRemaining(actualListRestore::add); } - assertThat( - actualListRestore, - containsInAnyOrder( - elementA100, elementA10, elementA20, elementB9, elementC9, elementC8)); + assertThat(actualListRestore) + .containsExactlyInAnyOrder( + elementA100, elementA10, elementA20, elementB9, elementC9, elementC8); - assertFalse(priorityQueueRestored.add(new TestType("d", 11))); + assertThat(priorityQueueRestored.add(new TestType("d", 11))).isFalse(); } finally { IOUtils.closeQuietly(keyedBackend); keyedBackend.dispose(); diff --git a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogStateDiscardTest.java b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogStateDiscardTest.java index 30a6fb5d13cb5d..31982c48068c0a 100644 --- a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogStateDiscardTest.java +++ b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogStateDiscardTest.java @@ -55,7 +55,7 @@ import org.apache.flink.util.function.BiConsumerWithException; import org.apache.flink.util.function.TriConsumerWithException; -import org.junit.Test; +import org.junit.jupiter.api.Test; import java.io.IOException; import java.util.ArrayList; @@ -79,18 +79,18 @@ import static org.apache.flink.runtime.state.SnapshotResult.empty; import static org.apache.flink.util.Preconditions.checkState; import static org.apache.flink.util.concurrent.Executors.directExecutor; -import static org.junit.Assert.assertTrue; +import static org.assertj.core.api.Assertions.assertThat; /** * Verifies that any unused state created by {@link ChangelogStateBackend} is discarded. This is * achieved by testing integration between {@link ChangelogKeyedStateBackend} and {@link * StateChangelogWriter} created by {@link FsStateChangelogStorage}. */ -public class ChangelogStateDiscardTest { +class ChangelogStateDiscardTest { private static final Random RANDOM = new Random(); @Test - public void testPreEmptiveUploadDiscardedOnMaterialization() throws Exception { + void testPreEmptiveUploadDiscardedOnMaterialization() throws Exception { singleBackendTest( (backend, writer, uploader) -> { changeAndLogRandomState(backend, uploader.results::size); @@ -103,7 +103,7 @@ public void testPreEmptiveUploadDiscardedOnMaterialization() throws Exception { } @Test - public void testPreEmptiveUploadDiscardedOnSubsumption() throws Exception { + void testPreEmptiveUploadDiscardedOnSubsumption() throws Exception { singleBackendTest( (backend, writer, uploader) -> { changeAndLogRandomState(backend, uploader.results::size); @@ -116,7 +116,7 @@ public void testPreEmptiveUploadDiscardedOnSubsumption() throws Exception { } @Test - public void testPreEmptiveUploadNotDiscardedWithoutNotification() throws Exception { + void testPreEmptiveUploadNotDiscardedWithoutNotification() throws Exception { singleBackendTest( (backend, writer, uploader) -> { changeAndLogRandomState(backend, uploader.results::size); @@ -127,7 +127,7 @@ public void testPreEmptiveUploadNotDiscardedWithoutNotification() throws Excepti } @Test - public void testPreEmptiveUploadDiscardedOnMaterializationIfCompletedLater() throws Exception { + void testPreEmptiveUploadDiscardedOnMaterializationIfCompletedLater() throws Exception { final TaskChangelogRegistry registry = TaskChangelogRegistry.defaultChangelogRegistry(directExecutor()); final TestingUploadScheduler scheduler = new TestingUploadScheduler(registry); @@ -144,13 +144,13 @@ public void testPreEmptiveUploadDiscardedOnMaterializationIfCompletedLater() thr assertDiscarded( results.stream() - .map(h -> (TestingStreamStateHandle) h.getStreamStateHandle()) + .map(h -> (TestingStreamStateHandle) h.getStreamStateHandle()) .collect(toList())); }); } @Test - public void testPreEmptiveUploadDiscardedOnClose() throws Exception { + void testPreEmptiveUploadDiscardedOnClose() throws Exception { final List afterCheckpoint = new ArrayList<>(); final List beforeCheckpoint = new ArrayList<>(); singleBackendTest( @@ -180,7 +180,7 @@ public void testPreEmptiveUploadDiscardedOnClose() throws Exception { * */ @Test - public void testPreEmptiveUploadForMultipleBackends() throws Exception { + void testPreEmptiveUploadForMultipleBackends() throws Exception { // using the same range (rescaling not involved) final KeyGroupRange kgRange = KeyGroupRange.of(0, 10); final JobID jobId = new JobID(); @@ -337,15 +337,15 @@ private static List uploadResult( } private static void assertRetained(List toRetain) { - assertTrue( - "Some state handles were discarded: \n" + toRetain, - toRetain.stream().noneMatch(TestingStreamStateHandle::isDisposed)); + assertThat(toRetain.stream().noneMatch(TestingStreamStateHandle::isDisposed)) + .as("Some state handles were discarded: \n" + toRetain) + .isTrue(); } private static void assertDiscarded(List toDiscard) { - assertTrue( - "Not all state handles were discarded: \n" + toDiscard, - toDiscard.stream().allMatch(TestingStreamStateHandle::isDisposed)); + assertThat(toDiscard.stream().allMatch(TestingStreamStateHandle::isDisposed)) + .as("Not all state handles were discarded: \n" + toDiscard) + .isTrue(); } private static void checkpoint(ChangelogKeyedStateBackend backend, long checkpointId) diff --git a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/KvStateChangeLoggerImplTest.java b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/KvStateChangeLoggerImplTest.java index 523401c5569056..d824c8f4a2818c 100644 --- a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/KvStateChangeLoggerImplTest.java +++ b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/KvStateChangeLoggerImplTest.java @@ -31,7 +31,7 @@ import static org.apache.flink.state.changelog.StateChangeOperation.MERGE_NS; /** {@link KvStateChangeLoggerImpl} test. */ -public class KvStateChangeLoggerImplTest extends StateChangeLoggerTestBase { +class KvStateChangeLoggerImplTest extends StateChangeLoggerTestBase { @Override protected StateChangeLogger getLogger( diff --git a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/PriorityQueueStateChangeLoggerImplTest.java b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/PriorityQueueStateChangeLoggerImplTest.java index ebb0a509561efe..f96acea7c0661b 100644 --- a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/PriorityQueueStateChangeLoggerImplTest.java +++ b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/PriorityQueueStateChangeLoggerImplTest.java @@ -22,7 +22,7 @@ import org.apache.flink.runtime.state.RegisteredPriorityQueueStateBackendMetaInfo; /** {@link PriorityQueueStateChangeLoggerImpl} test. */ -public class PriorityQueueStateChangeLoggerImplTest extends StateChangeLoggerTestBase { +class PriorityQueueStateChangeLoggerImplTest extends StateChangeLoggerTestBase { @Override protected StateChangeLogger getLogger( diff --git a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/StateChangeLoggerTestBase.java b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/StateChangeLoggerTestBase.java index 38efdebb25578b..bbcd63a8252190 100644 --- a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/StateChangeLoggerTestBase.java +++ b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/StateChangeLoggerTestBase.java @@ -23,7 +23,7 @@ import org.apache.flink.runtime.state.changelog.SequenceNumber; import org.apache.flink.runtime.state.changelog.StateChangelogWriter; -import org.junit.Test; +import org.junit.jupiter.api.Test; import java.io.IOException; import java.util.ArrayList; @@ -33,12 +33,12 @@ import static org.apache.flink.runtime.state.changelog.StateChange.META_KEY_GROUP; import static org.apache.flink.state.changelog.StateChangeOperation.METADATA; -import static org.junit.Assert.assertEquals; +import static org.assertj.core.api.Assertions.assertThat; abstract class StateChangeLoggerTestBase { /** A basic test for appending the metadata on first state access. */ @Test - public void testMetadataOperationLogged() throws IOException { + void testMetadataOperationLogged() throws IOException { TestingStateChangelogWriter writer = new TestingStateChangelogWriter(); InternalKeyContextImpl keyContext = new InternalKeyContextImpl<>(KeyGroupRange.of(1, 1000), 1000); @@ -55,7 +55,7 @@ public void testMetadataOperationLogged() throws IOException { StateChangeOperation.byCode((byte) (i % numOpTypes)); log(operation, element, logger, keyContext).ifPresent(expectedAppends::add); } - assertEquals(expectedAppends, writer.appends); + assertThat(writer.appends).isEqualTo(expectedAppends); } }