This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 6627acb4231 [FLINK-39767][tests][JUnit5 migration] Module:
flink-statebackend-forst
6627acb4231 is described below
commit 6627acb4231931e0fa54f8d4303e2a83e763c9e8
Author: Purushottam Sinha <[email protected]>
AuthorDate: Sat Jun 20 13:11:11 2026 +0530
[FLINK-39767][tests][JUnit5 migration] Module: flink-statebackend-forst
---
.../ForStIncrementalCheckpointRescalingTest.java | 191 ++++++------
.../apache/flink/state/forst/ForStInitITCase.java | 22 +-
.../forst/ForStMemoryControllerUtilsTest.java | 77 +++--
.../state/forst/ForStMultiClassLoaderTest.java | 21 +-
.../state/forst/ForStNativeMetricOptionsTest.java | 25 +-
.../state/forst/ForStOperationsUtilsTest.java | 31 +-
.../state/forst/ForStResourceContainerTest.java | 124 ++++----
.../state/forst/ForStStateBackendConfigTest.java | 337 ++++++++++-----------
.../state/forst/ForStStateBackendFactoryTest.java | 56 ++--
9 files changed, 410 insertions(+), 474 deletions(-)
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 94b7186dcf8..4fb2f3f7e34 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
@@ -35,35 +35,34 @@ import
org.apache.flink.streaming.api.operators.KeyedProcessOperator;
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.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 java.util.Arrays;
-import java.util.Collection;
+
+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.nio.file.Path;
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 Path rootFolder;
- @Parameterized.Parameters(name = "useIngestDbRestoreMode: {0}")
- public static Collection<Boolean> parameters() {
- return Arrays.asList(false, true);
+ @Parameters(name = "useIngestDbRestoreMode: {0}")
+ public static List<Boolean> parameters() {
+ return List.of(false, true);
}
- @Parameterized.Parameter public boolean useIngestDbRestoreMode;
+ @Parameter public boolean useIngestDbRestoreMode;
private final int maxParallelism = 10;
@@ -71,73 +70,73 @@ public class ForStIncrementalCheckpointRescalingTest
extends TestLogger {
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))
+ .isZero(); // 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))
+ .isOne(); // 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
// <---------------------------------------
@@ -148,8 +147,7 @@ public class ForStIncrementalCheckpointRescalingTest
extends TestLogger {
getHarnessTest(keySelector, maxParallelism, 1, 0)) {
harness.setStateBackend(getStateBackend());
harness.setCheckpointStorage(
- new FileSystemCheckpointStorage(
- "file://" +
rootFolder.newFolder().getAbsolutePath()));
+ new FileSystemCheckpointStorage("file://" +
rootFolder.toAbsolutePath()));
harness.open();
validHarnessResult(harness, 1, records);
@@ -181,24 +179,22 @@ public class ForStIncrementalCheckpointRescalingTest
extends TestLogger {
// 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()));
+ new FileSystemCheckpointStorage("file://" +
rootFolder.toAbsolutePath()));
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()));
+ new FileSystemCheckpointStorage("file://" +
rootFolder.toAbsolutePath()));
harness2[1].setup();
harness2[1].initializeState(initState2);
harness2[1].open();
@@ -244,36 +240,33 @@ public class ForStIncrementalCheckpointRescalingTest
extends TestLogger {
// 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()));
+ new FileSystemCheckpointStorage("file://" +
rootFolder.toAbsolutePath()));
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()));
+ new FileSystemCheckpointStorage("file://" +
rootFolder.toAbsolutePath()));
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()));
+ new FileSystemCheckpointStorage("file://" +
rootFolder.toAbsolutePath()));
harness3[2].setup();
harness3[2].initializeState(initState3);
harness3[2].open();
@@ -286,9 +279,9 @@ public class ForStIncrementalCheckpointRescalingTest
extends TestLogger {
}
}
- @Test
+ @TestTemplate
@SuppressWarnings("unchecked")
- public void testScalingDown() throws Exception {
+ void testScalingDown() throws Exception {
// -----------------------------------------> test with initial
parallelism 3
// <---------------------------------------
@@ -303,32 +296,29 @@ public class ForStIncrementalCheckpointRescalingTest
extends TestLogger {
// 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()));
+ new FileSystemCheckpointStorage("file://" +
rootFolder.toAbsolutePath()));
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()));
+ new FileSystemCheckpointStorage("file://" +
rootFolder.toAbsolutePath()));
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()));
+ new FileSystemCheckpointStorage("file://" +
rootFolder.toAbsolutePath()));
harness3[2].open();
validHarnessResult(harness3[0], 1, records[0], records[1],
records[2], records[3]);
@@ -371,12 +361,11 @@ public class ForStIncrementalCheckpointRescalingTest
extends TestLogger {
// 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()));
+ new FileSystemCheckpointStorage("file://" +
rootFolder.toAbsolutePath()));
harness2[0].setup();
harness2[0].initializeState(initState1);
harness2[0].open();
@@ -384,12 +373,11 @@ public class ForStIncrementalCheckpointRescalingTest
extends TestLogger {
// 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()));
+ new FileSystemCheckpointStorage("file://" +
rootFolder.toAbsolutePath()));
harness2[1].setup();
harness2[1].initializeState(initState2);
harness2[1].open();
@@ -422,8 +410,7 @@ public class ForStIncrementalCheckpointRescalingTest
extends TestLogger {
// without any clipping.
harness.setStateBackend(getStateBackend());
harness.setCheckpointStorage(
- new FileSystemCheckpointStorage(
- "file://" +
rootFolder.newFolder().getAbsolutePath()));
+ new FileSystemCheckpointStorage("file://" +
rootFolder.toAbsolutePath()));
harness.setup();
harness.initializeState(initState1);
harness.open();
@@ -450,8 +437,8 @@ public class ForStIncrementalCheckpointRescalingTest
extends TestLogger {
for (String record : records) {
harness.processElement(new StreamRecord<>(record, 1));
StreamRecord<Integer> outputRecord = (StreamRecord<Integer>)
harness.getOutput().poll();
- Assert.assertNotNull(outputRecord);
- Assert.assertEquals(expectedValue, outputRecord.getValue());
+ assertThat(outputRecord).isNotNull();
+ assertThat(outputRecord.getValue()).isEqualTo(expectedValue);
}
}
@@ -470,7 +457,7 @@ public class ForStIncrementalCheckpointRescalingTest
extends TestLogger {
subtaskIdx);
}
- private StateBackend getStateBackend() throws Exception {
+ private StateBackend getStateBackend() {
ForStStateBackend forStStateBackend = new ForStStateBackend();
Configuration configuration = new Configuration();
configuration.set(USE_INGEST_DB_RESTORE_MODE, useIngestDbRestoreMode);
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 a23762d0169..e4b3a7e083f 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
@@ -21,33 +21,29 @@ package org.apache.flink.state.forst;
import org.apache.flink.runtime.operators.testutils.ExpectedTestException;
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 {
-
- @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
+class ForStInitITCase {
/**
* 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(@TempDir File tempFolder) {
ForStStateBackend.setForStInitialized(false);
- File tempFolder = temporaryFolder.newFolder();
try {
ForStStateBackend.ensureForStIsLoaded(
tempFolder.getAbsolutePath(),
@@ -60,7 +56,7 @@ public class ForStInitITCase {
// 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 e29ce68747d..5d0b79ac540 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
@@ -21,31 +21,25 @@ package org.apache.flink.state.forst;
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();
-
- @Before
- public void ensureRocksDbNativeLibraryLoaded() throws IOException {
- NativeLibraryLoader.getInstance()
- .loadLibrary(temporaryFolder.newFolder().getAbsolutePath());
+ @BeforeEach
+ void ensureRocksDbNativeLibraryLoaded(@TempDir Path temporaryFolder)
throws IOException {
+
NativeLibraryLoader.getInstance().loadLibrary(temporaryFolder.toFile().getAbsolutePath());
}
@Test
- public void testCreateSharedResourcesWithExpectedCapacity() {
+ void testCreateSharedResourcesWithExpectedCapacity() {
long totalMemorySize = 2048L;
double writeBufferRatio = 0.5;
double highPriPoolRatio = 0.1;
@@ -60,50 +54,55 @@ public class ForStMemoryControllerUtilsTest {
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 2753ea63320..423d33e8997 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
@@ -22,28 +22,26 @@ 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 {
-
- @Rule public final TemporaryFolder tmp = new TemporaryFolder();
+class ForStMultiClassLoaderTest {
@Test
- public void testTwoSeparateClassLoaders() throws Exception {
+ void testTwoSeparateClassLoaders(@TempDir Path tmp) throws Exception {
// collect the libraries / class folders with ForSt related code: the
state backend and
// ForSt itself
final URL codePath1 =
@@ -70,13 +68,14 @@ public class ForStMultiClassLoaderTest {
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 = tmp.toFile().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 a7f9cd22403..c1fa565ab7a 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 @@ package org.apache.flink.state.forst;
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 class ForStNativeMetricOptionsTest {
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 abbb63c1c74..29af4375646 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
@@ -24,35 +24,31 @@ 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.nio.file.Path;
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();
-
- @BeforeClass
- public static void loadRocksLibrary() throws Exception {
-
NativeLibraryLoader.getInstance().loadLibrary(TMP_DIR.newFolder().getAbsolutePath());
+ @BeforeAll
+ static void loadRocksLibrary(@TempDir Path libDir) throws Exception {
+
NativeLibraryLoader.getInstance().loadLibrary(libDir.toFile().getAbsolutePath());
}
@Test
- public void testPathExceptionOnWindows() throws Exception {
+ void testPathExceptionOnWindows(@TempDir File folder) throws Exception {
assumeTrue(OperatingSystem.isWindows());
- final File folder = TMP_DIR.newFolder();
final File rocksDir =
new File(folder, getLongString(247 -
folder.getAbsolutePath().length()));
@@ -73,9 +69,8 @@ public class ForStOperationsUtilsTest {
// 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 9085f6a1fd2..e3d11ba31e1 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.memory.OpaqueMemoryResource;
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.RocksDB;
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.ArrayList;
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(@TempDir java.nio.file.Path
libDir)
+ throws IOException {
+
NativeLibraryLoader.getInstance().loadLibrary(libDir.toFile().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> dbOptions = new ArrayList<>(optionNumber);
@@ -92,7 +90,7 @@ public class ForStResourceContainerTest {
}
container.close();
for (DBOptions dbOption : dbOptions) {
- assertThat(dbOption.isOwningHandle(), is(false));
+ assertThat(dbOption.isOwningHandle()).isFalse();
}
}
@@ -103,13 +101,13 @@ public class ForStResourceContainerTest {
* @throws Exception if unexpected error happened.
*/
@Test
- public void testSharedResourcesAfterClose() throws Exception {
+ void testSharedResourcesAfterClose() throws Exception {
OpaqueMemoryResource<ForStSharedResources> 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 class ForStResourceContainerTest {
* @throws Exception if unexpected error happened.
*/
@Test
- public void testGetDbOptionsWithSharedResources() throws Exception {
+ void testGetDbOptionsWithSharedResources() throws Exception {
final int optionNumber = 20;
OpaqueMemoryResource<ForStSharedResources> sharedResources =
getSharedResources();
ForStResourceContainer container = new ForStResourceContainer(null,
sharedResources);
@@ -130,10 +128,9 @@ public class ForStResourceContainerTest {
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 class ForStResourceContainerTest {
* @throws Exception if unexpected error happened.
*/
@Test
- public void testGetColumnFamilyOptionsWithSharedResources() throws
Exception {
+ void testGetColumnFamilyOptionsWithSharedResources() throws Exception {
final int optionNumber = 20;
OpaqueMemoryResource<ForStSharedResources> sharedResources =
getSharedResources();
ForStResourceContainer container = new ForStResourceContainer(null,
sharedResources);
@@ -155,8 +152,9 @@ public class ForStResourceContainerTest {
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 @@ public class ForStResourceContainerTest {
}
@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> columnFamilyOptions = new
ArrayList<>(optionNumber);
@@ -228,12 +226,12 @@ public class ForStResourceContainerTest {
}
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 class ForStResourceContainerTest {
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 class ForStResourceContainerTest {
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 class ForStResourceContainerTest {
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 class ForStResourceContainerTest {
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<ColumnFamilyHandle> columnFamilyHandles = new ArrayList<>(1);
ArrayList<ColumnFamilyDescriptor> 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 624bbf0b577..1505057f8e2 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.KeyedStateBackendParametersImpl;
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.DBOptions;
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 java.util.Collections;
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 class ForStStateBackendConfigTest {
/** 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<Integer> 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 class ForStStateBackendConfigTest {
}
@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 class ForStStateBackendConfigTest {
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 @@ public class ForStStateBackendConfigTest {
final ForStStateBackend forStBackend = new ForStStateBackend();
forStBackend.setLocalDbStoragePath(configuredPath);
- final MockEnvironment env = getMockEnvironment(tempFolder.newFolder());
+ final MockEnvironment env =
getMockEnvironment(TempDirUtils.newFolder(tempFolder));
ForStKeyedStateBackend<Integer> 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 @@ public class ForStStateBackendConfigTest {
}
/** 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 class ForStStateBackendConfigTest {
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<Integer> keyedBackend =
createKeyedStateBackend(forStBackend, env,
IntSerializer.INSTANCE);
@@ -298,7 +297,7 @@ public class ForStStateBackendConfigTest {
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 class ForStStateBackendConfigTest {
}
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 class ForStStateBackendConfigTest {
* {@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 class ForStStateBackendConfigTest {
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 class ForStStateBackendConfigTest {
// ------------------------------------------------------------------------
@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 class ForStStateBackendConfigTest {
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 class ForStStateBackendConfigTest {
}
@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 class ForStStateBackendConfigTest {
// ForSt Options
// ------------------------------------------------------------------------
@Test
- public void testConfigurableOptionsFromConfig() throws Exception {
+ void testConfigurableOptionsFromConfig() throws Exception {
Configuration configuration = new Configuration();
// verify illegal configuration
@@ -537,40 +536,39 @@ public class ForStStateBackendConfigTest {
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 class ForStStateBackendConfigTest {
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 class ForStStateBackendConfigTest {
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 class ForStStateBackendConfigTest {
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 class ForStStateBackendConfigTest {
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 class ForStStateBackendConfigTest {
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 class ForStStateBackendConfigTest {
// ------------------------------------------------------------------------
@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 class ForStStateBackendConfigTest {
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 class ForStStateBackendConfigTest {
// ------------------------------------------------------------------------
@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 class ForStStateBackendConfigTest {
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 class ForStStateBackendConfigTest {
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<Integer> keyedBackend = null;
@@ -783,13 +769,10 @@ public class ForStStateBackendConfigTest {
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 class ForStStateBackendConfigTest {
}
@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 class ForStStateBackendConfigTest {
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 class ForStStateBackendConfigTest {
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 @@ public class ForStStateBackendConfigTest {
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 e799f8f85fb..3e6aaa901c3 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.Configuration;
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 class ForStStateBackendFactoryTest {
* 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 class ForStStateBackendFactoryTest {
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 class ForStStateBackendFactoryTest {
* 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 class ForStStateBackendFactoryTest {
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,8 @@ public class ForStStateBackendFactoryTest {
// ------------------------------------------------------------------------
private static void checkPaths(String[] pathsArray, String... paths) {
- assertNotNull(pathsArray);
- assertNotNull(paths);
-
- assertEquals(pathsArray.length, paths.length);
-
- HashSet<String> pathsSet = new HashSet<>(Arrays.asList(pathsArray));
-
- for (String path : paths) {
- assertTrue(pathsSet.contains(path));
- }
+ assertThat(pathsArray).isNotNull();
+ assertThat(paths).isNotNull();
+ assertThat(pathsArray).containsExactlyInAnyOrder(paths);
}
}
