diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStIncrementalCheckpointRescalingTest.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStIncrementalCheckpointRescalingTest.java index 94b7186dcf8e3..de070b31622f1 100644 --- a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStIncrementalCheckpointRescalingTest.java +++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStIncrementalCheckpointRescalingTest.java @@ -7,7 +7,7 @@ * "License"); you may not use this file except in compliance * with the License. You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, @@ -35,35 +35,36 @@ import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; import org.apache.flink.streaming.util.AbstractStreamOperatorTestHarness; import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness; +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.testutils.junit.utils.TempDirUtils; import org.apache.flink.util.Collector; -import org.apache.flink.util.TestLogger; -import org.junit.Assert; -import org.junit.Before; -import org.junit.Rule; -import org.junit.Test; -import org.junit.rules.TemporaryFolder; -import org.junit.runner.RunWith; -import org.junit.runners.Parameterized; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.TestTemplate; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.api.io.TempDir; import java.util.Arrays; import java.util.Collection; import java.util.List; import static org.apache.flink.state.forst.ForStConfigurableOptions.USE_INGEST_DB_RESTORE_MODE; +import static org.assertj.core.api.Assertions.assertThat; /** Tests to guard rescaling from checkpoint. */ -@RunWith(Parameterized.class) -public class ForStIncrementalCheckpointRescalingTest extends TestLogger { +@ExtendWith(ParameterizedTestExtension.class) +class ForStIncrementalCheckpointRescalingTest { - @Rule public TemporaryFolder rootFolder = new TemporaryFolder(); + @TempDir java.nio.file.Path rootFolder; - @Parameterized.Parameters(name = "useIngestDbRestoreMode: {0}") + @Parameters(name = "useIngestDbRestoreMode: {0}") public static Collection parameters() { return Arrays.asList(false, true); } - @Parameterized.Parameter public boolean useIngestDbRestoreMode; + @Parameter public boolean useIngestDbRestoreMode; private final int maxParallelism = 10; @@ -71,73 +72,73 @@ public static Collection parameters() { private String[] records; - @Before - public void initRecords() throws Exception { + @BeforeEach + void initRecords() throws Exception { records = new String[10]; records[0] = "8"; - Assert.assertEquals( - 0, - KeyGroupRangeAssignment.assignToKeyGroup( - keySelector.getKey(records[0]), maxParallelism)); // group 0 + assertThat( + KeyGroupRangeAssignment.assignToKeyGroup( + keySelector.getKey(records[0]), maxParallelism)) + .isEqualTo(0); // group 0 records[1] = "5"; - Assert.assertEquals( - 1, - KeyGroupRangeAssignment.assignToKeyGroup( - keySelector.getKey(records[1]), maxParallelism)); // group 1 + assertThat( + KeyGroupRangeAssignment.assignToKeyGroup( + keySelector.getKey(records[1]), maxParallelism)) + .isEqualTo(1); // group 1 records[2] = "25"; - Assert.assertEquals( - 2, - KeyGroupRangeAssignment.assignToKeyGroup( - keySelector.getKey(records[2]), maxParallelism)); // group 2 + assertThat( + KeyGroupRangeAssignment.assignToKeyGroup( + keySelector.getKey(records[2]), maxParallelism)) + .isEqualTo(2); // group 2 records[3] = "13"; - Assert.assertEquals( - 3, - KeyGroupRangeAssignment.assignToKeyGroup( - keySelector.getKey(records[3]), maxParallelism)); // group 3 + assertThat( + KeyGroupRangeAssignment.assignToKeyGroup( + keySelector.getKey(records[3]), maxParallelism)) + .isEqualTo(3); // group 3 records[4] = "4"; - Assert.assertEquals( - 4, - KeyGroupRangeAssignment.assignToKeyGroup( - keySelector.getKey(records[4]), maxParallelism)); // group 4 + assertThat( + KeyGroupRangeAssignment.assignToKeyGroup( + keySelector.getKey(records[4]), maxParallelism)) + .isEqualTo(4); // group 4 records[5] = "7"; - Assert.assertEquals( - 5, - KeyGroupRangeAssignment.assignToKeyGroup( - keySelector.getKey(records[5]), maxParallelism)); // group 5 + assertThat( + KeyGroupRangeAssignment.assignToKeyGroup( + keySelector.getKey(records[5]), maxParallelism)) + .isEqualTo(5); // group 5 records[6] = "1"; - Assert.assertEquals( - 6, - KeyGroupRangeAssignment.assignToKeyGroup( - keySelector.getKey(records[6]), maxParallelism)); // group 6 + assertThat( + KeyGroupRangeAssignment.assignToKeyGroup( + keySelector.getKey(records[6]), maxParallelism)) + .isEqualTo(6); // group 6 records[7] = "6"; - Assert.assertEquals( - 7, - KeyGroupRangeAssignment.assignToKeyGroup( - keySelector.getKey(records[7]), maxParallelism)); // group 7 + assertThat( + KeyGroupRangeAssignment.assignToKeyGroup( + keySelector.getKey(records[7]), maxParallelism)) + .isEqualTo(7); // group 7 records[8] = "9"; - Assert.assertEquals( - 8, - KeyGroupRangeAssignment.assignToKeyGroup( - keySelector.getKey(records[8]), maxParallelism)); // group 8 + assertThat( + KeyGroupRangeAssignment.assignToKeyGroup( + keySelector.getKey(records[8]), maxParallelism)) + .isEqualTo(8); // group 8 records[9] = "3"; - Assert.assertEquals( - 9, - KeyGroupRangeAssignment.assignToKeyGroup( - keySelector.getKey(records[9]), maxParallelism)); // group 9 + assertThat( + KeyGroupRangeAssignment.assignToKeyGroup( + keySelector.getKey(records[9]), maxParallelism)) + .isEqualTo(9); // group 9 } - @Test + @TestTemplate @SuppressWarnings("unchecked") - public void testScalingUp() throws Exception { + void testScalingUp() throws Exception { // -----------------------------------------> test with initial parallelism 1 // <--------------------------------------- @@ -149,7 +150,7 @@ public void testScalingUp() throws Exception { harness.setStateBackend(getStateBackend()); harness.setCheckpointStorage( new FileSystemCheckpointStorage( - "file://" + rootFolder.newFolder().getAbsolutePath())); + "file://" + TempDirUtils.newFolder(rootFolder).getAbsolutePath())); harness.open(); validHarnessResult(harness, 1, records); @@ -181,24 +182,24 @@ public void testScalingUp() throws Exception { // task's key-group [0, 4] KeyGroupRange localKeyGroupRange20 = keyGroupPartitions.get(0); - Assert.assertEquals(new KeyGroupRange(0, 4), localKeyGroupRange20); + assertThat(localKeyGroupRange20).isEqualTo(new KeyGroupRange(0, 4)); harness2[0] = getHarnessTest(keySelector, maxParallelism, 2, 0); harness2[0].setStateBackend(getStateBackend()); harness2[0].setCheckpointStorage( new FileSystemCheckpointStorage( - "file://" + rootFolder.newFolder().getAbsolutePath())); + "file://" + TempDirUtils.newFolder(rootFolder).getAbsolutePath())); harness2[0].setup(); harness2[0].initializeState(initState1); harness2[0].open(); // task's key-group [5, 9] KeyGroupRange localKeyGroupRange21 = keyGroupPartitions.get(1); - Assert.assertEquals(new KeyGroupRange(5, 9), localKeyGroupRange21); + assertThat(localKeyGroupRange21).isEqualTo(new KeyGroupRange(5, 9)); harness2[1] = getHarnessTest(keySelector, maxParallelism, 2, 1); harness2[1].setStateBackend(getStateBackend()); harness2[1].setCheckpointStorage( new FileSystemCheckpointStorage( - "file://" + rootFolder.newFolder().getAbsolutePath())); + "file://" + TempDirUtils.newFolder(rootFolder).getAbsolutePath())); harness2[1].setup(); harness2[1].initializeState(initState2); harness2[1].open(); @@ -244,36 +245,36 @@ public void testScalingUp() throws Exception { // task's key-group [0, 3] // this will choose the state handle to harness2[0] to init the target db with clipping. KeyGroupRange localKeyGroupRange30 = keyGroupPartitions.get(0); - Assert.assertEquals(new KeyGroupRange(0, 3), localKeyGroupRange30); + assertThat(localKeyGroupRange30).isEqualTo(new KeyGroupRange(0, 3)); harness3[0] = getHarnessTest(keySelector, maxParallelism, 3, 0); harness3[0].setStateBackend(getStateBackend()); harness3[0].setCheckpointStorage( new FileSystemCheckpointStorage( - "file://" + rootFolder.newFolder().getAbsolutePath())); + "file://" + TempDirUtils.newFolder(rootFolder).getAbsolutePath())); harness3[0].setup(); harness3[0].initializeState(initState1); harness3[0].open(); // task's key-group [4, 6] KeyGroupRange localKeyGroupRange31 = keyGroupPartitions.get(1); - Assert.assertEquals(new KeyGroupRange(4, 6), localKeyGroupRange31); + assertThat(localKeyGroupRange31).isEqualTo(new KeyGroupRange(4, 6)); harness3[1] = getHarnessTest(keySelector, maxParallelism, 3, 1); harness3[1].setStateBackend(getStateBackend()); harness3[1].setCheckpointStorage( new FileSystemCheckpointStorage( - "file://" + rootFolder.newFolder().getAbsolutePath())); + "file://" + TempDirUtils.newFolder(rootFolder).getAbsolutePath())); harness3[1].setup(); harness3[1].initializeState(initState2); harness3[1].open(); // task's key-group [7, 9] KeyGroupRange localKeyGroupRange32 = keyGroupPartitions.get(2); - Assert.assertEquals(new KeyGroupRange(7, 9), localKeyGroupRange32); + assertThat(localKeyGroupRange32).isEqualTo(new KeyGroupRange(7, 9)); harness3[2] = getHarnessTest(keySelector, maxParallelism, 3, 2); harness3[2].setStateBackend(getStateBackend()); harness3[2].setCheckpointStorage( new FileSystemCheckpointStorage( - "file://" + rootFolder.newFolder().getAbsolutePath())); + "file://" + TempDirUtils.newFolder(rootFolder).getAbsolutePath())); harness3[2].setup(); harness3[2].initializeState(initState3); harness3[2].open(); @@ -286,9 +287,9 @@ public void testScalingUp() throws Exception { } } - @Test + @TestTemplate @SuppressWarnings("unchecked") - public void testScalingDown() throws Exception { + void testScalingDown() throws Exception { // -----------------------------------------> test with initial parallelism 3 // <--------------------------------------- @@ -303,32 +304,32 @@ public void testScalingDown() throws Exception { // task's key-group [0, 3], this should trigger the condition to use clip KeyGroupRange localKeyGroupRange30 = keyGroupPartitions.get(0); - Assert.assertEquals(new KeyGroupRange(0, 3), localKeyGroupRange30); + assertThat(localKeyGroupRange30).isEqualTo(new KeyGroupRange(0, 3)); harness3[0] = getHarnessTest(keySelector, maxParallelism, 3, 0); harness3[0].setStateBackend(getStateBackend()); harness3[0].setCheckpointStorage( new FileSystemCheckpointStorage( - "file://" + rootFolder.newFolder().getAbsolutePath())); + "file://" + TempDirUtils.newFolder(rootFolder).getAbsolutePath())); harness3[0].open(); // task's key-group [4, 6] KeyGroupRange localKeyGroupRange31 = keyGroupPartitions.get(1); - Assert.assertEquals(new KeyGroupRange(4, 6), localKeyGroupRange31); + assertThat(localKeyGroupRange31).isEqualTo(new KeyGroupRange(4, 6)); harness3[1] = getHarnessTest(keySelector, maxParallelism, 3, 1); harness3[1].setStateBackend(getStateBackend()); harness3[1].setCheckpointStorage( new FileSystemCheckpointStorage( - "file://" + rootFolder.newFolder().getAbsolutePath())); + "file://" + TempDirUtils.newFolder(rootFolder).getAbsolutePath())); harness3[1].open(); // task's key-group [7, 9] KeyGroupRange localKeyGroupRange32 = keyGroupPartitions.get(2); - Assert.assertEquals(new KeyGroupRange(7, 9), localKeyGroupRange32); + assertThat(localKeyGroupRange32).isEqualTo(new KeyGroupRange(7, 9)); harness3[2] = getHarnessTest(keySelector, maxParallelism, 3, 2); harness3[2].setStateBackend(getStateBackend()); harness3[2].setCheckpointStorage( new FileSystemCheckpointStorage( - "file://" + rootFolder.newFolder().getAbsolutePath())); + "file://" + TempDirUtils.newFolder(rootFolder).getAbsolutePath())); harness3[2].open(); validHarnessResult(harness3[0], 1, records[0], records[1], records[2], records[3]); @@ -371,12 +372,12 @@ public void testScalingDown() throws Exception { // this will choose the state handle generated by harness3[0] to init the target db // without any clipping. KeyGroupRange localKeyGroupRange20 = keyGroupPartitions.get(0); - Assert.assertEquals(new KeyGroupRange(0, 4), localKeyGroupRange20); + assertThat(localKeyGroupRange20).isEqualTo(new KeyGroupRange(0, 4)); harness2[0] = getHarnessTest(keySelector, maxParallelism, 2, 0); harness2[0].setStateBackend(getStateBackend()); harness2[0].setCheckpointStorage( new FileSystemCheckpointStorage( - "file://" + rootFolder.newFolder().getAbsolutePath())); + "file://" + TempDirUtils.newFolder(rootFolder).getAbsolutePath())); harness2[0].setup(); harness2[0].initializeState(initState1); harness2[0].open(); @@ -384,12 +385,12 @@ public void testScalingDown() throws Exception { // task's key-group [5, 9], this will open a empty db, and insert records from two state // handles. KeyGroupRange localKeyGroupRange21 = keyGroupPartitions.get(1); - Assert.assertEquals(new KeyGroupRange(5, 9), localKeyGroupRange21); + assertThat(localKeyGroupRange21).isEqualTo(new KeyGroupRange(5, 9)); harness2[1] = getHarnessTest(keySelector, maxParallelism, 2, 1); harness2[1].setStateBackend(getStateBackend()); harness2[1].setCheckpointStorage( new FileSystemCheckpointStorage( - "file://" + rootFolder.newFolder().getAbsolutePath())); + "file://" + TempDirUtils.newFolder(rootFolder).getAbsolutePath())); harness2[1].setup(); harness2[1].initializeState(initState2); harness2[1].open(); @@ -423,7 +424,7 @@ public void testScalingDown() throws Exception { harness.setStateBackend(getStateBackend()); harness.setCheckpointStorage( new FileSystemCheckpointStorage( - "file://" + rootFolder.newFolder().getAbsolutePath())); + "file://" + TempDirUtils.newFolder(rootFolder).getAbsolutePath())); harness.setup(); harness.initializeState(initState1); harness.open(); @@ -450,8 +451,8 @@ private void validHarnessResult( for (String record : records) { harness.processElement(new StreamRecord<>(record, 1)); StreamRecord outputRecord = (StreamRecord) harness.getOutput().poll(); - Assert.assertNotNull(outputRecord); - Assert.assertEquals(expectedValue, outputRecord.getValue()); + assertThat(outputRecord).isNotNull(); + assertThat(outputRecord.getValue()).isEqualTo(expectedValue); } } diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStInitITCase.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStInitITCase.java index a23762d016960..b5315c5686470 100644 --- a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStInitITCase.java +++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStInitITCase.java @@ -19,35 +19,35 @@ package org.apache.flink.state.forst; import org.apache.flink.runtime.operators.testutils.ExpectedTestException; +import org.apache.flink.testutils.junit.utils.TempDirUtils; import org.apache.flink.util.concurrent.Executors; -import org.junit.Assert; -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.io.File; import java.io.IOException; -import static org.junit.Assert.fail; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.fail; /** Tests for {@link ForStStateBackend} on initialization. */ -public class ForStInitITCase { +class ForStInitITCase { - @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder(); + @TempDir private java.nio.file.Path temporaryFolder; /** * This test checks that the ForSt native code loader still responds to resetting the init flag. */ @Test - public void testResetInitFlag() throws Exception { + void testResetInitFlag() throws Exception { ForStStateBackend.resetForStLoadedFlag(); } @Test - public void testTempLibFolderDeletedOnFail() throws Exception { + void testTempLibFolderDeletedOnFail() throws Exception { ForStStateBackend.setForStInitialized(false); - File tempFolder = temporaryFolder.newFolder(); + File tempFolder = TempDirUtils.newFolder(temporaryFolder); try { ForStStateBackend.ensureForStIsLoaded( tempFolder.getAbsolutePath(), @@ -60,7 +60,7 @@ public void testTempLibFolderDeletedOnFail() throws Exception { // ignored } File[] files = tempFolder.listFiles(); - Assert.assertNotNull(files); - Assert.assertEquals(0, files.length); + assertThat(files).isNotNull(); + assertThat(files).isEmpty(); } } diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStMemoryControllerUtilsTest.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStMemoryControllerUtilsTest.java index e29ce68747d2d..f6198fb218f38 100644 --- a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStMemoryControllerUtilsTest.java +++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStMemoryControllerUtilsTest.java @@ -18,34 +18,33 @@ package org.apache.flink.state.forst; +import org.apache.flink.testutils.junit.utils.TempDirUtils; + import org.forstdb.Cache; import org.forstdb.NativeLibraryLoader; import org.forstdb.WriteBufferManager; -import org.junit.Before; -import org.junit.Rule; -import org.junit.Test; -import org.junit.rules.TemporaryFolder; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; import java.io.IOException; +import java.nio.file.Path; -import static org.hamcrest.CoreMatchers.is; -import static org.hamcrest.MatcherAssert.assertThat; -import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertTrue; +import static org.assertj.core.api.Assertions.assertThat; /** Tests to guard {@link ForStMemoryControllerUtils}. */ -public class ForStMemoryControllerUtilsTest { +class ForStMemoryControllerUtilsTest { - @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder(); + @TempDir private Path temporaryFolder; - @Before - public void ensureRocksDbNativeLibraryLoaded() throws IOException { + @BeforeEach + void ensureRocksDbNativeLibraryLoaded() throws IOException { NativeLibraryLoader.getInstance() - .loadLibrary(temporaryFolder.newFolder().getAbsolutePath()); + .loadLibrary(TempDirUtils.newFolder(temporaryFolder).getAbsolutePath()); } @Test - public void testCreateSharedResourcesWithExpectedCapacity() { + void testCreateSharedResourcesWithExpectedCapacity() { long totalMemorySize = 2048L; double writeBufferRatio = 0.5; double highPriPoolRatio = 0.1; @@ -60,50 +59,55 @@ public void testCreateSharedResourcesWithExpectedCapacity() { ForStMemoryControllerUtils.calculateWriteBufferManagerCapacity( totalMemorySize, writeBufferRatio); - assertThat(factory.actualCacheCapacity, is(expectedCacheCapacity)); - assertThat(factory.actualWbmCapacity, is(expectedWbmCapacity)); - assertThat(forStSharedResources.getWriteBufferManagerCapacity(), is(expectedWbmCapacity)); + assertThat(factory.actualCacheCapacity).isEqualTo(expectedCacheCapacity); + assertThat(factory.actualWbmCapacity).isEqualTo(expectedWbmCapacity); + assertThat(forStSharedResources.getWriteBufferManagerCapacity()) + .isEqualTo(expectedWbmCapacity); } @Test - public void testCalculateForStDefaultArenaBlockSize() { + void testCalculateForStDefaultArenaBlockSize() { final long align = 4 * 1024; final long writeBufferSize = 64 * 1024 * 1024; final long expectArenaBlockSize = writeBufferSize / 8; // Normal case test - assertThat( - "Arena block size calculation error for normal case", - ForStMemoryControllerUtils.calculateForStDefaultArenaBlockSize(writeBufferSize), - is(expectArenaBlockSize)); + assertThat(ForStMemoryControllerUtils.calculateForStDefaultArenaBlockSize(writeBufferSize)) + .as("Arena block size calculation error for normal case") + .isEqualTo(expectArenaBlockSize); // Alignment tests assertThat( - "Arena block size calculation error for alignment case", - ForStMemoryControllerUtils.calculateForStDefaultArenaBlockSize(writeBufferSize - 1), - is(expectArenaBlockSize)); + ForStMemoryControllerUtils.calculateForStDefaultArenaBlockSize( + writeBufferSize - 1)) + .as("Arena block size calculation error for alignment case") + .isEqualTo(expectArenaBlockSize); assertThat( - "Arena block size calculation error for alignment case2", - ForStMemoryControllerUtils.calculateForStDefaultArenaBlockSize(writeBufferSize + 8), - is(expectArenaBlockSize + align)); + ForStMemoryControllerUtils.calculateForStDefaultArenaBlockSize( + writeBufferSize + 8)) + .as("Arena block size calculation error for alignment case2") + .isEqualTo(expectArenaBlockSize + align); } @Test - public void testCalculateForStMutableLimit() { + void testCalculateForStMutableLimit() { long bufferSize = 64 * 1024 * 1024; long limit = bufferSize * 7 / 8; - assertThat(ForStMemoryControllerUtils.calculateForStMutableLimit(bufferSize), is(limit)); + assertThat(ForStMemoryControllerUtils.calculateForStMutableLimit(bufferSize)) + .isEqualTo(limit); } @Test - public void testValidateArenaBlockSize() { + void testValidateArenaBlockSize() { long arenaBlockSize = 8 * 1024 * 1024; - assertFalse( - ForStMemoryControllerUtils.validateArenaBlockSize( - arenaBlockSize, (long) (arenaBlockSize * 0.5))); - assertTrue( - ForStMemoryControllerUtils.validateArenaBlockSize( - arenaBlockSize, (long) (arenaBlockSize * 1.5))); + assertThat( + ForStMemoryControllerUtils.validateArenaBlockSize( + arenaBlockSize, (long) (arenaBlockSize * 0.5))) + .isFalse(); + assertThat( + ForStMemoryControllerUtils.validateArenaBlockSize( + arenaBlockSize, (long) (arenaBlockSize * 1.5))) + .isTrue(); } private static final class TestingForStMemoryFactory diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStMultiClassLoaderTest.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStMultiClassLoaderTest.java index 2753ea63320fc..d6b838752dcc8 100644 --- a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStMultiClassLoaderTest.java +++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStMultiClassLoaderTest.java @@ -18,32 +18,33 @@ package org.apache.flink.state.forst; +import org.apache.flink.testutils.junit.utils.TempDirUtils; import org.apache.flink.util.FlinkUserCodeClassLoaders; import org.apache.flink.util.concurrent.Executors; import org.forstdb.RocksDB; -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.lang.reflect.Method; import java.net.URL; +import java.nio.file.Path; import java.util.concurrent.Executor; import static org.apache.flink.util.FlinkUserCodeClassLoader.NOOP_EXCEPTION_HANDLER; -import static org.junit.Assert.assertNotEquals; +import static org.assertj.core.api.Assertions.assertThat; /** * This test validates that the ForSt JNI library loading works properly in the presence of the * ForSt code being loaded dynamically via reflection. That can happen when ForSt is in the user * code JAR, or in certain test setups. TODO: test working with both ForSt and RocksDB */ -public class ForStMultiClassLoaderTest { +class ForStMultiClassLoaderTest { - @Rule public final TemporaryFolder tmp = new TemporaryFolder(); + @TempDir private Path tmp; @Test - public void testTwoSeparateClassLoaders() throws Exception { + void testTwoSeparateClassLoaders() throws Exception { // collect the libraries / class folders with ForSt related code: the state backend and // ForSt itself final URL codePath1 = @@ -70,13 +71,14 @@ public void testTwoSeparateClassLoaders() throws Exception { final Class clazz1 = Class.forName(className, false, loader1); final Class clazz2 = Class.forName(className, false, loader2); - assertNotEquals( - "Test broken - the two reflectively loaded classes are equal", clazz1, clazz2); + assertThat(clazz1) + .as("Test broken - the two reflectively loaded classes are equal") + .isNotEqualTo(clazz2); final Object instance1 = clazz1.getConstructor().newInstance(); final Object instance2 = clazz2.getConstructor().newInstance(); - final String tempDir = tmp.newFolder().getAbsolutePath(); + final String tempDir = TempDirUtils.newFolder(tmp).getAbsolutePath(); final Method meth1 = clazz1.getDeclaredMethod("ensureForStIsLoaded", String.class, Executor.class); diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStNativeMetricOptionsTest.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStNativeMetricOptionsTest.java index a7f9cd2240377..c1fa565ab7a97 100644 --- a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStNativeMetricOptionsTest.java +++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStNativeMetricOptionsTest.java @@ -20,15 +20,15 @@ import org.apache.flink.configuration.Configuration; -import org.junit.Assert; -import org.junit.Test; +import org.junit.jupiter.api.Test; import static org.apache.flink.configuration.ConfigurationUtils.getBooleanConfigOption; +import static org.assertj.core.api.Assertions.assertThat; /** Test all native metrics can be set using configuration. */ -public class ForStNativeMetricOptionsTest { +class ForStNativeMetricOptionsTest { @Test - public void testNativeMetricsConfigurable() { + void testNativeMetricsConfigurable() { for (ForStProperty property : ForStProperty.values()) { Configuration config = new Configuration(); if (property.getConfigKey().contains("num-files-at-level")) { @@ -39,17 +39,12 @@ public void testNativeMetricsConfigurable() { ForStNativeMetricOptions options = ForStNativeMetricOptions.fromConfig(config); - Assert.assertTrue( - String.format( - "Failed to enable native metrics with property %s", - property.getConfigKey()), - options.isEnabled()); - - Assert.assertTrue( - String.format( - "Failed to enable native metric %s using config", - property.getConfigKey()), - options.getProperties().contains(property)); + assertThat(options.isEnabled()) + .as("Failed to enable native metrics with property %s", property.getConfigKey()) + .isTrue(); + assertThat(options.getProperties()) + .as("Failed to enable native metric %s using config", property.getConfigKey()) + .contains(property); } } } diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStOperationsUtilsTest.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStOperationsUtilsTest.java index abbb63c1c74b8..e711f476c4d73 100644 --- a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStOperationsUtilsTest.java +++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStOperationsUtilsTest.java @@ -18,41 +18,41 @@ package org.apache.flink.state.forst; +import org.apache.flink.testutils.junit.utils.TempDirUtils; import org.apache.flink.util.OperatingSystem; import org.forstdb.ColumnFamilyOptions; import org.forstdb.DBOptions; import org.forstdb.NativeLibraryLoader; import org.forstdb.RocksDB; -import org.junit.BeforeClass; -import org.junit.ClassRule; -import org.junit.Test; -import org.junit.rules.TemporaryFolder; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; import java.io.File; import java.io.IOException; import java.nio.file.Files; import java.util.Collections; -import static org.hamcrest.Matchers.containsString; -import static org.junit.Assert.assertThat; -import static org.junit.Assume.assumeTrue; +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assumptions.assumeTrue; /** Tests for the {@link ForStOperationUtils}. */ -public class ForStOperationsUtilsTest { +class ForStOperationsUtilsTest { - @ClassRule public static final TemporaryFolder TMP_DIR = new TemporaryFolder(); + @TempDir static java.nio.file.Path tmpDir; - @BeforeClass - public static void loadRocksLibrary() throws Exception { - NativeLibraryLoader.getInstance().loadLibrary(TMP_DIR.newFolder().getAbsolutePath()); + @BeforeAll + static void loadRocksLibrary() throws Exception { + NativeLibraryLoader.getInstance() + .loadLibrary(TempDirUtils.newFolder(tmpDir).getAbsolutePath()); } @Test - public void testPathExceptionOnWindows() throws Exception { + void testPathExceptionOnWindows() throws Exception { assumeTrue(OperatingSystem.isWindows()); - final File folder = TMP_DIR.newFolder(); + final File folder = TempDirUtils.newFolder(tmpDir); final File rocksDir = new File(folder, getLongString(247 - folder.getAbsolutePath().length())); @@ -73,9 +73,8 @@ public void testPathExceptionOnWindows() throws Exception { // do not provoke a test failure if this passes, because some setups may actually // support long paths, in which case: great! } catch (IOException e) { - assertThat( - e.getMessage(), - containsString("longer than the directory path length limit for Windows")); + assertThat(e.getMessage()) + .contains("longer than the directory path length limit for Windows"); } } diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStResourceContainerTest.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStResourceContainerTest.java index 9085f6a1fd2df..cfcf489be460f 100644 --- a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStResourceContainerTest.java +++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStResourceContainerTest.java @@ -25,6 +25,7 @@ import org.apache.flink.runtime.state.filesystem.FsCheckpointStorageAccess; import org.apache.flink.state.forst.fs.ForStFlinkFileSystem; import org.apache.flink.state.forst.fs.StringifiedForStFileSystem; +import org.apache.flink.testutils.junit.utils.TempDirUtils; import org.apache.flink.util.function.ThrowingRunnable; import org.forstdb.BlockBasedTableConfig; @@ -43,10 +44,9 @@ import org.forstdb.TableFormatConfig; import org.forstdb.WriteBufferManager; import org.forstdb.WriteOptions; -import org.junit.BeforeClass; -import org.junit.ClassRule; -import org.junit.Test; -import org.junit.rules.TemporaryFolder; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; import java.io.File; import java.io.IOException; @@ -55,35 +55,33 @@ import java.util.Collection; import java.util.HashSet; -import static org.hamcrest.CoreMatchers.is; -import static org.hamcrest.MatcherAssert.assertThat; -import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertTrue; -import static org.junit.Assert.fail; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.fail; /** Tests to guard {@link ForStResourceContainer}. */ -public class ForStResourceContainerTest { +class ForStResourceContainerTest { - @ClassRule public static final TemporaryFolder TMP_FOLDER = new TemporaryFolder(); + @TempDir static java.nio.file.Path tmpFolder; - @BeforeClass - public static void ensureForStNativeLibraryLoaded() throws IOException { - NativeLibraryLoader.getInstance().loadLibrary(TMP_FOLDER.newFolder().getAbsolutePath()); + @BeforeAll + static void ensureForStNativeLibraryLoaded() throws IOException { + NativeLibraryLoader.getInstance() + .loadLibrary(TempDirUtils.newFolder(tmpFolder).getAbsolutePath()); } // ------------------------------------------------------------------------ @Test - public void testFreeDBOptionsAfterClose() throws Exception { + void testFreeDBOptionsAfterClose() throws Exception { ForStResourceContainer container = new ForStResourceContainer(); DBOptions dbOptions = container.getDbOptions(); - assertThat(dbOptions.isOwningHandle(), is(true)); + assertThat(dbOptions.isOwningHandle()).isTrue(); container.close(); - assertThat(dbOptions.isOwningHandle(), is(false)); + assertThat(dbOptions.isOwningHandle()).isFalse(); } @Test - public void testFreeMultipleDBOptionsAfterClose() throws Exception { + void testFreeMultipleDBOptionsAfterClose() throws Exception { ForStResourceContainer container = new ForStResourceContainer(); final int optionNumber = 20; ArrayList dbOptions = new ArrayList<>(optionNumber); @@ -92,7 +90,7 @@ public void testFreeMultipleDBOptionsAfterClose() throws Exception { } container.close(); for (DBOptions dbOption : dbOptions) { - assertThat(dbOption.isOwningHandle(), is(false)); + assertThat(dbOption.isOwningHandle()).isFalse(); } } @@ -103,13 +101,13 @@ public void testFreeMultipleDBOptionsAfterClose() throws Exception { * @throws Exception if unexpected error happened. */ @Test - public void testSharedResourcesAfterClose() throws Exception { + void testSharedResourcesAfterClose() throws Exception { OpaqueMemoryResource sharedResources = getSharedResources(); ForStResourceContainer container = new ForStResourceContainer(null, sharedResources); container.close(); ForStSharedResources forStSharedResources = sharedResources.getResourceHandle(); - assertThat(forStSharedResources.getCache().isOwningHandle(), is(false)); - assertThat(forStSharedResources.getWriteBufferManager().isOwningHandle(), is(false)); + assertThat(forStSharedResources.getCache().isOwningHandle()).isFalse(); + assertThat(forStSharedResources.getWriteBufferManager().isOwningHandle()).isFalse(); } /** @@ -120,7 +118,7 @@ public void testSharedResourcesAfterClose() throws Exception { * @throws Exception if unexpected error happened. */ @Test - public void testGetDbOptionsWithSharedResources() throws Exception { + void testGetDbOptionsWithSharedResources() throws Exception { final int optionNumber = 20; OpaqueMemoryResource sharedResources = getSharedResources(); ForStResourceContainer container = new ForStResourceContainer(null, sharedResources); @@ -130,10 +128,9 @@ public void testGetDbOptionsWithSharedResources() throws Exception { WriteBufferManager writeBufferManager = getWriteBufferManager(dbOptions); writeBufferManagers.add(writeBufferManager); } - assertThat(writeBufferManagers.size(), is(1)); - assertThat( - writeBufferManagers.iterator().next(), - is(sharedResources.getResourceHandle().getWriteBufferManager())); + assertThat(writeBufferManagers).hasSize(1); + assertThat(writeBufferManagers.iterator().next()) + .isEqualTo(sharedResources.getResourceHandle().getWriteBufferManager()); container.close(); } @@ -145,7 +142,7 @@ public void testGetDbOptionsWithSharedResources() throws Exception { * @throws Exception if unexpected error happened. */ @Test - public void testGetColumnFamilyOptionsWithSharedResources() throws Exception { + void testGetColumnFamilyOptionsWithSharedResources() throws Exception { final int optionNumber = 20; OpaqueMemoryResource sharedResources = getSharedResources(); ForStResourceContainer container = new ForStResourceContainer(null, sharedResources); @@ -155,8 +152,9 @@ public void testGetColumnFamilyOptionsWithSharedResources() throws Exception { Cache cache = getBlockCache(columnOptions); caches.add(cache); } - assertThat(caches.size(), is(1)); - assertThat(caches.iterator().next(), is(sharedResources.getResourceHandle().getCache())); + assertThat(caches).hasSize(1); + assertThat(caches.iterator().next()) + .isEqualTo(sharedResources.getResourceHandle().getCache()); container.close(); } @@ -210,16 +208,16 @@ private WriteBufferManager getWriteBufferManager(DBOptions dbOptions) { } @Test - public void testFreeColumnOptionsAfterClose() throws Exception { + void testFreeColumnOptionsAfterClose() throws Exception { ForStResourceContainer container = new ForStResourceContainer(); ColumnFamilyOptions columnFamilyOptions = container.getColumnOptions(); - assertThat(columnFamilyOptions.isOwningHandle(), is(true)); + assertThat(columnFamilyOptions.isOwningHandle()).isTrue(); container.close(); - assertThat(columnFamilyOptions.isOwningHandle(), is(false)); + assertThat(columnFamilyOptions.isOwningHandle()).isFalse(); } @Test - public void testFreeMultipleColumnOptionsAfterClose() throws Exception { + void testFreeMultipleColumnOptionsAfterClose() throws Exception { ForStResourceContainer container = new ForStResourceContainer(); final int optionNumber = 20; ArrayList columnFamilyOptions = new ArrayList<>(optionNumber); @@ -228,12 +226,12 @@ public void testFreeMultipleColumnOptionsAfterClose() throws Exception { } container.close(); for (ColumnFamilyOptions columnFamilyOption : columnFamilyOptions) { - assertThat(columnFamilyOption.isOwningHandle(), is(false)); + assertThat(columnFamilyOption.isOwningHandle()).isFalse(); } } @Test - public void testFreeSharedResourcesAfterClose() throws Exception { + void testFreeSharedResourcesAfterClose() throws Exception { LRUCache cache = new LRUCache(1024L); WriteBufferManager wbm = new WriteBufferManager(1024L, cache); ForStSharedResources sharedResources = new ForStSharedResources(cache, wbm, 1024L, false); @@ -244,24 +242,24 @@ public void testFreeSharedResourcesAfterClose() throws Exception { ForStResourceContainer container = new ForStResourceContainer(null, opaqueResource); container.close(); - assertThat(cache.isOwningHandle(), is(false)); - assertThat(wbm.isOwningHandle(), is(false)); + assertThat(cache.isOwningHandle()).isFalse(); + assertThat(wbm.isOwningHandle()).isFalse(); } @Test - public void testFreeWriteReadOptionsAfterClose() throws Exception { + void testFreeWriteReadOptionsAfterClose() throws Exception { ForStResourceContainer container = new ForStResourceContainer(); WriteOptions writeOptions = container.getWriteOptions(); ReadOptions readOptions = container.getReadOptions(); - assertThat(writeOptions.isOwningHandle(), is(true)); - assertThat(readOptions.isOwningHandle(), is(true)); + assertThat(writeOptions.isOwningHandle()).isTrue(); + assertThat(readOptions.isOwningHandle()).isTrue(); container.close(); - assertThat(writeOptions.isOwningHandle(), is(false)); - assertThat(readOptions.isOwningHandle(), is(false)); + assertThat(writeOptions.isOwningHandle()).isFalse(); + assertThat(readOptions.isOwningHandle()).isFalse(); } @Test - public void testGetColumnFamilyOptionsWithPartitionedIndex() throws Exception { + void testGetColumnFamilyOptionsWithPartitionedIndex() throws Exception { LRUCache cache = new LRUCache(1024L); WriteBufferManager wbm = new WriteBufferManager(1024L, cache); ForStSharedResources sharedResources = new ForStSharedResources(cache, wbm, 1024L, true); @@ -298,20 +296,22 @@ public ColumnFamilyOptions createColumnOptions( ColumnFamilyOptions columnOptions = container.getColumnOptions(); BlockBasedTableConfig actual = (BlockBasedTableConfig) columnOptions.tableFormatConfig(); - assertThat(actual.indexType(), is(IndexType.kTwoLevelIndexSearch)); - assertThat(actual.partitionFilters(), is(true)); - assertThat(actual.pinTopLevelIndexAndFilter(), is(true)); - assertFalse(actual.filterPolicy() == blockBasedFilter); + assertThat(actual.indexType()).isEqualTo(IndexType.kTwoLevelIndexSearch); + assertThat(actual.partitionFilters()).isTrue(); + assertThat(actual.pinTopLevelIndexAndFilter()).isTrue(); + assertThat(actual.filterPolicy()).isNotSameAs(blockBasedFilter); } - assertFalse("Block based filter is left unclosed.", blockBasedFilter.isOwningHandle()); + assertThat(blockBasedFilter.isOwningHandle()) + .as("Block based filter is left unclosed.") + .isFalse(); } @Test - public void testDirectoryResources() throws Exception { - Path localJobPath = new Path(TMP_FOLDER.newFolder().getPath()); + void testDirectoryResources() throws Exception { + Path localJobPath = new Path(TempDirUtils.newFolder(tmpFolder).getPath()); Path localBasePath = new Path(localJobPath, "base"); localBasePath.getFileSystem().mkdirs(localBasePath); - Path remoteJobPath = new Path(TMP_FOLDER.newFolder().getPath()); + Path remoteJobPath = new Path(TempDirUtils.newFolder(tmpFolder).getPath()); Path remoteBasePath = new Path(remoteJobPath, "base"); remoteBasePath.getFileSystem().mkdirs(remoteBasePath); try (final ForStResourceContainer optionsContainer = @@ -323,7 +323,7 @@ public void testDirectoryResources() throws Exception { localJobPath, localBasePath, remoteJobPath, remoteBasePath), null, new FsCheckpointStorageAccess( - new Path(TMP_FOLDER.newFolder().getPath()), + new Path(TempDirUtils.newFolder(tmpFolder).getPath()), null, new JobID(), 1024, @@ -331,25 +331,25 @@ public void testDirectoryResources() throws Exception { null, false)) { optionsContainer.prepareDirectories(); - assertTrue(new File(localBasePath.getPath()).exists()); - assertTrue(new File(remoteBasePath.getPath()).exists()); - assertTrue(optionsContainer.getDbOptions().getEnv() instanceof FlinkEnv); + assertThat(new File(localBasePath.getPath())).exists(); + assertThat(new File(remoteBasePath.getPath())).exists(); + assertThat(optionsContainer.getDbOptions().getEnv()).isInstanceOf(FlinkEnv.class); optionsContainer.clearDirectories(); - assertFalse(new File(localBasePath.getPath()).exists()); + assertThat(new File(localBasePath.getPath())).doesNotExist(); - assertTrue(new File(remoteBasePath.getPath()).exists()); + assertThat(new File(remoteBasePath.getPath())).exists(); optionsContainer.forceClearRemoteDirectories(); // Do not delete remote directory because it is not created by ForStResourceContainer - assertTrue(new File(remoteBasePath.getPath()).exists()); + assertThat(new File(remoteBasePath.getPath())).exists(); } } @Test - public void testFileSystemInit() throws Exception { - Path localBasePath = new Path(TMP_FOLDER.newFolder().getPath()); - Path remoteBasePath = new Path(TMP_FOLDER.newFolder().getPath()); + void testFileSystemInit() throws Exception { + Path localBasePath = new Path(TempDirUtils.newFolder(tmpFolder).getPath()); + Path remoteBasePath = new Path(TempDirUtils.newFolder(tmpFolder).getPath()); ArrayList columnFamilyHandles = new ArrayList<>(1); ArrayList columnFamilyDescriptors = new ArrayList<>(1); columnFamilyDescriptors.add(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY)); diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendConfigTest.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendConfigTest.java index 624bbf0b577db..1505057f8e2f2 100644 --- a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendConfigTest.java +++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendConfigTest.java @@ -42,6 +42,7 @@ import org.apache.flink.runtime.state.filesystem.FsCheckpointStorageAccess; import org.apache.flink.runtime.state.ttl.TtlTimeProvider; import org.apache.flink.runtime.util.TestingTaskManagerRuntimeInfo; +import org.apache.flink.testutils.junit.utils.TempDirUtils; import org.apache.flink.util.FileUtils; import org.apache.commons.lang3.RandomUtils; @@ -54,11 +55,9 @@ import org.forstdb.FlushOptions; import org.forstdb.InfoLogLevel; import org.forstdb.util.SizeUnit; -import org.junit.Assume; -import org.junit.Rule; -import org.junit.Test; +import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Timeout; -import org.junit.rules.TemporaryFolder; +import org.junit.jupiter.api.io.TempDir; import java.io.File; import java.io.IOException; @@ -71,51 +70,44 @@ import static org.apache.flink.state.forst.ForStStateBackend.LOCAL_DIR_AS_PRIMARY_SHORTCUT; import static org.apache.flink.state.forst.ForStTestUtils.createKeyedStateBackend; -import static org.hamcrest.CoreMatchers.anyOf; -import static org.hamcrest.MatcherAssert.assertThat; -import static org.hamcrest.Matchers.startsWith; -import static org.junit.Assert.assertArrayEquals; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertNull; -import static org.junit.Assert.assertTrue; -import static org.junit.Assert.fail; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.assertj.core.api.Assertions.fail; +import static org.junit.jupiter.api.Assumptions.assumeTrue; /** Tests for configuring the ForSt State Backend. */ -public class ForStStateBackendConfigTest { +class ForStStateBackendConfigTest { - @Rule public final TemporaryFolder tempFolder = new TemporaryFolder(); + @TempDir java.nio.file.Path tempFolder; // ------------------------------------------------------------------------ // default values // ------------------------------------------------------------------------ @Test - public void testDefaultDbLogDir() throws Exception { + void testDefaultDbLogDir() throws Exception { final ForStStateBackend backend = new ForStStateBackend(); final File logFile = File.createTempFile(getClass().getSimpleName() + "-", ".log"); // set the environment variable 'log.file' with the Flink log file location System.setProperty("log.file", logFile.getPath()); try (ForStResourceContainer container = backend.createOptionsAndResourceContainer(new Path(tempFolder.toString()))) { - assertEquals( - ForStConfigurableOptions.LOG_LEVEL.defaultValue(), - container.getDbOptions().infoLogLevel()); - assertEquals(logFile.getParent(), container.getDbOptions().dbLogDir()); + assertThat(container.getDbOptions().infoLogLevel()) + .isEqualTo(ForStConfigurableOptions.LOG_LEVEL.defaultValue()); + assertThat(container.getDbOptions().dbLogDir()).isEqualTo(logFile.getParent()); } finally { logFile.delete(); } StringBuilder longInstanceBasePath = - new StringBuilder(tempFolder.newFolder().getAbsolutePath()); + new StringBuilder(TempDirUtils.newFolder(tempFolder).getAbsolutePath()); while (longInstanceBasePath.length() < 255) { longInstanceBasePath.append("/append-for-long-path"); } try (ForStResourceContainer container = backend.createOptionsAndResourceContainer( new Path(longInstanceBasePath.toString()))) { - assertTrue(container.getDbOptions().dbLogDir().isEmpty()); + assertThat(container.getDbOptions().dbLogDir()).isEmpty(); } finally { logFile.delete(); } @@ -127,36 +119,37 @@ public void testDefaultDbLogDir() throws Exception { /** This test checks the behavior for basic setting of local DB directories. */ @Test - public void testSetDbPath() throws Exception { + void testSetDbPath() throws Exception { final ForStStateBackend forStStateBackend = new ForStStateBackend(); - final String testDir1 = tempFolder.newFolder().getAbsolutePath(); - final String testDir2 = tempFolder.newFolder().getAbsolutePath(); + final String testDir1 = TempDirUtils.newFolder(tempFolder).getAbsolutePath(); + final String testDir2 = TempDirUtils.newFolder(tempFolder).getAbsolutePath(); - assertNull(forStStateBackend.getLocalDbStoragePaths()); + assertThat(forStStateBackend.getLocalDbStoragePaths()).isNull(); forStStateBackend.setLocalDbStoragePath(testDir1); - assertArrayEquals(new String[] {testDir1}, forStStateBackend.getLocalDbStoragePaths()); + assertThat(forStStateBackend.getLocalDbStoragePaths()).containsExactly(testDir1); forStStateBackend.setLocalDbStoragePath(null); - assertNull(forStStateBackend.getLocalDbStoragePaths()); + assertThat(forStStateBackend.getLocalDbStoragePaths()).isNull(); forStStateBackend.setLocalDbStoragePaths(testDir1, testDir2); - assertArrayEquals( - new String[] {testDir1, testDir2}, forStStateBackend.getLocalDbStoragePaths()); + assertThat(forStStateBackend.getLocalDbStoragePaths()).containsExactly(testDir1, testDir2); - final MockEnvironment env = getMockEnvironment(tempFolder.newFolder()); + final MockEnvironment env = getMockEnvironment(TempDirUtils.newFolder(tempFolder)); final ForStKeyedStateBackend keyedBackend = createKeyedStateBackend(forStStateBackend, env, IntSerializer.INSTANCE); try { Path instanceBasePath = keyedBackend.getLocalBasePath(); - assertThat( - instanceBasePath.getPath(), anyOf(startsWith(testDir1), startsWith(testDir2))); + assertThat(instanceBasePath.getPath()) + .satisfiesAnyOf( + p -> assertThat(p).startsWith(testDir1), + p -> assertThat(p).startsWith(testDir2)); //noinspection NullArgumentToVariableArgMethod forStStateBackend.setLocalDbStoragePaths(null); - assertNull(forStStateBackend.getLocalDbStoragePaths()); + assertThat(forStStateBackend.getLocalDbStoragePaths()).isNull(); } finally { keyedBackend.dispose(); keyedBackend.close(); @@ -165,8 +158,8 @@ public void testSetDbPath() throws Exception { } @Test - public void testConfigureForStCompressionPerLevel() throws Exception { - final MockEnvironment env = getMockEnvironment(tempFolder.newFolder()); + void testConfigureForStCompressionPerLevel() throws Exception { + final MockEnvironment env = getMockEnvironment(TempDirUtils.newFolder(tempFolder)); ForStStateBackend forStStateBackend = new ForStStateBackend(); CompressionType[] compressionTypes = { CompressionType.NO_COMPRESSION, CompressionType.SNAPPY_COMPRESSION @@ -181,40 +174,40 @@ public void testConfigureForStCompressionPerLevel() throws Exception { ForStResourceContainer resourceContainer = forStStateBackend.createOptionsAndResourceContainer( - new Path(tempFolder.newFile().getAbsolutePath())); + new Path(TempDirUtils.newFile(tempFolder).getAbsolutePath())); ColumnFamilyOptions columnFamilyOptions = resourceContainer.getColumnOptions(); - assertArrayEquals(compressionTypes, columnFamilyOptions.compressionPerLevel().toArray()); + assertThat(columnFamilyOptions.compressionPerLevel().toArray()).isEqualTo(compressionTypes); resourceContainer.close(); env.close(); } @Test - public void testStoragePathWithFilePrefix() throws Exception { - final File folder = tempFolder.newFolder(); + void testStoragePathWithFilePrefix() throws Exception { + final File folder = TempDirUtils.newFolder(tempFolder); final String dbStoragePath = new Path(folder.toURI().toString()).toString(); - assertTrue(dbStoragePath.startsWith("file:")); + assertThat(dbStoragePath).startsWith("file:"); testLocalDbPaths(dbStoragePath, folder); } @Test - public void testWithDefaultFsSchemeNoStoragePath() throws Exception { + void testWithDefaultFsSchemeNoStoragePath() throws Exception { try { // set the default file system scheme Configuration config = new Configuration(); config.set(CoreOptions.DEFAULT_FILESYSTEM_SCHEME, "file:///mydomain.com:8020/flink"); FileSystem.initialize(config); - testLocalDbPaths(null, tempFolder.getRoot()); + testLocalDbPaths(null, tempFolder.toFile()); } finally { FileSystem.initialize(new Configuration()); } } @Test - public void testWithDefaultFsSchemeAbsoluteStoragePath() throws Exception { - final File folder = tempFolder.newFolder(); + void testWithDefaultFsSchemeAbsoluteStoragePath() throws Exception { + final File folder = TempDirUtils.newFolder(tempFolder); final String dbStoragePath = folder.getAbsolutePath(); try { @@ -233,17 +226,17 @@ private void testLocalDbPaths(String configuredPath, File expectedPath) throws E final ForStStateBackend forStBackend = new ForStStateBackend(); forStBackend.setLocalDbStoragePath(configuredPath); - final MockEnvironment env = getMockEnvironment(tempFolder.newFolder()); + final MockEnvironment env = getMockEnvironment(TempDirUtils.newFolder(tempFolder)); ForStKeyedStateBackend keyedBackend = createKeyedStateBackend(forStBackend, env, IntSerializer.INSTANCE); try { Path instanceBasePath = keyedBackend.getLocalBasePath(); - assertThat(instanceBasePath.getPath(), startsWith(expectedPath.getAbsolutePath())); + assertThat(instanceBasePath.getPath()).startsWith(expectedPath.getAbsolutePath()); //noinspection NullArgumentToVariableArgMethod forStBackend.setLocalDbStoragePaths(null); - assertNull(forStBackend.getLocalDbStoragePaths()); + assertThat(forStBackend.getLocalDbStoragePaths()).isNull(); } finally { keyedBackend.dispose(); keyedBackend.close(); @@ -252,30 +245,36 @@ private void testLocalDbPaths(String configuredPath, File expectedPath) throws E } /** Validates that empty arguments for the local DB path are invalid. */ - @Test(expected = IllegalArgumentException.class) - public void testSetEmptyPaths() throws Exception { + @Test + void testSetEmptyPaths() { ForStStateBackend forStStateBackend = new ForStStateBackend(); - forStStateBackend.setLocalDbStoragePaths(); + assertThatThrownBy(forStStateBackend::setLocalDbStoragePaths) + .isInstanceOf(IllegalArgumentException.class); } /** Validates that schemes other than 'file:/' are not allowed. */ - @Test(expected = IllegalArgumentException.class) - public void testNonFileSchemePath() throws Exception { + @Test + void testNonFileSchemePath() { ForStStateBackend forStStateBackend = new ForStStateBackend(); - forStStateBackend.setLocalDbStoragePath("hdfs:///some/path/to/perdition"); + assertThatThrownBy( + () -> + forStStateBackend.setLocalDbStoragePath( + "hdfs:///some/path/to/perdition")) + .isInstanceOf(IllegalArgumentException.class); } - @Test(expected = IllegalArgumentException.class) - public void testDbPathRelativePaths() throws Exception { + @Test + void testDbPathRelativePaths() { ForStStateBackend forStStateBackend = new ForStStateBackend(); - forStStateBackend.setLocalDbStoragePath("relative/path"); + assertThatThrownBy(() -> forStStateBackend.setLocalDbStoragePath("relative/path")) + .isInstanceOf(IllegalArgumentException.class); } @Test @Timeout(value = 60) - public void testCleanRelocatedDbLogs() throws Exception { - final File folder = tempFolder.newFolder(); - final File relocatedDBLogDir = tempFolder.newFolder("db_logs"); + void testCleanRelocatedDbLogs() throws Exception { + final File folder = TempDirUtils.newFolder(tempFolder); + final File relocatedDBLogDir = TempDirUtils.newFolder(tempFolder, "db_logs"); final File logFile = new File(relocatedDBLogDir, "taskManager.log"); Files.createFile(logFile.toPath()); System.setProperty("log.file", logFile.getAbsolutePath()); @@ -290,7 +289,7 @@ public void testCleanRelocatedDbLogs() throws Exception { final String dbStoragePath = new Path(folder.toURI().toString()).toString(); forStBackend.setLocalDbStoragePath(dbStoragePath); - final MockEnvironment env = getMockEnvironment(tempFolder.newFolder()); + final MockEnvironment env = getMockEnvironment(TempDirUtils.newFolder(tempFolder)); ForStKeyedStateBackend keyedBackend = createKeyedStateBackend(forStBackend, env, IntSerializer.INSTANCE); @@ -298,7 +297,7 @@ public void testCleanRelocatedDbLogs() throws Exception { Path localForStPath = new Path(localBasePath, "db"); // avoid tests without relocate. - Assume.assumeTrue(localForStPath.getPath().length() <= 255 - "_LOG".length()); + assumeTrue(localForStPath.getPath().length() <= 255 - "_LOG".length()); java.nio.file.Path[] relocatedDbLogs; try { @@ -318,8 +317,8 @@ public void testCleanRelocatedDbLogs() throws Exception { } relocatedDbLogs = FileUtils.listDirectory(relocatedDBLogDir.toPath()); - assertEquals(1, relocatedDbLogs.length); - assertEquals("taskManager.log", relocatedDbLogs[0].toFile().getName()); + assertThat(relocatedDbLogs).hasSize(1); + assertThat(relocatedDbLogs[0].toFile().getName()).isEqualTo("taskManager.log"); } // ------------------------------------------------------------------------ @@ -331,12 +330,12 @@ public void testCleanRelocatedDbLogs() throws Exception { * {@link Environment} when no db storage path is set. */ @Test - public void testUseTempDirectories() throws Exception { + void testUseTempDirectories() throws Exception { ForStStateBackend forStStateBackend = new ForStStateBackend(); - File dir1 = tempFolder.newFolder(); + File dir1 = TempDirUtils.newFolder(tempFolder); - assertNull(forStStateBackend.getLocalDbStoragePaths()); + assertThat(forStStateBackend.getLocalDbStoragePaths()).isNull(); final MockEnvironment env = getMockEnvironment(dir1); JobID jobID = env.getJobID(); @@ -360,7 +359,7 @@ public void testUseTempDirectories() throws Exception { try { Path instanceBasePath = keyedBackend.getLocalBasePath(); - assertThat(instanceBasePath.getPath(), startsWith(dir1.getAbsolutePath())); + assertThat(instanceBasePath.getPath()).startsWith(dir1.getAbsolutePath()); } finally { keyedBackend.dispose(); keyedBackend.close(); @@ -373,14 +372,13 @@ public void testUseTempDirectories() throws Exception { // ------------------------------------------------------------------------ @Test - public void testFailWhenNoLocalStorageDir() throws Exception { - final File targetDir = tempFolder.newFolder(); - Assume.assumeTrue( - "Cannot mark directory non-writable", targetDir.setWritable(false, false)); + void testFailWhenNoLocalStorageDir() throws Exception { + final File targetDir = TempDirUtils.newFolder(tempFolder); + assumeTrue(targetDir.setWritable(false, false), "Cannot mark directory non-writable"); ForStStateBackend forStStateBackend = new ForStStateBackend(); - try (MockEnvironment env = getMockEnvironment(tempFolder.newFolder())) { + try (MockEnvironment env = getMockEnvironment(TempDirUtils.newFolder(tempFolder))) { forStStateBackend.setLocalDbStoragePath(targetDir.getAbsolutePath()); boolean hasFailure = false; @@ -404,12 +402,14 @@ public void testFailWhenNoLocalStorageDir() throws Exception { Collections.emptyList(), cancelStreamRegistry)); } catch (Exception e) { - assertTrue(e.getMessage().contains("No local storage directories available")); - assertTrue(e.getMessage().contains(targetDir.getAbsolutePath())); + assertThat(e.getMessage()) + .contains("No local storage directories available") + .contains(targetDir.getAbsolutePath()); hasFailure = true; } - assertTrue( - "We must see a failure because no storaged directory is feasible.", hasFailure); + assertThat(hasFailure) + .as("We must see a failure because no storaged directory is feasible.") + .isTrue(); } finally { //noinspection ResultOfMethodCallIgnored targetDir.setWritable(true, false); @@ -417,15 +417,14 @@ public void testFailWhenNoLocalStorageDir() throws Exception { } @Test - public void testContinueOnSomeDbDirectoriesMissing() throws Exception { - final File targetDir1 = tempFolder.newFolder(); - final File targetDir2 = tempFolder.newFolder(); - Assume.assumeTrue( - "Cannot mark directory non-writable", targetDir1.setWritable(false, false)); + void testContinueOnSomeDbDirectoriesMissing() throws Exception { + final File targetDir1 = TempDirUtils.newFolder(tempFolder); + final File targetDir2 = TempDirUtils.newFolder(tempFolder); + assumeTrue(targetDir1.setWritable(false, false), "Cannot mark directory non-writable"); ForStStateBackend forStStateBackend = new ForStStateBackend(); - try (MockEnvironment env = getMockEnvironment(tempFolder.newFolder())) { + try (MockEnvironment env = getMockEnvironment(TempDirUtils.newFolder(tempFolder))) { forStStateBackend.setLocalDbStoragePaths( targetDir1.getAbsolutePath(), targetDir2.getAbsolutePath()); @@ -466,7 +465,7 @@ public void testContinueOnSomeDbDirectoriesMissing() throws Exception { // ForSt Options // ------------------------------------------------------------------------ @Test - public void testConfigurableOptionsFromConfig() throws Exception { + void testConfigurableOptionsFromConfig() throws Exception { Configuration configuration = new Configuration(); // verify illegal configuration @@ -537,40 +536,39 @@ public void testConfigurableOptionsFromConfig() throws Exception { false)) { DBOptions dbOptions = optionsContainer.getDbOptions(); - assertEquals(-1, dbOptions.maxOpenFiles()); - assertEquals(InfoLogLevel.DEBUG_LEVEL, dbOptions.infoLogLevel()); - assertEquals("/tmp/ForSt-logs/", dbOptions.dbLogDir()); - assertEquals(10, dbOptions.keepLogFileNum()); - assertEquals(2 * SizeUnit.MB, dbOptions.maxLogFileSize()); + assertThat(dbOptions.maxOpenFiles()).isEqualTo(-1); + assertThat(dbOptions.infoLogLevel()).isEqualTo(InfoLogLevel.DEBUG_LEVEL); + assertThat(dbOptions.dbLogDir()).isEqualTo("/tmp/ForSt-logs/"); + assertThat(dbOptions.keepLogFileNum()).isEqualTo(10); + assertThat(dbOptions.maxLogFileSize()).isEqualTo(2 * SizeUnit.MB); ColumnFamilyOptions columnOptions = optionsContainer.getColumnOptions(); - assertEquals(CompactionStyle.LEVEL, columnOptions.compactionStyle()); - assertTrue(columnOptions.levelCompactionDynamicLevelBytes()); - assertEquals(8 * SizeUnit.MB, columnOptions.targetFileSizeBase()); - assertEquals(128 * SizeUnit.MB, columnOptions.maxBytesForLevelBase()); - assertEquals(4, columnOptions.maxWriteBufferNumber()); - assertEquals(2, columnOptions.minWriteBufferNumberToMerge()); - assertEquals(64 * SizeUnit.MB, columnOptions.writeBufferSize()); - assertEquals( - Arrays.asList( + assertThat(columnOptions.compactionStyle()).isEqualTo(CompactionStyle.LEVEL); + assertThat(columnOptions.levelCompactionDynamicLevelBytes()).isTrue(); + assertThat(columnOptions.targetFileSizeBase()).isEqualTo(8 * SizeUnit.MB); + assertThat(columnOptions.maxBytesForLevelBase()).isEqualTo(128 * SizeUnit.MB); + assertThat(columnOptions.maxWriteBufferNumber()).isEqualTo(4); + assertThat(columnOptions.minWriteBufferNumberToMerge()).isEqualTo(2); + assertThat(columnOptions.writeBufferSize()).isEqualTo(64 * SizeUnit.MB); + assertThat(columnOptions.compressionPerLevel()) + .containsExactly( CompressionType.NO_COMPRESSION, CompressionType.SNAPPY_COMPRESSION, - CompressionType.LZ4_COMPRESSION), - columnOptions.compressionPerLevel()); - assertEquals(3600, columnOptions.periodicCompactionSeconds()); + CompressionType.LZ4_COMPRESSION); + assertThat(columnOptions.periodicCompactionSeconds()).isEqualTo(3600); BlockBasedTableConfig tableConfig = (BlockBasedTableConfig) columnOptions.tableFormatConfig(); - assertEquals(4 * SizeUnit.KB, tableConfig.blockSize()); - assertEquals(8 * SizeUnit.KB, tableConfig.metadataBlockSize()); - assertEquals(512 * SizeUnit.MB, tableConfig.blockCacheSize()); - assertTrue(tableConfig.filterPolicy() instanceof BloomFilter); + assertThat(tableConfig.blockSize()).isEqualTo(4 * SizeUnit.KB); + assertThat(tableConfig.metadataBlockSize()).isEqualTo(8 * SizeUnit.KB); + assertThat(tableConfig.blockCacheSize()).isEqualTo(512 * SizeUnit.MB); + assertThat(tableConfig.filterPolicy()).isInstanceOf(BloomFilter.class); } } } @Test - public void testOptionsFactory() throws Exception { + void testOptionsFactory() throws Exception { ForStStateBackend forStStateBackend = new ForStStateBackend(); // verify that user-defined options factory could be configured via config.yaml @@ -580,12 +578,12 @@ public void testOptionsFactory() throws Exception { forStStateBackend = forStStateBackend.configure(config, getClass().getClassLoader()); - assertTrue(forStStateBackend.getForStOptions() instanceof TestOptionsFactory); + assertThat(forStStateBackend.getForStOptions()).isInstanceOf(TestOptionsFactory.class); try (ForStResourceContainer optionsContainer = forStStateBackend.createOptionsAndResourceContainer(null)) { DBOptions dbOptions = optionsContainer.getDbOptions(); - assertEquals(4, dbOptions.maxBackgroundJobs()); + assertThat(dbOptions.maxBackgroundJobs()).isEqualTo(4); } // verify that user-defined options factory could be set programmatically and override @@ -609,12 +607,12 @@ public ColumnFamilyOptions createColumnOptions( try (ForStResourceContainer optionsContainer = forStStateBackend.createOptionsAndResourceContainer(null)) { ColumnFamilyOptions colCreated = optionsContainer.getColumnOptions(); - assertEquals(CompactionStyle.FIFO, colCreated.compactionStyle()); + assertThat(colCreated.compactionStyle()).isEqualTo(CompactionStyle.FIFO); } } @Test - public void testConfigurableOptions() throws Exception { + void testConfigurableOptions() throws Exception { Configuration configuration = new Configuration(); configuration.set(ForStConfigurableOptions.COMPACTION_STYLE, CompactionStyle.UNIVERSAL); try (final ForStResourceContainer optionsContainer = @@ -629,8 +627,8 @@ public void testConfigurableOptions() throws Exception { false)) { final ColumnFamilyOptions columnFamilyOptions = optionsContainer.getColumnOptions(); - assertNotNull(columnFamilyOptions); - assertEquals(CompactionStyle.UNIVERSAL, columnFamilyOptions.compactionStyle()); + assertThat(columnFamilyOptions).isNotNull(); + assertThat(columnFamilyOptions.compactionStyle()).isEqualTo(CompactionStyle.UNIVERSAL); } try (final ForStResourceContainer optionsContainer = @@ -645,13 +643,13 @@ public void testConfigurableOptions() throws Exception { false)) { final ColumnFamilyOptions columnFamilyOptions = optionsContainer.getColumnOptions(); - assertNotNull(columnFamilyOptions); - assertEquals(CompactionStyle.LEVEL, columnFamilyOptions.compactionStyle()); + assertThat(columnFamilyOptions).isNotNull(); + assertThat(columnFamilyOptions.compactionStyle()).isEqualTo(CompactionStyle.LEVEL); } } @Test - public void testPredefinedAndOptionsFactory() throws Exception { + void testPredefinedAndOptionsFactory() throws Exception { final ForStOptionsFactory optionsFactory = new ForStOptionsFactory() { @Override @@ -672,8 +670,8 @@ public ColumnFamilyOptions createColumnOptions( new ForStResourceContainer(optionsFactory)) { final ColumnFamilyOptions columnFamilyOptions = optionsContainer.getColumnOptions(); - assertNotNull(columnFamilyOptions); - assertEquals(CompactionStyle.UNIVERSAL, columnFamilyOptions.compactionStyle()); + assertThat(columnFamilyOptions).isNotNull(); + assertThat(columnFamilyOptions.compactionStyle()).isEqualTo(CompactionStyle.UNIVERSAL); } } @@ -682,15 +680,15 @@ public ColumnFamilyOptions createColumnOptions( // ------------------------------------------------------------------------ @Test - public void testForStReconfigurationCopiesExistingValues() throws Exception { + void testForStReconfigurationCopiesExistingValues() throws Exception { final ForStStateBackend original = new ForStStateBackend(); final ForStOptionsFactory optionsFactory = new TestOptionsFactory(); original.setForStOptions(optionsFactory); final String[] localDirs = new String[] { - tempFolder.newFolder().getAbsolutePath(), - tempFolder.newFolder().getAbsolutePath() + TempDirUtils.newFolder(tempFolder).getAbsolutePath(), + TempDirUtils.newFolder(tempFolder).getAbsolutePath() }; original.setLocalDbStoragePaths(localDirs); @@ -698,8 +696,8 @@ public void testForStReconfigurationCopiesExistingValues() throws Exception { original.configure( new Configuration(), Thread.currentThread().getContextClassLoader()); - assertArrayEquals(original.getLocalDbStoragePaths(), copy.getLocalDbStoragePaths()); - assertEquals(original.getForStOptions(), copy.getForStOptions()); + assertThat(copy.getLocalDbStoragePaths()).isEqualTo(original.getLocalDbStoragePaths()); + assertThat(copy.getForStOptions()).isEqualTo(original.getForStOptions()); } // ------------------------------------------------------------------------ @@ -707,36 +705,28 @@ public void testForStReconfigurationCopiesExistingValues() throws Exception { // ------------------------------------------------------------------------ @Test - public void testDefaultMemoryControlParameters() { + void testDefaultMemoryControlParameters() { ForStMemoryConfiguration memSettings = new ForStMemoryConfiguration(); - assertTrue(memSettings.isUsingManagedMemory()); - assertFalse(memSettings.isUsingFixedMemoryPerSlot()); - assertEquals( - ForStOptions.HIGH_PRIORITY_POOL_RATIO.defaultValue(), - memSettings.getHighPriorityPoolRatio(), - 0.0); - assertEquals( - ForStOptions.WRITE_BUFFER_RATIO.defaultValue(), - memSettings.getWriteBufferRatio(), - 0.0); + assertThat(memSettings.isUsingManagedMemory()).isTrue(); + assertThat(memSettings.isUsingFixedMemoryPerSlot()).isFalse(); + assertThat(memSettings.getHighPriorityPoolRatio()) + .isEqualTo(ForStOptions.HIGH_PRIORITY_POOL_RATIO.defaultValue()); + assertThat(memSettings.getWriteBufferRatio()) + .isEqualTo(ForStOptions.WRITE_BUFFER_RATIO.defaultValue()); ForStMemoryConfiguration configured = ForStMemoryConfiguration.fromOtherAndConfiguration( memSettings, new Configuration()); - assertTrue(configured.isUsingManagedMemory()); - assertFalse(configured.isUsingFixedMemoryPerSlot()); - assertEquals( - ForStOptions.HIGH_PRIORITY_POOL_RATIO.defaultValue(), - configured.getHighPriorityPoolRatio(), - 0.0); - assertEquals( - ForStOptions.WRITE_BUFFER_RATIO.defaultValue(), - configured.getWriteBufferRatio(), - 0.0); + assertThat(configured.isUsingManagedMemory()).isTrue(); + assertThat(configured.isUsingFixedMemoryPerSlot()).isFalse(); + assertThat(configured.getHighPriorityPoolRatio()) + .isEqualTo(ForStOptions.HIGH_PRIORITY_POOL_RATIO.defaultValue()); + assertThat(configured.getWriteBufferRatio()) + .isEqualTo(ForStOptions.WRITE_BUFFER_RATIO.defaultValue()); } @Test - public void testConfigureManagedMemory() { + void testConfigureManagedMemory() { final Configuration config = new Configuration(); config.set(ForStOptions.USE_MANAGED_MEMORY, true); @@ -744,11 +734,11 @@ public void testConfigureManagedMemory() { ForStMemoryConfiguration.fromOtherAndConfiguration( new ForStMemoryConfiguration(), config); - assertTrue(memSettings.isUsingManagedMemory()); + assertThat(memSettings.isUsingManagedMemory()).isTrue(); } @Test - public void testConfigureIllegalMemoryControlParameters() { + void testConfigureIllegalMemoryControlParameters() { ForStMemoryConfiguration memSettings = new ForStMemoryConfiguration(); verifySetParameter(() -> memSettings.setFixedMemoryPerSlot("-1B")); @@ -761,21 +751,17 @@ public void testConfigureIllegalMemoryControlParameters() { memSettings.setWriteBufferRatio(0.6); memSettings.setHighPriorityPoolRatio(0.6); - try { - // sum of writeBufferRatio and highPriPoolRatio larger than 1.0 - memSettings.validate(); - fail("Expected an IllegalArgumentException."); - } catch (IllegalArgumentException expected) { - // expected exception - } + // sum of writeBufferRatio and highPriPoolRatio larger than 1.0 + assertThatThrownBy(memSettings::validate).isInstanceOf(IllegalArgumentException.class); } @Test - public void testPrimaryDirectory() throws Exception { + void testPrimaryDirectory() throws Exception { FileSystem.initialize(new Configuration(), null); Configuration configuration = new Configuration(); configuration.set( - ForStOptions.PRIMARY_DIRECTORY, tempFolder.newFolder().toURI().toString()); + ForStOptions.PRIMARY_DIRECTORY, + TempDirUtils.newFolder(tempFolder).toURI().toString()); ForStStateBackend forStStateBackend = new ForStStateBackend().configure(configuration, null); ForStKeyedStateBackend keyedBackend = null; @@ -783,13 +769,10 @@ public void testPrimaryDirectory() throws Exception { keyedBackend = createKeyedStateBackend( forStStateBackend, - getMockEnvironment(tempFolder.newFolder()), + getMockEnvironment(TempDirUtils.newFolder(tempFolder)), IntSerializer.INSTANCE); - assertTrue( - keyedBackend - .getRemoteBasePath() - .toString() - .startsWith(configuration.get(ForStOptions.PRIMARY_DIRECTORY))); + assertThat(keyedBackend.getRemoteBasePath().toString()) + .startsWith(configuration.get(ForStOptions.PRIMARY_DIRECTORY)); } finally { if (keyedBackend != null) { keyedBackend.dispose(); @@ -799,14 +782,15 @@ public void testPrimaryDirectory() throws Exception { } @Test - public void testSupportSavepoint() { + void testSupportSavepoint() { ForStStateBackend forStStateBackend = new ForStStateBackend(); - assertFalse(forStStateBackend.supportsSavepointFormat(SavepointFormatType.CANONICAL)); - assertTrue(forStStateBackend.supportsSavepointFormat(SavepointFormatType.NATIVE)); + assertThat(forStStateBackend.supportsSavepointFormat(SavepointFormatType.CANONICAL)) + .isFalse(); + assertThat(forStStateBackend.supportsSavepointFormat(SavepointFormatType.NATIVE)).isTrue(); } @Test - public void testConfigurePeriodicCompactionTime() throws Exception { + void testConfigurePeriodicCompactionTime() throws Exception { ForStStateBackend forStStateBackend = new ForStStateBackend(); Configuration configuration = new Configuration(); configuration.setString( @@ -814,12 +798,12 @@ public void testConfigurePeriodicCompactionTime() throws Exception { forStStateBackend = forStStateBackend.configure(configuration, getClass().getClassLoader()); try (ForStResourceContainer resourceContainer = forStStateBackend.createOptionsAndResourceContainer(null)) { - assertEquals(Duration.ofDays(1), resourceContainer.getPeriodicCompactionTime()); + assertThat(resourceContainer.getPeriodicCompactionTime()).isEqualTo(Duration.ofDays(1)); } } @Test - public void testConfigureQueryTimeAfterNumEntries() throws Exception { + void testConfigureQueryTimeAfterNumEntries() throws Exception { ForStStateBackend forStStateBackend = new ForStStateBackend(); Configuration configuration = new Configuration(); configuration.setString( @@ -827,17 +811,12 @@ public void testConfigureQueryTimeAfterNumEntries() throws Exception { forStStateBackend = forStStateBackend.configure(configuration, getClass().getClassLoader()); try (ForStResourceContainer resourceContainer = forStStateBackend.createOptionsAndResourceContainer(null)) { - assertEquals(100L, resourceContainer.getQueryTimeAfterNumEntries().longValue()); + assertThat(resourceContainer.getQueryTimeAfterNumEntries().longValue()).isEqualTo(100L); } } private void verifySetParameter(Runnable setter) { - try { - setter.run(); - fail("No expected IllegalArgumentException."); - } catch (IllegalArgumentException expected) { - // expected exception - } + assertThatThrownBy(setter::run).isInstanceOf(IllegalArgumentException.class); } // ------------------------------------------------------------------------ @@ -867,12 +846,8 @@ private void verifyIllegalArgument(ConfigOption configOption, String configVa configuration.setString(configOption.key(), configValue); ForStStateBackend stateBackend = new ForStStateBackend(); - try { - stateBackend.configure(configuration, null); - fail("Not throwing expected IllegalArgumentException."); - } catch (IllegalArgumentException e) { - // ignored - } + assertThatThrownBy(() -> stateBackend.configure(configuration, null)) + .isInstanceOf(IllegalArgumentException.class); } /** An implementation of options factory for testing. */ diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendFactoryTest.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendFactoryTest.java index e799f8f85fb05..2c1f9e84c478c 100644 --- a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendFactoryTest.java +++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendFactoryTest.java @@ -23,23 +23,20 @@ import org.apache.flink.configuration.StateBackendOptions; import org.apache.flink.runtime.state.StateBackend; import org.apache.flink.runtime.state.StateBackendLoader; +import org.apache.flink.testutils.junit.utils.TempDirUtils; -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.io.File; -import java.util.Arrays; -import java.util.HashSet; +import java.nio.file.Path; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertTrue; +import static org.assertj.core.api.Assertions.assertThat; /** Tests for the ForStStateBackendFactory. */ -public class ForStStateBackendFactoryTest { +class ForStStateBackendFactoryTest { - @Rule public final TemporaryFolder tmp = new TemporaryFolder(); + @TempDir Path tmp; private final ClassLoader cl = getClass().getClassLoader(); @@ -48,13 +45,13 @@ public class ForStStateBackendFactoryTest { // ------------------------------------------------------------------------ @Test - public void testEmbeddedFactoryName() { + void testEmbeddedFactoryName() { // construct the name such that it will not be automatically adjusted on refactorings String factoryName = "org.apache.flink.state.forst.For"; factoryName += "StStateBackendFactory"; // !!! if this fails, the code in StateBackendLoader must be adjusted - assertEquals(factoryName, ForStStateBackendFactory.class.getName()); + assertThat(ForStStateBackendFactory.class.getName()).isEqualTo(factoryName); } /** @@ -62,9 +59,9 @@ public void testEmbeddedFactoryName() { * configuration. */ @Test - public void testLoadForStStateBackend() throws Exception { - final String localDir1 = tmp.newFolder().getAbsolutePath(); - final String localDir2 = tmp.newFolder().getAbsolutePath(); + void testLoadForStStateBackend() throws Exception { + final String localDir1 = TempDirUtils.newFolder(tmp).getAbsolutePath(); + final String localDir2 = TempDirUtils.newFolder(tmp).getAbsolutePath(); final String localDirs = localDir1 + File.pathSeparator + localDir2; final boolean incremental = !CheckpointingOptions.INCREMENTAL_CHECKPOINTS.defaultValue(); @@ -81,8 +78,8 @@ public void testLoadForStStateBackend() throws Exception { StateBackend backend1 = StateBackendLoader.loadStateBackendFromConfig(config1, cl, null); StateBackend backend2 = StateBackendLoader.loadStateBackendFromConfig(config2, cl, null); - assertTrue(backend1 instanceof ForStStateBackend); - assertTrue(backend2 instanceof ForStStateBackend); + assertThat(backend1).isInstanceOf(ForStStateBackend.class); + assertThat(backend2).isInstanceOf(ForStStateBackend.class); ForStStateBackend fs1 = (ForStStateBackend) backend1; ForStStateBackend fs2 = (ForStStateBackend) backend1; @@ -97,11 +94,11 @@ public void testLoadForStStateBackend() throws Exception { * parameters over configuration-defined parameters. */ @Test - public void testLoadForStStateBackendMixed() throws Exception { - final String localDir1 = tmp.newFolder().getAbsolutePath(); - final String localDir2 = tmp.newFolder().getAbsolutePath(); - final String localDir3 = tmp.newFolder().getAbsolutePath(); - final String localDir4 = tmp.newFolder().getAbsolutePath(); + void testLoadForStStateBackendMixed() throws Exception { + final String localDir1 = TempDirUtils.newFolder(tmp).getAbsolutePath(); + final String localDir2 = TempDirUtils.newFolder(tmp).getAbsolutePath(); + final String localDir3 = TempDirUtils.newFolder(tmp).getAbsolutePath(); + final String localDir4 = TempDirUtils.newFolder(tmp).getAbsolutePath(); final ForStStateBackend backend = new ForStStateBackend(); backend.setLocalDbStoragePaths(localDir1, localDir2); @@ -115,7 +112,7 @@ public void testLoadForStStateBackendMixed() throws Exception { final StateBackend loadedBackend = StateBackendLoader.fromApplicationOrConfigOrDefault( backend, new Configuration(), config, cl, null); - assertTrue(loadedBackend instanceof ForStStateBackend); + assertThat(loadedBackend).isInstanceOf(ForStStateBackend.class); final ForStStateBackend loadedRocks = (ForStStateBackend) loadedBackend; @@ -125,15 +122,9 @@ public void testLoadForStStateBackendMixed() throws Exception { // ------------------------------------------------------------------------ private static void checkPaths(String[] pathsArray, String... paths) { - assertNotNull(pathsArray); - assertNotNull(paths); - - assertEquals(pathsArray.length, paths.length); - - HashSet pathsSet = new HashSet<>(Arrays.asList(pathsArray)); - - for (String path : paths) { - assertTrue(pathsSet.contains(path)); - } + assertThat(pathsArray).isNotNull(); + assertThat(paths).isNotNull(); + assertThat(pathsArray).hasSize(paths.length); + assertThat(pathsArray).contains(paths); } }