diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/HeapTimersSnapshottingTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/HeapTimersSnapshottingTest.java index 81a4a209d61b43..40f5541b69a5c6 100644 --- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/HeapTimersSnapshottingTest.java +++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/HeapTimersSnapshottingTest.java @@ -31,12 +31,12 @@ import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness; import org.apache.flink.util.Collector; -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 static org.hamcrest.CoreMatchers.equalTo; -import static org.junit.Assert.assertThat; +import java.io.File; + +import static org.assertj.core.api.Assertions.assertThat; /** * The tests verify that {@link PriorityQueueStateType#HEAP heap timers} are not serialized into raw @@ -44,19 +44,19 @@ * timers still need to be serialized into the raw operator state because of RocksDB incremental * checkpoints. */ -public class HeapTimersSnapshottingTest { +class HeapTimersSnapshottingTest { - @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder(); + @TempDir private File temporaryFolder; @Test - public void testNotSerializingTimersInRawStateForSavepoints() throws Exception { + void testNotSerializingTimersInRawStateForSavepoints() throws Exception { try (KeyedOneInputStreamOperatorTestHarness testHarness = getTestHarness()) { EmbeddedRocksDBStateBackend backend = new EmbeddedRocksDBStateBackend(); backend.setPriorityQueueStateType(PriorityQueueStateType.HEAP); testHarness.setStateBackend(backend); testHarness.setCheckpointStorage( - new FileSystemCheckpointStorage(temporaryFolder.newFolder().toURI())); + new FileSystemCheckpointStorage(temporaryFolder.toURI())); testHarness.open(); testHarness.processElement(0, 0L); @@ -65,19 +65,19 @@ public void testNotSerializingTimersInRawStateForSavepoints() throws Exception { .snapshotWithLocalState( 0L, 1L, SavepointType.savepoint(SavepointFormatType.CANONICAL)) .getJobManagerOwnedState(); - assertThat(state.getRawKeyedState().isEmpty(), equalTo(true)); + assertThat(state.getRawKeyedState()).isEmpty(); } } @Test - public void testSerializingTimersInRawStateForCheckpoints() throws Exception { + void testSerializingTimersInRawStateForCheckpoints() throws Exception { try (KeyedOneInputStreamOperatorTestHarness testHarness = getTestHarness()) { EmbeddedRocksDBStateBackend backend = new EmbeddedRocksDBStateBackend(); backend.setPriorityQueueStateType(PriorityQueueStateType.HEAP); testHarness.setStateBackend(backend); testHarness.setCheckpointStorage( - new FileSystemCheckpointStorage(temporaryFolder.newFolder().toURI())); + new FileSystemCheckpointStorage(temporaryFolder.toURI())); testHarness.open(); testHarness.processElement(0, 0L); @@ -85,7 +85,7 @@ public void testSerializingTimersInRawStateForCheckpoints() throws Exception { testHarness .snapshotWithLocalState(0L, 1L, CheckpointType.CHECKPOINT) .getJobManagerOwnedState(); - assertThat(state.getRawKeyedState().isEmpty(), equalTo(false)); + assertThat(state.getRawKeyedState()).isNotEmpty(); } } diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBAsyncSnapshotTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBAsyncSnapshotTest.java index 6b7979864e2fae..8ae960a0bcfe78 100644 --- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBAsyncSnapshotTest.java +++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBAsyncSnapshotTest.java @@ -71,20 +71,15 @@ import org.apache.flink.streaming.runtime.tasks.OneInputStreamTaskTestHarness; import org.apache.flink.streaming.runtime.tasks.StreamMockEnvironment; import org.apache.flink.util.IOUtils; -import org.apache.flink.util.TestLogger; import org.apache.flink.util.concurrent.FutureUtils; -import org.assertj.core.api.Assertions; -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 javax.annotation.Nullable; import java.io.File; import java.io.IOException; -import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.Map; @@ -101,15 +96,12 @@ import static org.apache.flink.runtime.state.FullSnapshotUtil.clearMetaDataFollowsFlag; import static org.apache.flink.runtime.state.FullSnapshotUtil.hasMetaDataFollowsFlag; import static org.apache.flink.runtime.state.FullSnapshotUtil.setMetaDataFollowsFlagInKey; -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; /** Tests for asynchronous RocksDB Key/Value state checkpoints. */ @SuppressWarnings("serial") -public class RocksDBAsyncSnapshotTest extends TestLogger { - - /** Temporary fold for test. */ - @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder(); +class RocksDBAsyncSnapshotTest { /** * This ensures that asynchronous state handles are actually materialized asynchronously. @@ -119,7 +111,7 @@ public class RocksDBAsyncSnapshotTest extends TestLogger { * simply lock forever. */ @Test - public void testFullyAsyncSnapshot() throws Exception { + void testFullyAsyncSnapshot(@TempDir File dbDir) throws Exception { final OneInputStreamTaskTestHarness testHarness = new OneInputStreamTaskTestHarness<>( @@ -139,8 +131,6 @@ public String getKey(String value) throws Exception { StreamConfig streamConfig = testHarness.getStreamConfig(); - File dbDir = temporaryFolder.newFolder(); - EmbeddedRocksDBStateBackend backend = new EmbeddedRocksDBStateBackend(); backend.setDbStoragePath(dbDir.getAbsolutePath()); @@ -181,7 +171,7 @@ public void acknowledgeCheckpoint( } // should be one k/v state - assertTrue(hasManagedKeyedState); + assertThat(hasManagedKeyedState).isTrue(); // we now know that the checkpoint went through ensureCheckpointLatch.trigger(); @@ -254,12 +244,10 @@ public void reportInitializationMetrics( ExecutorService threadPool = task.getAsyncOperationsThreadPool(); threadPool.shutdown(); - Assert.assertTrue(threadPool.awaitTermination(60_000, TimeUnit.MILLISECONDS)); + assertThat(threadPool.awaitTermination(60_000, TimeUnit.MILLISECONDS)).isTrue(); testHarness.waitForTaskCompletion(); - if (errorRef.get() != null) { - fail("Unexpected exception during execution."); - } + assertThat(errorRef.get()).isNull(); } /** @@ -267,7 +255,7 @@ public void reportInitializationMetrics( * block. */ @Test - public void testCancelFullyAsyncCheckpoints() throws Exception { + void testCancelFullyAsyncCheckpoints(@TempDir File dbDir) throws Exception { final OneInputStreamTaskTestHarness testHarness = new OneInputStreamTaskTestHarness<>( OneInputStreamTask::new, @@ -280,8 +268,6 @@ public void testCancelFullyAsyncCheckpoints() throws Exception { StreamConfig streamConfig = testHarness.getStreamConfig(); - File dbDir = temporaryFolder.newFolder(); - final EmbeddedRocksDBStateBackend.PriorityQueueStateType timerServicePriorityQueueType = RocksDBOptions.TIMER_SERVICE_FACTORY.defaultValue(); @@ -371,30 +357,18 @@ public CheckpointStateOutputStream createCheckpointStateOutputStream( ExecutorService threadPool = task.getAsyncOperationsThreadPool(); threadPool.shutdown(); - Assert.assertTrue(threadPool.awaitTermination(60_000, TimeUnit.MILLISECONDS)); + assertThat(threadPool.awaitTermination(60_000, TimeUnit.MILLISECONDS)).isTrue(); Set createdStreams = blockerCheckpointStreamFactory.getAllCreatedStreams(); - for (BlockingCheckpointOutputStream stream : createdStreams) { - Assert.assertTrue( - "Not all of the " - + createdStreams.size() - + " created streams have been closed.", - stream.isClosed()); - } + assertThat(createdStreams) + .as("Not all of the %d created streams have been closed.", createdStreams.size()) + .allMatch(BlockingCheckpointOutputStream::isClosed); - try { - testHarness.waitForTaskCompletion(); - fail("Operation completed. Cancel failed."); - } catch (Exception expected) { - - Throwable cause = expected.getCause(); - - if (!(cause instanceof CancelTaskException)) { - fail("Unexpected exception: " + expected); - } - } + assertThatThrownBy( + testHarness::waitForTaskCompletion, "Operation completed. Cancel failed.") + .hasCauseInstanceOf(CancelTaskException.class); } /** @@ -402,7 +376,7 @@ public CheckpointStateOutputStream createCheckpointStateOutputStream( * procedure. */ @Test - public void testCleanupOfSnapshotsInFailureCase() throws Exception { + void testCleanupOfSnapshotsInFailureCase(@TempDir File dbDir) throws Exception { long checkpointId = 1L; long timestamp = 42L; @@ -413,7 +387,7 @@ public void testCleanupOfSnapshotsInFailureCase() throws Exception { EmbeddedRocksDBStateBackend backend = new EmbeddedRocksDBStateBackend(); - backend.setDbStoragePath(temporaryFolder.newFolder().toURI().toString()); + backend.setDbStoragePath(dbDir.toURI().toString()); CheckpointableKeyedStateBackend keyedStateBackend = backend.createKeyedStateBackend( @@ -446,14 +420,12 @@ public void testCleanupOfSnapshotsInFailureCase() throws Exception { new TestCheckpointStreamFactory(() -> outputStream), CheckpointOptions.forCheckpointWithDefaultLocation()); - try { - FutureUtils.runIfNotDoneAndGet(snapshotFuture); - fail("Expected an exception to be thrown here."); - } catch (ExecutionException e) { - Assert.assertEquals(testException, e.getCause()); - } + assertThatThrownBy(() -> FutureUtils.runIfNotDoneAndGet(snapshotFuture)) + .isInstanceOf(ExecutionException.class) + .cause() + .isSameAs(testException); - Assertions.assertThat(outputStream.isCloseCalled()).isEqualTo(true); + assertThat(outputStream.isCloseCalled()).isTrue(); } finally { IOUtils.closeQuietly(keyedStateBackend); keyedStateBackend.dispose(); @@ -462,23 +434,23 @@ public void testCleanupOfSnapshotsInFailureCase() throws Exception { } @Test - public void testConsistentSnapshotSerializationFlagsAndMasks() { + void testConsistentSnapshotSerializationFlagsAndMasks() { - Assert.assertEquals(0xFFFF, END_OF_KEY_GROUP_MARK); - Assert.assertEquals(0x80, FIRST_BIT_IN_BYTE_MASK); + assertThat(END_OF_KEY_GROUP_MARK).isEqualTo(0xFFFF); + assertThat(FIRST_BIT_IN_BYTE_MASK).isEqualTo(0x80); byte[] expectedKey = new byte[] {42, 42}; byte[] modKey = expectedKey.clone(); - Assert.assertFalse(hasMetaDataFollowsFlag(modKey)); + assertThat(hasMetaDataFollowsFlag(modKey)).isFalse(); setMetaDataFollowsFlagInKey(modKey); - Assert.assertTrue(hasMetaDataFollowsFlag(modKey)); + assertThat(hasMetaDataFollowsFlag(modKey)).isTrue(); clearMetaDataFollowsFlag(modKey); - Assert.assertFalse(hasMetaDataFollowsFlag(modKey)); + assertThat(hasMetaDataFollowsFlag(modKey)).isFalse(); - Assert.assertTrue(Arrays.equals(expectedKey, modKey)); + assertThat(modKey).isEqualTo(expectedKey); } // ------------------------------------------------------------------------ diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBExtension.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBExtension.java index 71b100803a0ef5..4309706901b7df 100644 --- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBExtension.java +++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBExtension.java @@ -18,13 +18,13 @@ package org.apache.flink.state.rocksdb; +import org.apache.flink.util.FileUtils; import org.apache.flink.util.FlinkRuntimeException; import org.apache.flink.util.IOUtils; import org.junit.jupiter.api.extension.AfterEachCallback; import org.junit.jupiter.api.extension.BeforeEachCallback; import org.junit.jupiter.api.extension.ExtensionContext; -import org.junit.rules.TemporaryFolder; import org.rocksdb.ColumnFamilyDescriptor; import org.rocksdb.ColumnFamilyHandle; import org.rocksdb.ColumnFamilyOptions; @@ -39,7 +39,8 @@ import javax.annotation.Nonnull; -import java.io.File; +import java.nio.file.Files; +import java.nio.file.Path; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; @@ -55,7 +56,7 @@ public class RocksDBExtension implements BeforeEachCallback, AfterEachCallback { private final boolean enableStatistics; /** Temporary folder that provides the working directory for the RocksDB instance. */ - private TemporaryFolder temporaryFolder; + private Path temporaryFolder; /** The options for the RocksDB instance. */ private DBOptions dbOptions; @@ -169,9 +170,8 @@ public ColumnFamilyHandle createNewColumnFamily(String name) { } public void before() throws Exception { - this.temporaryFolder = new TemporaryFolder(); - this.temporaryFolder.create(); - final File rocksFolder = temporaryFolder.newFolder(); + this.temporaryFolder = Files.createTempDirectory("rocksdb-extension"); + final Path rocksFolder = Files.createDirectory(temporaryFolder.resolve("db")); this.dbOptions = optionsFactory .createDBOptions( @@ -195,7 +195,7 @@ public void before() throws Exception { this.rocksDB = RocksDB.open( dbOptions, - rocksFolder.getAbsolutePath(), + rocksFolder.toAbsolutePath().toString(), Collections.singletonList( new ColumnFamilyDescriptor( "default".getBytes(), columnFamilyOptions)), @@ -215,7 +215,7 @@ public void after() throws Exception { IOUtils.closeQuietly(this.columnFamilyOptions); IOUtils.closeQuietly(this.dbOptions); handlesToClose.forEach(IOUtils::closeQuietly); - temporaryFolder.delete(); + FileUtils.deleteDirectoryQuietly(temporaryFolder.toFile()); } @Override diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBIncrementalCheckpointUtilsTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBIncrementalCheckpointUtilsTest.java index e5eae90abbb834..7123736688d680 100644 --- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBIncrementalCheckpointUtilsTest.java +++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBIncrementalCheckpointUtilsTest.java @@ -22,33 +22,33 @@ import org.apache.flink.runtime.state.CompositeKeySerializationUtils; import org.apache.flink.runtime.state.KeyGroupRange; import org.apache.flink.runtime.state.KeyedStateHandle; -import org.apache.flink.util.TestLogger; +import org.apache.flink.testutils.junit.utils.TempDirUtils; -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 org.rocksdb.ColumnFamilyDescriptor; import org.rocksdb.ColumnFamilyHandle; import org.rocksdb.RocksDB; import org.rocksdb.RocksDBException; import java.io.IOException; +import java.nio.file.Path; import java.util.ArrayList; import java.util.Collections; import java.util.List; import static org.apache.flink.state.rocksdb.RocksDBConfigurableOptions.RESTORE_OVERLAP_FRACTION_THRESHOLD; +import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; /** Tests to guard {@link RocksDBIncrementalCheckpointUtils}. */ -public class RocksDBIncrementalCheckpointUtilsTest extends TestLogger { +class RocksDBIncrementalCheckpointUtilsTest { - @Rule public final TemporaryFolder tmp = new TemporaryFolder(); + @TempDir private Path tmp; @Test - public void testClipDBWithKeyGroupRange() throws Exception { + void testClipDBWithKeyGroupRange() throws Exception { testClipDBWithKeyGroupRangeHelper(new KeyGroupRange(0, 1), new KeyGroupRange(0, 2), 1); @@ -82,7 +82,7 @@ public void testClipDBWithKeyGroupRange() throws Exception { } @Test - public void testChooseTheBestStateHandleForInitial() { + void testChooseTheBestStateHandleForInitial() { List keyedStateHandles = new ArrayList<>(3); @@ -100,40 +100,40 @@ public void testChooseTheBestStateHandleForInitial() { // this should choose keyedStateHandle2, because keyedStateHandle2's key-group range // satisfies the overlap fraction demand. - Assert.assertEquals( - keyedStateHandle2, - RocksDBIncrementalCheckpointUtils.chooseTheBestStateHandleForInitial( - keyedStateHandles, - new KeyGroupRange(3, 6), - RESTORE_OVERLAP_FRACTION_THRESHOLD.defaultValue())); + assertThat( + RocksDBIncrementalCheckpointUtils.chooseTheBestStateHandleForInitial( + keyedStateHandles, + new KeyGroupRange(3, 6), + RESTORE_OVERLAP_FRACTION_THRESHOLD.defaultValue())) + .isEqualTo(keyedStateHandle2); // both keyedStateHandle2 & keyedStateHandle3's key-group range satisfies the overlap // fraction, but keyedStateHandle3's key group range is better. - Assert.assertEquals( - keyedStateHandle3, - RocksDBIncrementalCheckpointUtils.chooseTheBestStateHandleForInitial( - keyedStateHandles, - new KeyGroupRange(5, 12), - RESTORE_OVERLAP_FRACTION_THRESHOLD.defaultValue())); + assertThat( + RocksDBIncrementalCheckpointUtils.chooseTheBestStateHandleForInitial( + keyedStateHandles, + new KeyGroupRange(5, 12), + RESTORE_OVERLAP_FRACTION_THRESHOLD.defaultValue())) + .isEqualTo(keyedStateHandle3); // The intersect key group number of keyedStateHandle2 & keyedStateHandle3's with [4, 11] // are 4. But the over fraction of keyedStateHandle2 is better. - Assert.assertEquals( - keyedStateHandle2, - RocksDBIncrementalCheckpointUtils.chooseTheBestStateHandleForInitial( - keyedStateHandles, - new KeyGroupRange(4, 11), - RESTORE_OVERLAP_FRACTION_THRESHOLD.defaultValue())); + assertThat( + RocksDBIncrementalCheckpointUtils.chooseTheBestStateHandleForInitial( + keyedStateHandles, + new KeyGroupRange(4, 11), + RESTORE_OVERLAP_FRACTION_THRESHOLD.defaultValue())) + .isEqualTo(keyedStateHandle2); // both keyedStateHandle2 & keyedStateHandle3's key-group range are covered by [3, 12], // but this should choose the keyedStateHandle3, because keyedStateHandle3's key-group is // bigger than keyedStateHandle2. - Assert.assertEquals( - keyedStateHandle3, - RocksDBIncrementalCheckpointUtils.chooseTheBestStateHandleForInitial( - keyedStateHandles, - new KeyGroupRange(3, 12), - RESTORE_OVERLAP_FRACTION_THRESHOLD.defaultValue())); + assertThat( + RocksDBIncrementalCheckpointUtils.chooseTheBestStateHandleForInitial( + keyedStateHandles, + new KeyGroupRange(3, 12), + RESTORE_OVERLAP_FRACTION_THRESHOLD.defaultValue())) + .isEqualTo(keyedStateHandle3); } private void testClipDBWithKeyGroupRangeHelper( @@ -142,7 +142,7 @@ private void testClipDBWithKeyGroupRangeHelper( int keyGroupPrefixBytes) throws RocksDBException, IOException { - try (RocksDB rocksDB = RocksDB.open(tmp.newFolder().getAbsolutePath()); + try (RocksDB rocksDB = RocksDB.open(TempDirUtils.newFolder(tmp).getAbsolutePath()); ColumnFamilyHandle columnFamilyHandle = rocksDB.createColumnFamily(new ColumnFamilyDescriptor("test".getBytes()))) { @@ -172,7 +172,7 @@ private void testClipDBWithKeyGroupRangeHelper( CompositeKeySerializationUtils.writeKey( j, IntSerializer.INSTANCE, outputView, false); byte[] value = rocksDB.get(columnFamilyHandle, outputView.getCopyOfBuffer()); - Assert.assertEquals(String.valueOf(j), new String(value)); + assertThat(new String(value)).isEqualTo(String.valueOf(j)); } } @@ -193,9 +193,9 @@ private void testClipDBWithKeyGroupRangeHelper( j, IntSerializer.INSTANCE, outputView, false); byte[] value = rocksDB.get(columnFamilyHandle, outputView.getCopyOfBuffer()); if (targetGroupRange.contains(i)) { - Assert.assertEquals(String.valueOf(j), new String(value)); + assertThat(new String(value)).isEqualTo(String.valueOf(j)); } else { - Assert.assertNull(value); + assertThat(value).isNull(); } } } diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBInitITCase.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBInitITCase.java index 52c1410cb457c4..29ee56a8d1a31d 100644 --- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBInitITCase.java +++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBInitITCase.java @@ -20,45 +20,37 @@ import org.apache.flink.runtime.operators.testutils.ExpectedTestException; -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.assertThatThrownBy; /** Tests for {@link EmbeddedRocksDBStateBackend} on initialization. */ -public class RocksDBInitITCase { - - @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder(); +class RocksDBInitITCase { /** * This test checks that the RocksDB native code loader still responds to resetting the init * flag. */ @Test - public void testResetInitFlag() throws Exception { + void testResetInitFlag() throws Exception { EmbeddedRocksDBStateBackend.resetRocksDBLoadedFlag(); } @Test - public void testTempLibFolderDeletedOnFail() throws Exception { - File tempFolder = temporaryFolder.newFolder(); - try { - EmbeddedRocksDBStateBackend.ensureRocksDBIsLoaded( - tempFolder.getAbsolutePath(), - () -> { - throw new ExpectedTestException(); - }); - fail("Not throwing expected exception."); - } catch (IOException ignored) { - // ignored - } - File[] files = tempFolder.listFiles(); - Assert.assertNotNull(files); - Assert.assertEquals(0, files.length); + void testTempLibFolderDeletedOnFail(@TempDir File tempFolder) { + assertThatThrownBy( + () -> + EmbeddedRocksDBStateBackend.ensureRocksDBIsLoaded( + tempFolder.getAbsolutePath(), + () -> { + throw new ExpectedTestException(); + })) + .isInstanceOf(IOException.class); + assertThat(tempFolder).isEmptyDirectory(); } } diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBKeyedStateBackendTestFactory.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBKeyedStateBackendTestFactory.java index a069253935d793..81ce58cb3241e7 100644 --- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBKeyedStateBackendTestFactory.java +++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBKeyedStateBackendTestFactory.java @@ -27,26 +27,25 @@ import org.apache.flink.runtime.state.KeyGroupRange; import org.apache.flink.runtime.state.KeyedStateBackendParametersImpl; import org.apache.flink.runtime.state.ttl.TtlTimeProvider; +import org.apache.flink.testutils.junit.utils.TempDirUtils; import org.apache.flink.util.IOUtils; import org.apache.flink.util.TernaryBoolean; -import org.junit.rules.TemporaryFolder; - import java.io.IOException; +import java.nio.file.Path; import java.util.Collections; import static org.mockito.Mockito.mock; /** External resource for tests that require an instance of RocksDBKeyedStateBackend. */ -public class RocksDBKeyedStateBackendTestFactory implements AutoCloseable { +class RocksDBKeyedStateBackendTestFactory implements AutoCloseable { private MockEnvironment env; private RocksDBKeyedStateBackend keyedStateBackend; - public RocksDBKeyedStateBackend create( - TemporaryFolder tmp, TypeSerializer keySerializer, int maxKeyGroupNumber) - throws Exception { + RocksDBKeyedStateBackend create( + Path tmp, TypeSerializer keySerializer, int maxKeyGroupNumber) throws Exception { EmbeddedRocksDBStateBackend backend = getRocksDBStateBackend(tmp); env = MockEnvironment.builder().build(); JobID jobID = new JobID(); @@ -81,10 +80,8 @@ public void close() { IOUtils.closeQuietly(env); } - private EmbeddedRocksDBStateBackend getRocksDBStateBackend(TemporaryFolder tmp) - throws IOException { - String dbPath = tmp.newFolder().getAbsolutePath(); - String checkpointPath = tmp.newFolder().toURI().toString(); + private EmbeddedRocksDBStateBackend getRocksDBStateBackend(Path tmp) throws IOException { + String dbPath = TempDirUtils.newFolder(tmp).getAbsolutePath(); EmbeddedRocksDBStateBackend backend = new EmbeddedRocksDBStateBackend(TernaryBoolean.TRUE); backend.setDbStoragePath(dbPath); return backend; diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBMemoryControllerUtilsTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBMemoryControllerUtilsTest.java index d696eaccde6a18..c7b32ca44c5234 100644 --- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBMemoryControllerUtilsTest.java +++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBMemoryControllerUtilsTest.java @@ -18,34 +18,28 @@ package org.apache.flink.state.rocksdb; -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 org.rocksdb.Cache; import org.rocksdb.NativeLibraryLoader; import org.rocksdb.WriteBufferManager; 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 RocksDBMemoryControllerUtils}. */ -public class RocksDBMemoryControllerUtilsTest { +class RocksDBMemoryControllerUtilsTest { - @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,53 +54,57 @@ public void testCreateSharedResourcesWithExpectedCapacity() { RocksDBMemoryControllerUtils.calculateWriteBufferManagerCapacity( totalMemorySize, writeBufferRatio); - assertThat(factory.actualCacheCapacity, is(expectedCacheCapacity)); - assertThat(factory.actualWbmCapacity, is(expectedWbmCapacity)); - assertThat(rocksDBSharedResources.getWriteBufferManagerCapacity(), is(expectedWbmCapacity)); + assertThat(factory.actualCacheCapacity).isEqualTo(expectedCacheCapacity); + assertThat(factory.actualWbmCapacity).isEqualTo(expectedWbmCapacity); + assertThat(rocksDBSharedResources.getWriteBufferManagerCapacity()) + .isEqualTo(expectedWbmCapacity); } @Test - public void testCalculateRocksDBDefaultArenaBlockSize() { + void testCalculateRocksDBDefaultArenaBlockSize() { 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", - RocksDBMemoryControllerUtils.calculateRocksDBDefaultArenaBlockSize(writeBufferSize), - is(expectArenaBlockSize)); + RocksDBMemoryControllerUtils.calculateRocksDBDefaultArenaBlockSize( + writeBufferSize)) + .as("Arena block size calculation error for normal case") + .isEqualTo(expectArenaBlockSize); // Alignment tests assertThat( - "Arena block size calculation error for alignment case", - RocksDBMemoryControllerUtils.calculateRocksDBDefaultArenaBlockSize( - writeBufferSize - 1), - is(expectArenaBlockSize)); + RocksDBMemoryControllerUtils.calculateRocksDBDefaultArenaBlockSize( + writeBufferSize - 1)) + .as("Arena block size calculation error for alignment case") + .isEqualTo(expectArenaBlockSize); assertThat( - "Arena block size calculation error for alignment case2", - RocksDBMemoryControllerUtils.calculateRocksDBDefaultArenaBlockSize( - writeBufferSize + 8), - is(expectArenaBlockSize + align)); + RocksDBMemoryControllerUtils.calculateRocksDBDefaultArenaBlockSize( + writeBufferSize + 8)) + .as("Arena block size calculation error for alignment case2") + .isEqualTo(expectArenaBlockSize + align); } @Test - public void testCalculateRocksDBMutableLimit() { + void testCalculateRocksDBMutableLimit() { long bufferSize = 64 * 1024 * 1024; long limit = bufferSize * 7 / 8; - assertThat( - RocksDBMemoryControllerUtils.calculateRocksDBMutableLimit(bufferSize), is(limit)); + assertThat(RocksDBMemoryControllerUtils.calculateRocksDBMutableLimit(bufferSize)) + .isEqualTo(limit); } @Test - public void testValidateArenaBlockSize() { + void testValidateArenaBlockSize() { long arenaBlockSize = 8 * 1024 * 1024; - assertFalse( - RocksDBMemoryControllerUtils.validateArenaBlockSize( - arenaBlockSize, (long) (arenaBlockSize * 0.5))); - assertTrue( - RocksDBMemoryControllerUtils.validateArenaBlockSize( - arenaBlockSize, (long) (arenaBlockSize * 1.5))); + assertThat( + RocksDBMemoryControllerUtils.validateArenaBlockSize( + arenaBlockSize, (long) (arenaBlockSize * 0.5))) + .isFalse(); + assertThat( + RocksDBMemoryControllerUtils.validateArenaBlockSize( + arenaBlockSize, (long) (arenaBlockSize * 1.5))) + .isTrue(); } private static final class TestingRocksDBMemoryFactory diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBNativeMetricOptionsTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBNativeMetricOptionsTest.java index 771cf9e0b414e8..32c7f984f6c905 100644 --- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBNativeMetricOptionsTest.java +++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBNativeMetricOptionsTest.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 RocksDBNativeMetricOptionsTest { +class RocksDBNativeMetricOptionsTest { @Test - public void testNativeMetricsConfigurable() { + void testNativeMetricsConfigurable() { for (RocksDBProperty property : RocksDBProperty.values()) { Configuration config = new Configuration(); if (property.getConfigKey().contains("num-files-at-level")) { @@ -39,17 +39,12 @@ public void testNativeMetricsConfigurable() { RocksDBNativeMetricOptions options = RocksDBNativeMetricOptions.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-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBOperationsUtilsTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBOperationsUtilsTest.java index 9d2e4d19596ea7..2839768d943a2f 100644 --- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBOperationsUtilsTest.java +++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBOperationsUtilsTest.java @@ -20,10 +20,9 @@ import org.apache.flink.util.OperatingSystem; -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 org.rocksdb.ColumnFamilyOptions; import org.rocksdb.DBOptions; import org.rocksdb.NativeLibraryLoader; @@ -32,28 +31,24 @@ 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.hamcrest.Matchers.is; -import static org.junit.Assert.assertThat; -import static org.junit.Assume.assumeTrue; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assumptions.assumeThat; /** Tests for the {@link RocksDBOperationUtils}. */ -public class RocksDBOperationsUtilsTest { +class RocksDBOperationsUtilsTest { - @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 { - assumeTrue(OperatingSystem.isWindows()); + void testPathExceptionOnWindows(@TempDir File folder) throws Exception { + assumeThat(OperatingSystem.isWindows()).isTrue(); - final File folder = TMP_DIR.newFolder(); final File rocksDir = new File(folder, getLongString(247 - folder.getAbsolutePath().length())); @@ -74,43 +69,46 @@ 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"); } } @Test - public void testSanityCheckArenaBlockSize() { + void testSanityCheckArenaBlockSize() { long testWriteBufferSize = 56 * 1024 * 1024L; long testDefaultArenaSize = RocksDBMemoryControllerUtils.calculateRocksDBDefaultArenaBlockSize( testWriteBufferSize); long testWriteBufferCapacityBoundary = testDefaultArenaSize * 8 / 7; assertThat( - "The sanity check should pass with default arena block size", - RocksDBOperationUtils.sanityCheckArenaBlockSize( - testWriteBufferSize, 0, testWriteBufferCapacityBoundary), - is(true)); + RocksDBOperationUtils.sanityCheckArenaBlockSize( + testWriteBufferSize, 0, testWriteBufferCapacityBoundary)) + .as("The sanity check should pass with default arena block size") + .isTrue(); assertThat( - "The sanity check should pass with default arena block size given as argument", - RocksDBOperationUtils.sanityCheckArenaBlockSize( - testWriteBufferSize, testDefaultArenaSize, testWriteBufferCapacityBoundary), - is(true)); + RocksDBOperationUtils.sanityCheckArenaBlockSize( + testWriteBufferSize, + testDefaultArenaSize, + testWriteBufferCapacityBoundary)) + .as("The sanity check should pass with default arena block size given as argument") + .isTrue(); assertThat( - "The sanity check should pass when the configured arena block size is smaller than the boundary.", - RocksDBOperationUtils.sanityCheckArenaBlockSize( - testWriteBufferSize, - testDefaultArenaSize - 1, - testWriteBufferCapacityBoundary), - is(true)); + RocksDBOperationUtils.sanityCheckArenaBlockSize( + testWriteBufferSize, + testDefaultArenaSize - 1, + testWriteBufferCapacityBoundary)) + .as( + "The sanity check should pass when the configured arena block size is smaller than the boundary.") + .isTrue(); assertThat( - "The sanity check should fail when the configured arena block size is higher than the boundary.", - RocksDBOperationUtils.sanityCheckArenaBlockSize( - testWriteBufferSize, - testDefaultArenaSize + 1, - testWriteBufferCapacityBoundary), - is(false)); + RocksDBOperationUtils.sanityCheckArenaBlockSize( + testWriteBufferSize, + testDefaultArenaSize + 1, + testWriteBufferCapacityBoundary)) + .as( + "The sanity check should fail when the configured arena block size is higher than the boundary.") + .isFalse(); } private static String getLongString(int numChars) { diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBResourceContainerTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBResourceContainerTest.java index 1c13ce6d2d6102..c2efa42683ff82 100644 --- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBResourceContainerTest.java +++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBResourceContainerTest.java @@ -21,10 +21,9 @@ import org.apache.flink.runtime.memory.OpaqueMemoryResource; import org.apache.flink.util.function.ThrowingRunnable; -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 org.rocksdb.BlockBasedTableConfig; import org.rocksdb.BloomFilter; import org.rocksdb.Cache; @@ -40,38 +39,35 @@ import java.io.IOException; import java.lang.reflect.Field; +import java.nio.file.Path; 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.fail; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.fail; /** Tests to guard {@link RocksDBResourceContainer}. */ -public class RocksDBResourceContainerTest { +class RocksDBResourceContainerTest { - @ClassRule public static final TemporaryFolder TMP_FOLDER = new TemporaryFolder(); - - @BeforeClass - public static void ensureRocksDbNativeLibraryLoaded() throws IOException { - NativeLibraryLoader.getInstance().loadLibrary(TMP_FOLDER.newFolder().getAbsolutePath()); + @BeforeAll + static void ensureRocksDbNativeLibraryLoaded(@TempDir Path libDir) throws IOException { + NativeLibraryLoader.getInstance().loadLibrary(libDir.toFile().getAbsolutePath()); } // ------------------------------------------------------------------------ @Test - public void testFreeDBOptionsAfterClose() throws Exception { + void testFreeDBOptionsAfterClose() throws Exception { RocksDBResourceContainer container = new RocksDBResourceContainer(); 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 { RocksDBResourceContainer container = new RocksDBResourceContainer(); final int optionNumber = 20; ArrayList dbOptions = new ArrayList<>(optionNumber); @@ -80,7 +76,7 @@ public void testFreeMultipleDBOptionsAfterClose() throws Exception { } container.close(); for (DBOptions dbOption : dbOptions) { - assertThat(dbOption.isOwningHandle(), is(false)); + assertThat(dbOption.isOwningHandle()).isFalse(); } } @@ -92,14 +88,14 @@ public void testFreeMultipleDBOptionsAfterClose() throws Exception { * @throws Exception if unexpected error happened. */ @Test - public void testSharedResourcesAfterClose() throws Exception { + void testSharedResourcesAfterClose() throws Exception { OpaqueMemoryResource sharedResources = getSharedResources(); RocksDBResourceContainer container = new RocksDBResourceContainer(PredefinedOptions.DEFAULT, null, sharedResources); container.close(); RocksDBSharedResources rocksDBSharedResources = sharedResources.getResourceHandle(); - assertThat(rocksDBSharedResources.getCache().isOwningHandle(), is(false)); - assertThat(rocksDBSharedResources.getWriteBufferManager().isOwningHandle(), is(false)); + assertThat(rocksDBSharedResources.getCache().isOwningHandle()).isFalse(); + assertThat(rocksDBSharedResources.getWriteBufferManager().isOwningHandle()).isFalse(); } /** @@ -110,7 +106,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(); RocksDBResourceContainer container = @@ -121,10 +117,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(); } @@ -136,7 +131,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(); RocksDBResourceContainer container = @@ -147,8 +142,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(); } @@ -202,16 +198,16 @@ private WriteBufferManager getWriteBufferManager(DBOptions dbOptions) { } @Test - public void testFreeColumnOptionsAfterClose() throws Exception { + void testFreeColumnOptionsAfterClose() throws Exception { RocksDBResourceContainer container = new RocksDBResourceContainer(); 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 { RocksDBResourceContainer container = new RocksDBResourceContainer(); final int optionNumber = 20; ArrayList columnFamilyOptions = new ArrayList<>(optionNumber); @@ -220,12 +216,12 @@ public void testFreeMultipleColumnOptionsAfterClose() throws Exception { } container.close(); for (ColumnFamilyOptions columnFamilyOption : columnFamilyOptions) { - assertThat(columnFamilyOption.isOwningHandle(), is(false)); + assertThat(columnFamilyOption.isOwningHandle()).isFalse(); } } @Test - public void testFreeMultipleColumnOptionsWithPredefinedOptions() throws Exception { + void testFreeMultipleColumnOptionsWithPredefinedOptions() throws Exception { for (PredefinedOptions predefinedOptions : PredefinedOptions.values()) { RocksDBResourceContainer container = new RocksDBResourceContainer(predefinedOptions, null); @@ -236,13 +232,13 @@ public void testFreeMultipleColumnOptionsWithPredefinedOptions() throws Exceptio } 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); RocksDBSharedResources sharedResources = @@ -255,24 +251,24 @@ public void testFreeSharedResourcesAfterClose() throws Exception { new RocksDBResourceContainer(PredefinedOptions.DEFAULT, 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 { RocksDBResourceContainer container = new RocksDBResourceContainer(); 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); RocksDBSharedResources sharedResources = @@ -313,11 +309,13 @@ 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(); } } diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBRocksStateKeysAndNamespacesIteratorTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBRocksStateKeysAndNamespacesIteratorTest.java index e2aa5593e550f2..82752cffa9d235 100644 --- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBRocksStateKeysAndNamespacesIteratorTest.java +++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBRocksStateKeysAndNamespacesIteratorTest.java @@ -27,24 +27,26 @@ import org.apache.flink.runtime.state.CompositeKeySerializationUtils; import org.apache.flink.state.rocksdb.iterator.RocksStateKeysAndNamespaceIterator; -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 org.rocksdb.ColumnFamilyHandle; +import java.nio.file.Path; import java.util.ArrayList; -import java.util.Comparator; import java.util.List; import java.util.function.Function; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +import static org.assertj.core.api.Assertions.assertThat; /** Tests for the RocksDBRocksStateKeysAndNamespacesIterator. */ -public class RocksDBRocksStateKeysAndNamespacesIteratorTest { +class RocksDBRocksStateKeysAndNamespacesIteratorTest { - @Rule public final TemporaryFolder tmp = new TemporaryFolder(); + @TempDir private Path tmp; @Test - public void testIterator() throws Exception { + void testIterator() throws Exception { // test for keyGroupPrefixBytes == 1 && ambiguousKeyPossible == false testIteratorHelper(IntSerializer.INSTANCE, 128, i -> i); @@ -119,13 +121,11 @@ void testIteratorHelper( fetchedKeys.add((Tuple2) entry); } - fetchedKeys.sort(Comparator.comparingInt(a -> a.f0)); - Assert.assertEquals(1000, fetchedKeys.size()); - - for (int i = 0; i < 1000; ++i) { - Assert.assertEquals(i, fetchedKeys.get(i).f0.intValue()); - Assert.assertEquals(namespace, fetchedKeys.get(i).f1); - } + assertThat(fetchedKeys) + .containsExactlyInAnyOrderElementsOf( + IntStream.range(0, 1000) + .mapToObj(i -> Tuple2.of(i, namespace)) + .collect(Collectors.toList())); } } } diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBRocksStateKeysIteratorTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBRocksStateKeysIteratorTest.java index 4755a7a882cdb3..0eafe81c14ac67 100644 --- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBRocksStateKeysIteratorTest.java +++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBRocksStateKeysIteratorTest.java @@ -26,24 +26,26 @@ import org.apache.flink.runtime.state.CompositeKeySerializationUtils; import org.apache.flink.state.rocksdb.iterator.RocksStateKeysIterator; -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 org.rocksdb.ColumnFamilyHandle; +import java.nio.file.Path; import java.util.ArrayList; -import java.util.Comparator; import java.util.List; import java.util.function.Function; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +import static org.assertj.core.api.Assertions.assertThat; /** Tests for the RocksIteratorWrapper. */ -public class RocksDBRocksStateKeysIteratorTest { +class RocksDBRocksStateKeysIteratorTest { - @Rule public final TemporaryFolder tmp = new TemporaryFolder(); + @TempDir private Path tmp; @Test - public void testIterator() throws Exception { + void testIterator() throws Exception { // test for keyGroupPrefixBytes == 1 && ambiguousKeyPossible == false testIteratorHelper(IntSerializer.INSTANCE, 128, i -> i); @@ -115,12 +117,9 @@ void testIteratorHelper( fetchedKeys.add(Integer.parseInt(iteratorWrapper.next().toString())); } - fetchedKeys.sort(Comparator.comparingInt(a -> a)); - Assert.assertEquals(1000, fetchedKeys.size()); - - for (int i = 0; i < 1000; ++i) { - Assert.assertEquals(i, fetchedKeys.get(i).intValue()); - } + assertThat(fetchedKeys) + .containsExactlyInAnyOrderElementsOf( + IntStream.range(0, 1000).boxed().collect(Collectors.toList())); } } } diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBStateBackendConfigTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBStateBackendConfigTest.java index 1a0a45f804e611..d49e0682de888d 100644 --- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBStateBackendConfigTest.java +++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBStateBackendConfigTest.java @@ -44,16 +44,14 @@ import org.apache.flink.runtime.state.heap.HeapPriorityQueueSetFactory; 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.flink.util.IOUtils; import org.apache.commons.lang3.RandomUtils; -import org.junit.Assert; -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 org.rocksdb.BlockBasedTableConfig; import org.rocksdb.BloomFilter; import org.rocksdb.ColumnFamilyOptions; @@ -71,63 +69,56 @@ import java.util.Arrays; import java.util.Collection; import java.util.Collections; +import java.util.concurrent.TimeUnit; import static org.apache.flink.state.rocksdb.RocksDBTestUtils.createKeyedStateBackend; -import static org.hamcrest.CoreMatchers.anyOf; -import static org.hamcrest.CoreMatchers.instanceOf; -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.assertThatCode; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.assertj.core.api.Assumptions.assumeThat; /** Tests for configuring the RocksDB State Backend. */ @SuppressWarnings("serial") -public class RocksDBStateBackendConfigTest { +class RocksDBStateBackendConfigTest { - @Rule public final TemporaryFolder tempFolder = new TemporaryFolder(); + @TempDir private java.nio.file.Path tempFolder; // ------------------------------------------------------------------------ // default values // ------------------------------------------------------------------------ @Test - public void testDefaultsInSync() throws Exception { + void testDefaultsInSync() { final boolean defaultIncremental = CheckpointingOptions.INCREMENTAL_CHECKPOINTS.defaultValue(); EmbeddedRocksDBStateBackend backend = new EmbeddedRocksDBStateBackend(); - assertEquals(defaultIncremental, backend.isIncrementalCheckpointsEnabled()); + assertThat(backend.isIncrementalCheckpointsEnabled()).isEqualTo(defaultIncremental); } @Test - public void testDefaultDbLogDir() throws Exception { + void testDefaultDbLogDir() throws Exception { final EmbeddedRocksDBStateBackend backend = new EmbeddedRocksDBStateBackend(); 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 (RocksDBResourceContainer container = backend.createOptionsAndResourceContainer(null)) { - assertEquals( - RocksDBConfigurableOptions.LOG_LEVEL.defaultValue(), - container.getDbOptions().infoLogLevel()); - assertEquals(logFile.getParent(), container.getDbOptions().dbLogDir()); + assertThat(container.getDbOptions().infoLogLevel()) + .isEqualTo(RocksDBConfigurableOptions.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 (RocksDBResourceContainer container = backend.createOptionsAndResourceContainer( new File(longInstanceBasePath.toString()))) { - assertTrue(container.getDbOptions().dbLogDir().isEmpty()); + assertThat(container.getDbOptions().dbLogDir()).isEmpty(); } finally { logFile.delete(); } @@ -139,36 +130,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 EmbeddedRocksDBStateBackend rocksDbBackend = new EmbeddedRocksDBStateBackend(); - 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(rocksDbBackend.getDbStoragePaths()); + assertThat(rocksDbBackend.getDbStoragePaths()).isNull(); rocksDbBackend.setDbStoragePath(testDir1); - assertArrayEquals(new String[] {testDir1}, rocksDbBackend.getDbStoragePaths()); + assertThat(rocksDbBackend.getDbStoragePaths()).containsExactly(testDir1); rocksDbBackend.setDbStoragePath(null); - assertNull(rocksDbBackend.getDbStoragePaths()); + assertThat(rocksDbBackend.getDbStoragePaths()).isNull(); rocksDbBackend.setDbStoragePaths(testDir1, testDir2); - assertArrayEquals(new String[] {testDir1, testDir2}, rocksDbBackend.getDbStoragePaths()); + assertThat(rocksDbBackend.getDbStoragePaths()).containsExactly(testDir1, testDir2); - final MockEnvironment env = getMockEnvironment(tempFolder.newFolder()); + final MockEnvironment env = getMockEnvironment(TempDirUtils.newFolder(tempFolder)); final RocksDBKeyedStateBackend keyedBackend = createKeyedStateBackend(rocksDbBackend, env, IntSerializer.INSTANCE); try { File instanceBasePath = keyedBackend.getInstanceBasePath(); - assertThat( - instanceBasePath.getAbsolutePath(), - anyOf(startsWith(testDir1), startsWith(testDir2))); + assertThat(instanceBasePath.getAbsolutePath()) + .satisfiesAnyOf( + p -> assertThat(p).startsWith(testDir1), + p -> assertThat(p).startsWith(testDir2)); //noinspection NullArgumentToVariableArgMethod rocksDbBackend.setDbStoragePaths(null); - assertNull(rocksDbBackend.getDbStoragePaths()); + assertThat(rocksDbBackend.getDbStoragePaths()).isNull(); } finally { IOUtils.closeQuietly(keyedBackend); keyedBackend.dispose(); @@ -177,34 +169,30 @@ public void testSetDbPath() throws Exception { } @Test - public void testConfigureTimerService() throws Exception { + void testConfigureTimerService() throws Exception { - final MockEnvironment env = getMockEnvironment(tempFolder.newFolder()); + final MockEnvironment env = getMockEnvironment(TempDirUtils.newFolder(tempFolder)); // Fix the option key string - Assert.assertEquals( - "state.backend.rocksdb.timer-service.factory", - RocksDBOptions.TIMER_SERVICE_FACTORY.key()); + assertThat(RocksDBOptions.TIMER_SERVICE_FACTORY.key()) + .isEqualTo("state.backend.rocksdb.timer-service.factory"); // Fix the option value string and ensure all are covered - Assert.assertEquals(2, EmbeddedRocksDBStateBackend.PriorityQueueStateType.values().length); - Assert.assertEquals( - "ROCKSDB", EmbeddedRocksDBStateBackend.PriorityQueueStateType.ROCKSDB.toString()); - Assert.assertEquals( - "HEAP", EmbeddedRocksDBStateBackend.PriorityQueueStateType.HEAP.toString()); + assertThat(EmbeddedRocksDBStateBackend.PriorityQueueStateType.values()).hasSize(2); + assertThat(EmbeddedRocksDBStateBackend.PriorityQueueStateType.ROCKSDB) + .hasToString("ROCKSDB"); + assertThat(EmbeddedRocksDBStateBackend.PriorityQueueStateType.HEAP).hasToString("HEAP"); // Fix the default - Assert.assertEquals( - EmbeddedRocksDBStateBackend.PriorityQueueStateType.ROCKSDB, - RocksDBOptions.TIMER_SERVICE_FACTORY.defaultValue()); + assertThat(RocksDBOptions.TIMER_SERVICE_FACTORY.defaultValue()) + .isEqualTo(EmbeddedRocksDBStateBackend.PriorityQueueStateType.ROCKSDB); EmbeddedRocksDBStateBackend rocksDbBackend = new EmbeddedRocksDBStateBackend(); RocksDBKeyedStateBackend keyedBackend = createKeyedStateBackend(rocksDbBackend, env, IntSerializer.INSTANCE); - Assert.assertEquals( - RocksDBPriorityQueueSetFactory.class, - keyedBackend.getPriorityQueueFactory().getClass()); + assertThat(keyedBackend.getPriorityQueueFactory()) + .isExactlyInstanceOf(RocksDBPriorityQueueSetFactory.class); keyedBackend.dispose(); Configuration conf = new Configuration(); @@ -215,16 +203,15 @@ public void testConfigureTimerService() throws Exception { rocksDbBackend = rocksDbBackend.configure(conf, Thread.currentThread().getContextClassLoader()); keyedBackend = createKeyedStateBackend(rocksDbBackend, env, IntSerializer.INSTANCE); - Assert.assertEquals( - HeapPriorityQueueSetFactory.class, - keyedBackend.getPriorityQueueFactory().getClass()); + assertThat(keyedBackend.getPriorityQueueFactory()) + .isExactlyInstanceOf(HeapPriorityQueueSetFactory.class); keyedBackend.dispose(); env.close(); } @Test - public void testConfigureRocksDBPriorityQueueFactoryCacheSize() throws Exception { - final MockEnvironment env = getMockEnvironment(tempFolder.newFolder()); + void testConfigureRocksDBPriorityQueueFactoryCacheSize() throws Exception { + final MockEnvironment env = getMockEnvironment(TempDirUtils.newFolder(tempFolder)); EmbeddedRocksDBStateBackend rocksDbBackend = new EmbeddedRocksDBStateBackend(); int cacheSize = 512; Configuration conf = new Configuration(); @@ -239,20 +226,19 @@ public void testConfigureRocksDBPriorityQueueFactoryCacheSize() throws Exception RocksDBKeyedStateBackend keyedBackend = createKeyedStateBackend(rocksDbBackend, env, IntSerializer.INSTANCE); - Assert.assertEquals( - RocksDBPriorityQueueSetFactory.class, - keyedBackend.getPriorityQueueFactory().getClass()); - Assert.assertEquals( - cacheSize, - ((RocksDBPriorityQueueSetFactory) keyedBackend.getPriorityQueueFactory()) - .getCacheSize()); + assertThat(keyedBackend.getPriorityQueueFactory()) + .isExactlyInstanceOf(RocksDBPriorityQueueSetFactory.class); + assertThat( + ((RocksDBPriorityQueueSetFactory) keyedBackend.getPriorityQueueFactory()) + .getCacheSize()) + .isEqualTo(cacheSize); keyedBackend.dispose(); env.close(); } /** Validates that user custom configuration from code should override the config.yaml. */ @Test - public void testConfigureTimerServiceLoadingFromApplication() throws Exception { + void testConfigureTimerServiceLoadingFromApplication() throws Exception { final MockEnvironment env = new MockEnvironmentBuilder().build(); // priorityQueueStateType of the job backend @@ -273,9 +259,8 @@ public void testConfigureTimerServiceLoadingFromApplication() throws Exception { createKeyedStateBackend(configuredRocksDBStateBackend, env, IntSerializer.INSTANCE); // priorityQueueStateType of the job backend should be preserved - assertThat( - keyedBackend.getPriorityQueueFactory(), - instanceOf(HeapPriorityQueueSetFactory.class)); + assertThat(keyedBackend.getPriorityQueueFactory()) + .isInstanceOf(HeapPriorityQueueSetFactory.class); keyedBackend.close(); keyedBackend.dispose(); @@ -283,8 +268,8 @@ public void testConfigureTimerServiceLoadingFromApplication() throws Exception { } @Test - public void testConfigureRocksDBCompressionPerLevel() throws Exception { - final MockEnvironment env = getMockEnvironment(tempFolder.newFolder()); + void testConfigureRocksDBCompressionPerLevel() throws Exception { + final MockEnvironment env = getMockEnvironment(TempDirUtils.newFolder(tempFolder)); EmbeddedRocksDBStateBackend rocksDbBackend = new EmbeddedRocksDBStateBackend(); CompressionType[] compressionTypes = { CompressionType.NO_COMPRESSION, CompressionType.SNAPPY_COMPRESSION @@ -298,40 +283,40 @@ public void testConfigureRocksDBCompressionPerLevel() throws Exception { rocksDbBackend.configure(conf, Thread.currentThread().getContextClassLoader()); RocksDBResourceContainer resourceContainer = - rocksDbBackend.createOptionsAndResourceContainer(tempFolder.newFile()); + rocksDbBackend.createOptionsAndResourceContainer(TempDirUtils.newFile(tempFolder)); ColumnFamilyOptions columnFamilyOptions = resourceContainer.getColumnOptions(); - assertArrayEquals(compressionTypes, columnFamilyOptions.compressionPerLevel().toArray()); + assertThat(columnFamilyOptions.compressionPerLevel()).containsExactly(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, "s3://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 { @@ -350,18 +335,18 @@ private void testLocalDbPaths(String configuredPath, File expectedPath) throws E final EmbeddedRocksDBStateBackend rocksDbBackend = new EmbeddedRocksDBStateBackend(); rocksDbBackend.setDbStoragePath(configuredPath); - final MockEnvironment env = getMockEnvironment(tempFolder.newFolder()); + final MockEnvironment env = getMockEnvironment(TempDirUtils.newFolder(tempFolder)); RocksDBKeyedStateBackend keyedBackend = createKeyedStateBackend(rocksDbBackend, env, IntSerializer.INSTANCE); try { File instanceBasePath = keyedBackend.getInstanceBasePath(); - assertThat( - instanceBasePath.getAbsolutePath(), startsWith(expectedPath.getAbsolutePath())); + assertThat(instanceBasePath.getAbsolutePath()) + .startsWith(expectedPath.getAbsolutePath()); //noinspection NullArgumentToVariableArgMethod rocksDbBackend.setDbStoragePaths(null); - assertNull(rocksDbBackend.getDbStoragePaths()); + assertThat(rocksDbBackend.getDbStoragePaths()).isNull(); } finally { IOUtils.closeQuietly(keyedBackend); keyedBackend.dispose(); @@ -370,10 +355,10 @@ private void testLocalDbPaths(String configuredPath, File expectedPath) throws E } @Test - @Timeout(value = 60) - public void testCleanRelocatedDbLogs() throws Exception { - final File folder = tempFolder.newFolder(); - final File relocatedDBLogDir = tempFolder.newFolder("db_logs"); + @Timeout(value = 60, unit = TimeUnit.SECONDS) + 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()); @@ -387,7 +372,7 @@ public void testCleanRelocatedDbLogs() throws Exception { final String dbStoragePath = new Path(folder.toURI().toString()).toString(); rocksDbBackend.setDbStoragePath(dbStoragePath); - final MockEnvironment env = getMockEnvironment(tempFolder.newFolder()); + final MockEnvironment env = getMockEnvironment(TempDirUtils.newFolder(tempFolder)); RocksDBKeyedStateBackend keyedBackend = createKeyedStateBackend(rocksDbBackend, env, IntSerializer.INSTANCE); @@ -396,7 +381,8 @@ public void testCleanRelocatedDbLogs() throws Exception { RocksDBKeyedStateBackendBuilder.getInstanceRocksDBPath(instanceBasePath); // avoid tests without relocate. - Assume.assumeTrue(instanceRocksDBPath.getAbsolutePath().length() <= 255 - "_LOG".length()); + assumeThat(instanceRocksDBPath.getAbsolutePath().length()) + .isLessThanOrEqualTo(255 - "_LOG".length()); java.nio.file.Path[] relocatedDbLogs; try { @@ -416,8 +402,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"); } // ------------------------------------------------------------------------ @@ -429,12 +415,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 { EmbeddedRocksDBStateBackend rocksDbBackend = new EmbeddedRocksDBStateBackend(); - File dir1 = tempFolder.newFolder(); + File dir1 = TempDirUtils.newFolder(tempFolder); - assertNull(rocksDbBackend.getDbStoragePaths()); + assertThat(rocksDbBackend.getDbStoragePaths()).isNull(); final MockEnvironment env = getMockEnvironment(dir1); JobID jobID = env.getJobID(); @@ -459,7 +445,7 @@ public void testUseTempDirectories() throws Exception { try { File instanceBasePath = keyedBackend.getInstanceBasePath(); - assertThat(instanceBasePath.getAbsolutePath(), startsWith(dir1.getAbsolutePath())); + assertThat(instanceBasePath.getAbsolutePath()).startsWith(dir1.getAbsolutePath()); } finally { IOUtils.closeQuietly(keyedBackend); keyedBackend.dispose(); @@ -472,44 +458,40 @@ 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); + assumeThat(targetDir.setWritable(false, false)) + .as("Cannot mark directory non-writable") + .isTrue(); - String checkpointPath = tempFolder.newFolder().toURI().toString(); EmbeddedRocksDBStateBackend rocksDbBackend = new EmbeddedRocksDBStateBackend(); - try (MockEnvironment env = getMockEnvironment(tempFolder.newFolder())) { + try (MockEnvironment env = getMockEnvironment(TempDirUtils.newFolder(tempFolder))) { rocksDbBackend.setDbStoragePath(targetDir.getAbsolutePath()); - boolean hasFailure = false; - try { - JobID jobID = env.getJobID(); - KeyGroupRange keyGroupRange = new KeyGroupRange(0, 0); - TaskKvStateRegistry kvStateRegistry = - new KvStateRegistry().createTaskRegistry(env.getJobID(), new JobVertexID()); - CloseableRegistry cancelStreamRegistry = new CloseableRegistry(); - rocksDbBackend.createKeyedStateBackend( - new KeyedStateBackendParametersImpl<>( - (Environment) env, - jobID, - "foobar", - IntSerializer.INSTANCE, - 1, - keyGroupRange, - kvStateRegistry, - TtlTimeProvider.DEFAULT, - (MetricGroup) new UnregisteredMetricsGroup(), - Collections.emptyList(), - cancelStreamRegistry)); - } catch (Exception e) { - assertTrue(e.getMessage().contains("No local storage directories available")); - assertTrue(e.getMessage().contains(targetDir.getAbsolutePath())); - hasFailure = true; - } - assertTrue( - "We must see a failure because no storaged directory is feasible.", hasFailure); + JobID jobID = env.getJobID(); + KeyGroupRange keyGroupRange = new KeyGroupRange(0, 0); + TaskKvStateRegistry kvStateRegistry = + new KvStateRegistry().createTaskRegistry(env.getJobID(), new JobVertexID()); + CloseableRegistry cancelStreamRegistry = new CloseableRegistry(); + assertThatThrownBy( + () -> + rocksDbBackend.createKeyedStateBackend( + new KeyedStateBackendParametersImpl<>( + (Environment) env, + jobID, + "foobar", + IntSerializer.INSTANCE, + 1, + keyGroupRange, + kvStateRegistry, + TtlTimeProvider.DEFAULT, + (MetricGroup) new UnregisteredMetricsGroup(), + Collections.emptyList(), + cancelStreamRegistry)), + "We must see a failure because no storaged directory is feasible.") + .hasMessageContaining("No local storage directories available") + .hasMessageContaining(targetDir.getAbsolutePath()); } finally { //noinspection ResultOfMethodCallIgnored targetDir.setWritable(true, false); @@ -517,46 +499,49 @@ 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); + assumeThat(targetDir1.setWritable(false, false)) + .as("Cannot mark directory non-writable") + .isTrue(); - String checkpointPath = tempFolder.newFolder().toURI().toString(); EmbeddedRocksDBStateBackend rocksDbBackend = new EmbeddedRocksDBStateBackend(); - try (MockEnvironment env = getMockEnvironment(tempFolder.newFolder())) { + try (MockEnvironment env = getMockEnvironment(TempDirUtils.newFolder(tempFolder))) { rocksDbBackend.setDbStoragePaths( targetDir1.getAbsolutePath(), targetDir2.getAbsolutePath()); - try { - JobID jobID = env.getJobID(); - KeyGroupRange keyGroupRange = new KeyGroupRange(0, 0); - TaskKvStateRegistry kvStateRegistry = - new KvStateRegistry().createTaskRegistry(env.getJobID(), new JobVertexID()); - CloseableRegistry cancelStreamRegistry = new CloseableRegistry(); - CheckpointableKeyedStateBackend keyedStateBackend = - rocksDbBackend.createKeyedStateBackend( - new KeyedStateBackendParametersImpl<>( - (Environment) env, - jobID, - "foobar", - IntSerializer.INSTANCE, - 1, - keyGroupRange, - kvStateRegistry, - TtlTimeProvider.DEFAULT, - (MetricGroup) new UnregisteredMetricsGroup(), - Collections.emptyList(), - cancelStreamRegistry)); - - IOUtils.closeQuietly(keyedStateBackend); - keyedStateBackend.dispose(); - } catch (Exception e) { - e.printStackTrace(); - fail("Backend initialization failed even though some paths were available"); - } + assertThatCode( + () -> { + JobID jobID = env.getJobID(); + KeyGroupRange keyGroupRange = new KeyGroupRange(0, 0); + TaskKvStateRegistry kvStateRegistry = + new KvStateRegistry() + .createTaskRegistry( + env.getJobID(), new JobVertexID()); + CloseableRegistry cancelStreamRegistry = new CloseableRegistry(); + CheckpointableKeyedStateBackend keyedStateBackend = + rocksDbBackend.createKeyedStateBackend( + new KeyedStateBackendParametersImpl<>( + (Environment) env, + jobID, + "foobar", + IntSerializer.INSTANCE, + 1, + keyGroupRange, + kvStateRegistry, + TtlTimeProvider.DEFAULT, + (MetricGroup) + new UnregisteredMetricsGroup(), + Collections.emptyList(), + cancelStreamRegistry)); + + IOUtils.closeQuietly(keyedStateBackend); + keyedStateBackend.dispose(); + }) + .as("Backend initialization failed even though some paths were available") + .doesNotThrowAnyException(); } finally { //noinspection ResultOfMethodCallIgnored targetDir1.setWritable(true, false); @@ -568,12 +553,11 @@ public void testContinueOnSomeDbDirectoriesMissing() throws Exception { // ------------------------------------------------------------------------ @Test - public void testPredefinedOptions() throws Exception { - String checkpointPath = tempFolder.newFolder().toURI().toString(); + void testPredefinedOptions() throws Exception { EmbeddedRocksDBStateBackend rocksDbBackend = new EmbeddedRocksDBStateBackend(); // verify that we would use PredefinedOptions.DEFAULT by default. - assertEquals(PredefinedOptions.DEFAULT, rocksDbBackend.getPredefinedOptions()); + assertThat(rocksDbBackend.getPredefinedOptions()).isEqualTo(PredefinedOptions.DEFAULT); // verify that user could configure predefined options via config.yaml Configuration configuration = new Configuration(); @@ -581,17 +565,18 @@ public void testPredefinedOptions() throws Exception { RocksDBOptions.PREDEFINED_OPTIONS, PredefinedOptions.FLASH_SSD_OPTIMIZED.name()); rocksDbBackend = new EmbeddedRocksDBStateBackend(); rocksDbBackend = rocksDbBackend.configure(configuration, getClass().getClassLoader()); - assertEquals(PredefinedOptions.FLASH_SSD_OPTIMIZED, rocksDbBackend.getPredefinedOptions()); + assertThat(rocksDbBackend.getPredefinedOptions()) + .isEqualTo(PredefinedOptions.FLASH_SSD_OPTIMIZED); // verify that predefined options could be set programmatically and override pre-configured // one. rocksDbBackend.setPredefinedOptions(PredefinedOptions.SPINNING_DISK_OPTIMIZED); - assertEquals( - PredefinedOptions.SPINNING_DISK_OPTIMIZED, rocksDbBackend.getPredefinedOptions()); + assertThat(rocksDbBackend.getPredefinedOptions()) + .isEqualTo(PredefinedOptions.SPINNING_DISK_OPTIMIZED); } @Test - public void testConfigurableOptionsFromConfig() throws Exception { + void testConfigurableOptionsFromConfig() throws Exception { Configuration configuration = new Configuration(); // verify illegal configuration @@ -661,41 +646,39 @@ public void testConfigurableOptionsFromConfig() throws Exception { configuration, PredefinedOptions.DEFAULT, null, null, null, false)) { DBOptions dbOptions = optionsContainer.getDbOptions(); - assertEquals(-1, dbOptions.maxOpenFiles()); - assertEquals(InfoLogLevel.DEBUG_LEVEL, dbOptions.infoLogLevel()); - assertEquals("/tmp/rocksdb-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/rocksdb-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 { - String checkpointPath = tempFolder.newFolder().toURI().toString(); + void testOptionsFactory() throws Exception { EmbeddedRocksDBStateBackend rocksDbBackend = new EmbeddedRocksDBStateBackend(); // verify that user-defined options factory could be configured via config.yaml @@ -705,12 +688,12 @@ public void testOptionsFactory() throws Exception { rocksDbBackend = rocksDbBackend.configure(config, getClass().getClassLoader()); - assertTrue(rocksDbBackend.getRocksDBOptions() instanceof TestOptionsFactory); + assertThat(rocksDbBackend.getRocksDBOptions()).isInstanceOf(TestOptionsFactory.class); try (RocksDBResourceContainer optionsContainer = rocksDbBackend.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 @@ -734,12 +717,12 @@ public ColumnFamilyOptions createColumnOptions( try (RocksDBResourceContainer optionsContainer = rocksDbBackend.createOptionsAndResourceContainer(null)) { ColumnFamilyOptions colCreated = optionsContainer.getColumnOptions(); - assertEquals(CompactionStyle.FIFO, colCreated.compactionStyle()); + assertThat(colCreated.compactionStyle()).isEqualTo(CompactionStyle.FIFO); } } @Test - public void testPredefinedAndConfigurableOptions() throws Exception { + void testPredefinedAndConfigurableOptions() throws Exception { Configuration configuration = new Configuration(); configuration.set(RocksDBConfigurableOptions.COMPACTION_STYLE, CompactionStyle.UNIVERSAL); try (final RocksDBResourceContainer optionsContainer = @@ -752,8 +735,8 @@ public void testPredefinedAndConfigurableOptions() 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 RocksDBResourceContainer optionsContainer = @@ -766,13 +749,13 @@ public void testPredefinedAndConfigurableOptions() 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 RocksDBOptionsFactory optionsFactory = new RocksDBOptionsFactory() { @Override @@ -794,8 +777,8 @@ public ColumnFamilyOptions createColumnOptions( PredefinedOptions.SPINNING_DISK_OPTIMIZED, optionsFactory)) { final ColumnFamilyOptions columnFamilyOptions = optionsContainer.getColumnOptions(); - assertNotNull(columnFamilyOptions); - assertEquals(CompactionStyle.UNIVERSAL, columnFamilyOptions.compactionStyle()); + assertThat(columnFamilyOptions).isNotNull(); + assertThat(columnFamilyOptions.compactionStyle()).isEqualTo(CompactionStyle.UNIVERSAL); } } @@ -804,36 +787,28 @@ public ColumnFamilyOptions createColumnOptions( // ------------------------------------------------------------------------ @Test - public void testDefaultMemoryControlParameters() { + void testDefaultMemoryControlParameters() { RocksDBMemoryConfiguration memSettings = new RocksDBMemoryConfiguration(); - assertTrue(memSettings.isUsingManagedMemory()); - assertFalse(memSettings.isUsingFixedMemoryPerSlot()); - assertEquals( - RocksDBOptions.HIGH_PRIORITY_POOL_RATIO.defaultValue(), - memSettings.getHighPriorityPoolRatio(), - 0.0); - assertEquals( - RocksDBOptions.WRITE_BUFFER_RATIO.defaultValue(), - memSettings.getWriteBufferRatio(), - 0.0); + assertThat(memSettings.isUsingManagedMemory()).isTrue(); + assertThat(memSettings.isUsingFixedMemoryPerSlot()).isFalse(); + assertThat(memSettings.getHighPriorityPoolRatio()) + .isEqualTo(RocksDBOptions.HIGH_PRIORITY_POOL_RATIO.defaultValue()); + assertThat(memSettings.getWriteBufferRatio()) + .isEqualTo(RocksDBOptions.WRITE_BUFFER_RATIO.defaultValue()); RocksDBMemoryConfiguration configured = RocksDBMemoryConfiguration.fromOtherAndConfiguration( memSettings, new Configuration()); - assertTrue(configured.isUsingManagedMemory()); - assertFalse(configured.isUsingFixedMemoryPerSlot()); - assertEquals( - RocksDBOptions.HIGH_PRIORITY_POOL_RATIO.defaultValue(), - configured.getHighPriorityPoolRatio(), - 0.0); - assertEquals( - RocksDBOptions.WRITE_BUFFER_RATIO.defaultValue(), - configured.getWriteBufferRatio(), - 0.0); + assertThat(configured.isUsingManagedMemory()).isTrue(); + assertThat(configured.isUsingFixedMemoryPerSlot()).isFalse(); + assertThat(configured.getHighPriorityPoolRatio()) + .isEqualTo(RocksDBOptions.HIGH_PRIORITY_POOL_RATIO.defaultValue()); + assertThat(configured.getWriteBufferRatio()) + .isEqualTo(RocksDBOptions.WRITE_BUFFER_RATIO.defaultValue()); } @Test - public void testConfigureManagedMemory() { + void testConfigureManagedMemory() { final Configuration config = new Configuration(); config.set(RocksDBOptions.USE_MANAGED_MEMORY, true); @@ -841,11 +816,11 @@ public void testConfigureManagedMemory() { RocksDBMemoryConfiguration.fromOtherAndConfiguration( new RocksDBMemoryConfiguration(), config); - assertTrue(memSettings.isUsingManagedMemory()); + assertThat(memSettings.isUsingManagedMemory()).isTrue(); } @Test - public void testConfigureIllegalMemoryControlParameters() { + void testConfigureIllegalMemoryControlParameters() { RocksDBMemoryConfiguration memSettings = new RocksDBMemoryConfiguration(); verifySetParameter(() -> memSettings.setFixedMemoryPerSlot("-1B")); @@ -858,62 +833,57 @@ 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 testDefaultRestoreOverlapThreshold() { + void testDefaultRestoreOverlapThreshold() { EmbeddedRocksDBStateBackend rocksDBStateBackend = new EmbeddedRocksDBStateBackend(true); - assertTrue( - RocksDBConfigurableOptions.RESTORE_OVERLAP_FRACTION_THRESHOLD.defaultValue() - == rocksDBStateBackend.getOverlapFractionThreshold()); + assertThat(rocksDBStateBackend.getOverlapFractionThreshold()) + .isEqualTo( + RocksDBConfigurableOptions.RESTORE_OVERLAP_FRACTION_THRESHOLD + .defaultValue()); } @Test - public void testConfigureRestoreOverlapThreshold() { + void testConfigureRestoreOverlapThreshold() { EmbeddedRocksDBStateBackend rocksDBStateBackend = new EmbeddedRocksDBStateBackend(true); Configuration configuration = new Configuration(); configuration.set(RocksDBConfigurableOptions.RESTORE_OVERLAP_FRACTION_THRESHOLD, 0.3); rocksDBStateBackend = rocksDBStateBackend.configure(configuration, getClass().getClassLoader()); - assertTrue(0.3 == rocksDBStateBackend.getOverlapFractionThreshold()); + assertThat(rocksDBStateBackend.getOverlapFractionThreshold()).isEqualTo(0.3); } @Test - public void testDefaultUseIngestDB() { + void testDefaultUseIngestDB() { EmbeddedRocksDBStateBackend rocksDBStateBackend = new EmbeddedRocksDBStateBackend(true); - assertEquals( - RocksDBConfigurableOptions.USE_INGEST_DB_RESTORE_MODE.defaultValue(), - rocksDBStateBackend.getUseIngestDbRestoreMode()); + assertThat(rocksDBStateBackend.getUseIngestDbRestoreMode()) + .isEqualTo(RocksDBConfigurableOptions.USE_INGEST_DB_RESTORE_MODE.defaultValue()); } @Test - public void testConfigureUseIngestDB() { + void testConfigureUseIngestDB() { EmbeddedRocksDBStateBackend rocksDBStateBackend = new EmbeddedRocksDBStateBackend(true); Configuration configuration = new Configuration(); configuration.set(RocksDBConfigurableOptions.USE_INGEST_DB_RESTORE_MODE, true); rocksDBStateBackend = rocksDBStateBackend.configure(configuration, getClass().getClassLoader()); - assertTrue(rocksDBStateBackend.getUseIngestDbRestoreMode()); + assertThat(rocksDBStateBackend.getUseIngestDbRestoreMode()).isTrue(); } @Test - public void testDefaultUseDeleteFilesInRange() { + void testDefaultUseDeleteFilesInRange() { EmbeddedRocksDBStateBackend rocksDBStateBackend = new EmbeddedRocksDBStateBackend(true); - assertEquals( - RocksDBConfigurableOptions.USE_DELETE_FILES_IN_RANGE_DURING_RESCALING - .defaultValue(), - rocksDBStateBackend.isRescalingUseDeleteFilesInRange()); + assertThat(rocksDBStateBackend.isRescalingUseDeleteFilesInRange()) + .isEqualTo( + RocksDBConfigurableOptions.USE_DELETE_FILES_IN_RANGE_DURING_RESCALING + .defaultValue()); } @Test - public void testConfigureUseFilesInRange() { + void testConfigureUseFilesInRange() { EmbeddedRocksDBStateBackend rocksDBStateBackend = new EmbeddedRocksDBStateBackend(true); Configuration configuration = new Configuration(); configuration.set( @@ -922,23 +892,23 @@ public void testConfigureUseFilesInRange() { .defaultValue()); rocksDBStateBackend = rocksDBStateBackend.configure(configuration, getClass().getClassLoader()); - assertEquals( - !RocksDBConfigurableOptions.USE_DELETE_FILES_IN_RANGE_DURING_RESCALING - .defaultValue(), - rocksDBStateBackend.isRescalingUseDeleteFilesInRange()); + assertThat(rocksDBStateBackend.isRescalingUseDeleteFilesInRange()) + .isEqualTo( + !RocksDBConfigurableOptions.USE_DELETE_FILES_IN_RANGE_DURING_RESCALING + .defaultValue()); } @Test - public void testDefaultIncrementalRestoreInstanceBufferSize() { + void testDefaultIncrementalRestoreInstanceBufferSize() { EmbeddedRocksDBStateBackend rocksDBStateBackend = new EmbeddedRocksDBStateBackend(true); - assertEquals( - RocksDBConfigurableOptions.INCREMENTAL_RESTORE_ASYNC_COMPACT_AFTER_RESCALE - .defaultValue(), - rocksDBStateBackend.getIncrementalRestoreAsyncCompactAfterRescale()); + assertThat(rocksDBStateBackend.getIncrementalRestoreAsyncCompactAfterRescale()) + .isEqualTo( + RocksDBConfigurableOptions.INCREMENTAL_RESTORE_ASYNC_COMPACT_AFTER_RESCALE + .defaultValue()); } @Test - public void testConfigureIncrementalRestoreInstanceBufferSize() { + void testConfigureIncrementalRestoreInstanceBufferSize() { EmbeddedRocksDBStateBackend rocksDBStateBackend = new EmbeddedRocksDBStateBackend(true); Configuration configuration = new Configuration(); boolean notDefault = @@ -949,12 +919,12 @@ public void testConfigureIncrementalRestoreInstanceBufferSize() { notDefault); rocksDBStateBackend = rocksDBStateBackend.configure(configuration, getClass().getClassLoader()); - assertEquals( - notDefault, rocksDBStateBackend.getIncrementalRestoreAsyncCompactAfterRescale()); + assertThat(rocksDBStateBackend.getIncrementalRestoreAsyncCompactAfterRescale()) + .isEqualTo(notDefault); } @Test - public void testConfigurePeriodicCompactionTime() throws Exception { + void testConfigurePeriodicCompactionTime() throws Exception { EmbeddedRocksDBStateBackend rocksDBStateBackend = new EmbeddedRocksDBStateBackend(true); Configuration configuration = new Configuration(); configuration.setString( @@ -963,12 +933,12 @@ public void testConfigurePeriodicCompactionTime() throws Exception { rocksDBStateBackend.configure(configuration, getClass().getClassLoader()); try (RocksDBResourceContainer resourceContainer = rocksDBStateBackend.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 { EmbeddedRocksDBStateBackend rocksDBStateBackend = new EmbeddedRocksDBStateBackend(true); Configuration configuration = new Configuration(); configuration.setString( @@ -978,17 +948,12 @@ public void testConfigureQueryTimeAfterNumEntries() throws Exception { rocksDBStateBackend.configure(configuration, getClass().getClassLoader()); try (RocksDBResourceContainer resourceContainer = rocksDBStateBackend.createOptionsAndResourceContainer(null)) { - assertEquals(100L, resourceContainer.getQueryTimeAfterNumEntries().longValue()); + assertThat(resourceContainer.getQueryTimeAfterNumEntries()).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); } // ------------------------------------------------------------------------ @@ -1008,12 +973,8 @@ private void verifyIllegalArgument(ConfigOption configOption, String configVa configuration.setString(configOption.key(), configValue); EmbeddedRocksDBStateBackend stateBackend = new EmbeddedRocksDBStateBackend(); - 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-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBStateDownloaderTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBStateDownloaderTest.java index 3f895d4370f871..4f566e70fc4b37 100644 --- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBStateDownloaderTest.java +++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBStateDownloaderTest.java @@ -27,20 +27,16 @@ import org.apache.flink.runtime.state.StreamStateHandle; import org.apache.flink.runtime.state.TestStreamStateHandle; import org.apache.flink.runtime.state.memory.ByteStreamStateHandle; +import org.apache.flink.testutils.junit.utils.TempDirUtils; import org.apache.flink.util.ExceptionUtils; -import org.apache.flink.util.TestLogger; -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 javax.annotation.Nonnull; import java.io.IOException; -import java.nio.file.Files; import java.nio.file.Path; -import java.nio.file.Paths; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; @@ -57,20 +53,20 @@ import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.function.Predicate; import java.util.stream.Collectors; import static java.util.Collections.emptyList; import static java.util.Collections.singletonList; -import static org.junit.Assert.assertArrayEquals; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.fail; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Test class for {@link RocksDBStateDownloader}. */ -public class RocksDBStateDownloaderTest extends TestLogger { - @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder(); +class RocksDBStateDownloaderTest { + @TempDir private Path temporaryFolder; @Test - public void testWaitForDownloadIfInterrupted() + void testWaitForDownloadIfInterrupted() throws IOException, InterruptedException, ExecutionException { StateHandleDownloadSpec spec = new StateHandleDownloadSpec( @@ -86,7 +82,7 @@ public void testWaitForDownloadIfInterrupted() emptyList(), new ByteStreamStateHandle("meta", new byte[] {1, 2, 3, 4}), 5L), - temporaryFolder.newFolder().toPath().resolve("dst")); + TempDirUtils.newFolder(temporaryFolder).toPath().resolve("dst")); CompletableFuture downloaderFuture = new CompletableFuture<>(); BlockingExecutorService executorService = new BlockingExecutorService(); Thread downloader = createDownloader(executorService, spec, downloaderFuture); @@ -94,9 +90,9 @@ public void testWaitForDownloadIfInterrupted() for (int attempt = 0; attempt < 5; attempt++) { downloader.interrupt(); Thread.sleep(50); - Assert.assertTrue( - "downloader should ignore interrupts while download is in progress", - downloader.isAlive()); + assertThat(downloader.isAlive()) + .as("downloader should ignore interrupts while download is in progress") + .isTrue(); } executorService.unblock(); downloaderFuture.get(); @@ -132,7 +128,7 @@ private static Thread createDownloader( /** Test that the exception arose in the thread pool will rethrow to the main thread. */ @Test - public void testMultiThreadRestoreThreadPoolExceptionRethrow() { + void testMultiThreadRestoreThreadPoolExceptionRethrow() throws Exception { SpecifiedException expectedCause = new SpecifiedException("throw exception while multi thread restore."); StreamStateHandle stateHandle = new ThrowingStateHandle(expectedCause); @@ -150,21 +146,23 @@ public void testMultiThreadRestoreThreadPoolExceptionRethrow() { stateHandle); try (RocksDBStateDownloader rocksDBStateDownloader = new RocksDBStateDownloader(5)) { - rocksDBStateDownloader.transferAllStateDataToDirectory( - Collections.singletonList( - new StateHandleDownloadSpec( - incrementalKeyedStateHandle, - temporaryFolder.newFolder().toPath())), - new CloseableRegistry()); - fail(); - } catch (Exception e) { - assertEquals(expectedCause, e.getCause()); + assertThatThrownBy( + () -> + rocksDBStateDownloader.transferAllStateDataToDirectory( + Collections.singletonList( + new StateHandleDownloadSpec( + incrementalKeyedStateHandle, + TempDirUtils.newFolder(temporaryFolder) + .toPath())), + new CloseableRegistry())) + .cause() + .isSameAs(expectedCause); } } /** Tests that download files with multi-thread correctly. */ @Test - public void testMultiThreadRestoreCorrectly() throws Exception { + void testMultiThreadRestoreCorrectly() throws Exception { int numRemoteHandles = 3; int numSubHandles = 6; byte[][][] contents = createContents(numRemoteHandles, numSubHandles); @@ -172,7 +170,7 @@ public void testMultiThreadRestoreCorrectly() throws Exception { for (int i = 0; i < numRemoteHandles; ++i) { downloadRequests.add( createDownloadRequestForContent( - temporaryFolder.newFolder().toPath(), contents[i], i)); + TempDirUtils.newFolder(temporaryFolder).toPath(), contents[i], i)); } try (RocksDBStateDownloader rocksDBStateDownloader = new RocksDBStateDownloader(4)) { @@ -183,7 +181,7 @@ public void testMultiThreadRestoreCorrectly() throws Exception { for (int i = 0; i < numRemoteHandles; ++i) { StateHandleDownloadSpec downloadRequest = downloadRequests.get(i); Path dstPath = downloadRequest.getDownloadDestination(); - Assert.assertTrue(dstPath.toFile().exists()); + assertThat(dstPath).exists(); for (int j = 0; j < numSubHandles; ++j) { assertStateContentEqual( contents[i][j], dstPath.resolve(String.format("sharedState-%d-%d", i, j))); @@ -193,7 +191,7 @@ public void testMultiThreadRestoreCorrectly() throws Exception { /** Tests cleanup on download failures. */ @Test - public void testMultiThreadCleanupOnFailure() throws Exception { + void testMultiThreadCleanupOnFailure() throws Exception { int numRemoteHandles = 3; int numSubHandles = 6; byte[][][] contents = createContents(numRemoteHandles, numSubHandles); @@ -201,7 +199,7 @@ public void testMultiThreadCleanupOnFailure() throws Exception { for (int i = 0; i < numRemoteHandles; ++i) { downloadRequests.add( createDownloadRequestForContent( - temporaryFolder.newFolder().toPath(), contents[i], i)); + TempDirUtils.newFolder(temporaryFolder).toPath(), contents[i], i)); } IncrementalRemoteKeyedStateHandle stateHandle = @@ -217,18 +215,19 @@ public void testMultiThreadCleanupOnFailure() throws Exception { CloseableRegistry closeableRegistry = new CloseableRegistry(); try (RocksDBStateDownloader rocksDBStateDownloader = new RocksDBStateDownloader(5)) { - rocksDBStateDownloader.transferAllStateDataToDirectory( - downloadRequests, closeableRegistry); - fail("Exception is expected"); - } catch (IOException ignore) { + assertThatThrownBy( + () -> + rocksDBStateDownloader.transferAllStateDataToDirectory( + downloadRequests, closeableRegistry)) + .isInstanceOf(IOException.class); } // Check that all download directories have been deleted for (StateHandleDownloadSpec downloadRequest : downloadRequests) { - Assert.assertFalse(downloadRequest.getDownloadDestination().toFile().exists()); + assertThat(downloadRequest.getDownloadDestination()).doesNotExist(); } // The passed in closable registry should not be closed by us on failure. - Assert.assertFalse(closeableRegistry.isClosed()); + assertThat(closeableRegistry.isClosed()).isFalse(); } /** @@ -236,7 +235,7 @@ public void testMultiThreadCleanupOnFailure() throws Exception { * without being wrapped in a merged "N downloads failed" message. */ @Test - public void testSingleDownloadFailureSurfacedDirectly() throws Exception { + void testSingleDownloadFailureSurfacedDirectly() throws Exception { IOException rootCause = new IOException("file not found on remote storage"); StreamStateHandle failingHandle = new ThrowingStateHandle(rootCause); @@ -250,17 +249,19 @@ public void testSingleDownloadFailureSurfacedDirectly() throws Exception { failingHandle); try (RocksDBStateDownloader downloader = new RocksDBStateDownloader(1)) { - downloader.transferAllStateDataToDirectory( - singletonList( - new StateHandleDownloadSpec( - stateHandle, temporaryFolder.newFolder().toPath())), - new CloseableRegistry()); - fail("Expected IOException"); - } catch (IOException e) { - assertEquals(rootCause, e.getCause()); - Assert.assertFalse( - "Single failure should not produce a merged message, got: " + e.getMessage(), - e.getMessage() != null && e.getMessage().contains("downloads failed")); + assertThatThrownBy( + () -> + downloader.transferAllStateDataToDirectory( + singletonList( + new StateHandleDownloadSpec( + stateHandle, + TempDirUtils.newFolder(temporaryFolder) + .toPath())), + new CloseableRegistry())) + .isInstanceOf(IOException.class) + .satisfies(e -> assertThat(e.getCause()).isSameAs(rootCause)) + .as("Single failure should not produce a merged message") + .hasMessageNotContaining("downloads failed"); } } @@ -272,7 +273,7 @@ public void testSingleDownloadFailureSurfacedDirectly() throws Exception { * the real cause (e.g. FileNotFoundException for a missing state file) was lost. */ @Test - public void testRootCauseVisibleAmongCascadeFailures() throws Exception { + void testRootCauseVisibleAmongCascadeFailures() throws Exception { int numRemoteHandles = 3; int numSubHandles = 6; byte[][][] contents = createContents(numRemoteHandles, numSubHandles); @@ -280,7 +281,7 @@ public void testRootCauseVisibleAmongCascadeFailures() throws Exception { for (int i = 0; i < numRemoteHandles; ++i) { downloadRequests.add( createDownloadRequestForContent( - temporaryFolder.newFolder().toPath(), contents[i], i)); + TempDirUtils.newFolder(temporaryFolder).toPath(), contents[i], i)); } IOException rootCause = new IOException("state file missing from remote storage"); @@ -290,24 +291,21 @@ public void testRootCauseVisibleAmongCascadeFailures() throws Exception { .getSharedState() .add(HandleAndLocalPath.of(new ThrowingStateHandle(rootCause), "error-handle")); + Predicate hasRootCauseMessage = + t -> rootCause.getMessage().equals(t.getMessage()); + try (RocksDBStateDownloader downloader = new RocksDBStateDownloader(5)) { - downloader.transferAllStateDataToDirectory(downloadRequests, new CloseableRegistry()); - fail("Expected IOException"); - } catch (IOException e) { - boolean rootCauseVisible = - (e.getCause() != null - && rootCause.getMessage().equals(e.getCause().getMessage())) - || (e.getMessage() != null - && e.getMessage().contains(rootCause.getMessage())) - || ExceptionUtils.findThrowable( - e, t -> rootCause.getMessage().equals(t.getMessage())) - .isPresent(); - Assert.assertTrue( - "Root cause '" - + rootCause.getMessage() - + "' should be visible in exception, got: " - + e, - rootCauseVisible); + assertThatThrownBy( + () -> + downloader.transferAllStateDataToDirectory( + downloadRequests, new CloseableRegistry())) + .isInstanceOf(IOException.class) + .as("Root cause '%s' should be visible in exception", rootCause.getMessage()) + .satisfiesAnyOf( + e -> assertThat(e).hasMessageContaining(rootCause.getMessage()), + e -> + assertThat(ExceptionUtils.findThrowable(e, hasRootCauseMessage)) + .isPresent()); } } @@ -317,7 +315,7 @@ public void testRootCauseVisibleAmongCascadeFailures() throws Exception { * their failure point before any registry closure, so each failure is captured independently. */ @Test - public void testMultipleDistinctFailuresMergedInMessage() throws Exception { + void testMultipleDistinctFailuresMergedInMessage() throws Exception { int n = 3; CyclicBarrier barrier = new CyclicBarrier(n); IOException causeA = new IOException("error-A: bucket not accessible"); @@ -339,29 +337,29 @@ public void testMultipleDistinctFailuresMergedInMessage() throws Exception { handles.get(0).getHandle()); try (RocksDBStateDownloader downloader = new RocksDBStateDownloader(n)) { - downloader.transferAllStateDataToDirectory( - singletonList( - new StateHandleDownloadSpec( - stateHandle, temporaryFolder.newFolder().toPath())), - new CloseableRegistry()); - fail("Expected IOException"); - } catch (IOException e) { - Assert.assertTrue( - "Expected merged error message, got: " + e.getMessage(), - e.getMessage() != null - && e.getMessage().contains("downloads failed with distinct errors")); - Assert.assertTrue( - "Expected causeA in message", e.getMessage().contains(causeA.getMessage())); - Assert.assertTrue( - "Expected causeB in message", e.getMessage().contains(causeB.getMessage())); - Assert.assertTrue( - "Expected causeC in message", e.getMessage().contains(causeC.getMessage())); + assertThatThrownBy( + () -> + downloader.transferAllStateDataToDirectory( + singletonList( + new StateHandleDownloadSpec( + stateHandle, + TempDirUtils.newFolder(temporaryFolder) + .toPath())), + new CloseableRegistry())) + .isInstanceOf(IOException.class) + .as("Expected merged error message") + .hasMessageContaining("downloads failed with distinct errors") + .as("Expected causeA in message") + .hasMessageContaining(causeA.getMessage()) + .as("Expected causeB in message") + .hasMessageContaining(causeB.getMessage()) + .as("Expected causeC in message") + .hasMessageContaining(causeC.getMessage()); } } - private void assertStateContentEqual(byte[] expected, Path path) throws IOException { - byte[] actual = Files.readAllBytes(Paths.get(path.toUri())); - assertArrayEquals(expected, actual); + private void assertStateContentEqual(byte[] expected, Path path) { + assertThat(path).hasBinaryContent(expected); } /** diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBStateOptionTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBStateOptionTest.java index beb01156670a6f..0b38741872f9d2 100644 --- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBStateOptionTest.java +++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBStateOptionTest.java @@ -30,9 +30,7 @@ import org.apache.flink.streaming.api.operators.TimerHeapInternalTimer; import org.apache.flink.streaming.api.operators.TimerSerializer; -import org.junit.Rule; -import org.junit.Test; -import org.junit.rules.TemporaryFolder; +import org.junit.jupiter.api.Test; import org.rocksdb.ColumnFamilyOptions; import org.rocksdb.DBOptions; @@ -49,8 +47,7 @@ import java.util.stream.IntStream; import static org.apache.flink.state.rocksdb.RocksDBTestUtils.createKeyedStateBackend; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertTrue; +import static org.assertj.core.api.Assertions.assertThat; /** * Tests to cover cases that if user choose options previously prone to misuse, embedded RocksDB @@ -61,16 +58,14 @@ * RocksDBOptionsFactory}, and might lead some operations could not get expected result, e.g. * FLINK-17800 */ -public class RocksDBStateOptionTest { - - @Rule public final TemporaryFolder tempFolder = new TemporaryFolder(); +class RocksDBStateOptionTest { /** * Tests to cover case when user choose optimizeForPointLookup with iterator interfaces on map * state. */ @Test - public void testUseOptimizePointLookupWithMapState() throws Exception { + void testUseOptimizePointLookupWithMapState() throws Exception { EmbeddedRocksDBStateBackend rocksDBStateBackend = createStateBackendWithOptimizePointLookup(); RocksDBKeyedStateBackend keyedStateBackend = @@ -99,11 +94,11 @@ public void testUseOptimizePointLookupWithMapState() throws Exception { Iterator> iterator = mapState.entries().iterator(); while (iterator.hasNext()) { Map.Entry entry = iterator.next(); - assertEquals(entry.getValue(), expectedResult.remove(entry.getKey())); + assertThat(entry.getValue()).isEqualTo(expectedResult.remove(entry.getKey())); iterator.remove(); } - assertTrue(expectedResult.isEmpty()); - assertTrue(mapState.isEmpty()); + assertThat(expectedResult).isEmpty(); + assertThat(mapState.isEmpty()).isTrue(); } finally { keyedStateBackend.dispose(); } @@ -114,7 +109,7 @@ public void testUseOptimizePointLookupWithMapState() throws Exception { * queue. */ @Test - public void testUseOptimizePointLookupWithPriorityQueue() throws IOException { + void testUseOptimizePointLookupWithPriorityQueue() throws IOException { EmbeddedRocksDBStateBackend rocksDBStateBackend = createStateBackendWithOptimizePointLookup(); RocksDBKeyedStateBackend keyedStateBackend = @@ -134,9 +129,8 @@ public void testUseOptimizePointLookupWithPriorityQueue() throws IOException { PriorityQueue> expectedPriorityQueue = new PriorityQueue<>((o1, o2) -> (int) (o1.getTimestamp() - o2.getTimestamp())); // ensure we insert timers more than cache capacity. - assertTrue( - keyedStateBackend.getPriorityQueueFactory() - instanceof RocksDBPriorityQueueSetFactory); + assertThat(keyedStateBackend.getPriorityQueueFactory()) + .isInstanceOf(RocksDBPriorityQueueSetFactory.class); int queueSize = ((RocksDBPriorityQueueSetFactory) keyedStateBackend.getPriorityQueueFactory()) .getCacheSize() @@ -150,13 +144,13 @@ public void testUseOptimizePointLookupWithPriorityQueue() throws IOException { priorityQueue.add(timer); expectedPriorityQueue.add(timer); } - assertEquals(queueSize, priorityQueue.size()); + assertThat(priorityQueue.size()).isEqualTo(queueSize); TimerHeapInternalTimer timer; while ((timer = priorityQueue.poll()) != null) { - assertEquals(expectedPriorityQueue.poll(), timer); + assertThat(timer).isEqualTo(expectedPriorityQueue.poll()); } - assertTrue(expectedPriorityQueue.isEmpty()); - assertTrue(priorityQueue.isEmpty()); + assertThat(expectedPriorityQueue).isEmpty(); + assertThat(priorityQueue.isEmpty()).isTrue(); } finally { keyedStateBackend.dispose(); } diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBStateUploaderTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBStateUploaderTest.java index 6b63c6011d82ca..5b12cf22789bda 100644 --- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBStateUploaderTest.java +++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBStateUploaderTest.java @@ -30,7 +30,6 @@ import org.apache.flink.runtime.state.memory.ByteStreamStateHandle; import org.apache.flink.testutils.junit.utils.TempDirUtils; import org.apache.flink.util.IOUtils; -import org.apache.flink.util.TestLogger; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; @@ -53,7 +52,7 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Test class for {@link RocksDBStateUploader}. */ -public class RocksDBStateUploaderTest extends TestLogger { +class RocksDBStateUploaderTest { @TempDir private Path temporaryFolder; diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBWriteBatchWrapperTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBWriteBatchWrapperTest.java index 52143c5cacaf30..95792779ff3fcc 100644 --- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBWriteBatchWrapperTest.java +++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBWriteBatchWrapperTest.java @@ -22,15 +22,14 @@ import org.apache.flink.core.fs.CloseableRegistry; import org.apache.flink.runtime.execution.CancelTaskException; -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 org.rocksdb.ColumnFamilyDescriptor; import org.rocksdb.ColumnFamilyHandle; import org.rocksdb.RocksDB; import org.rocksdb.WriteOptions; +import java.io.File; import java.io.IOException; import java.util.ArrayList; import java.util.List; @@ -39,17 +38,13 @@ import static org.apache.flink.state.rocksdb.RocksDBConfigurableOptions.WRITE_BATCH_SIZE; import static org.assertj.core.api.Assertions.assertThat; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertTrue; +import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Tests to guard {@link RocksDBWriteBatchWrapper}. */ -public class RocksDBWriteBatchWrapperTest { +class RocksDBWriteBatchWrapperTest { - @Rule public TemporaryFolder folder = new TemporaryFolder(); - - @Test(expected = CancelTaskException.class) - public void testAsyncCancellation() throws Exception { + @Test + void testAsyncCancellation(@TempDir File folder) { final CompletableFuture writeStartedFuture = new CompletableFuture<>(); final CompletableFuture cancellationRequestedFuture = new CompletableFuture<>(); final CloseableRegistry registry = new CloseableRegistry(); @@ -69,56 +64,68 @@ public void testAsyncCancellation() throws Exception { final int cancellationCheckInterval = 1; long batchSizeBytes = WRITE_BATCH_SIZE.defaultValue().getBytes(); - try (RocksDB db = RocksDB.open(folder.newFolder().getAbsolutePath()); - WriteOptions options = new WriteOptions().setDisableWAL(true); - ColumnFamilyHandle handle = - db.createColumnFamily(new ColumnFamilyDescriptor("test".getBytes())); - RocksDBWriteBatchWrapper writeBatchWrapper = - new RocksDBWriteBatchWrapper( - db, - options, - capacity, - batchSizeBytes, - cancellationCheckInterval, - batchSizeBytes)) { - registry.registerCloseable(writeBatchWrapper.getCancelCloseable()); - // After the `writeStartedFuture` completes, the registry will start to close. - writeStartedFuture.complete(null); - - // In the infinite loop, we want to verify that the `put` method will check cancellation - // state on every `batch.count() % cancellationCheckInterval == 0`. We set - // cancellationCheckInterval to 1, So, we expect it will throw CancelTaskException - // no later than batch count becoming 2 in this test case. - //noinspection InfiniteLoopStatement - for (int i = 0; ; i++) { - try { - writeBatchWrapper.put( - handle, ("key:" + i).getBytes(), ("value:" + i).getBytes()); - } catch (Exception e) { - cancellationRequestedFuture.join(); // shouldn't have any errors - throw e; - } - // make sure that cancellation is triggered earlier than periodic flush - // but allow some delay of cancellation propagation - assertThat(i).isLessThan(cancellationCheckInterval * 2); - if (i == 0) { - // make sure the registry is closed at least after the first run, so that we - // can verify the cancellation check is validating correctly. - cancellationRequestedFuture.join(); - } - } - } + assertThatThrownBy( + () -> { + try (RocksDB db = RocksDB.open(folder.getAbsolutePath()); + WriteOptions options = new WriteOptions().setDisableWAL(true); + ColumnFamilyHandle handle = + db.createColumnFamily( + new ColumnFamilyDescriptor("test".getBytes())); + RocksDBWriteBatchWrapper writeBatchWrapper = + new RocksDBWriteBatchWrapper( + db, + options, + capacity, + batchSizeBytes, + cancellationCheckInterval, + batchSizeBytes)) { + registry.registerCloseable(writeBatchWrapper.getCancelCloseable()); + // After the `writeStartedFuture` completes, the registry will start + // to close. + writeStartedFuture.complete(null); + + // In the infinite loop, we want to verify that the `put` method + // will check cancellation state on every `batch.count() % + // cancellationCheckInterval == 0`. We set cancellationCheckInterval + // to 1, So, we expect it will throw CancelTaskException no later + // than batch count becoming 2 in this test case. + //noinspection InfiniteLoopStatement + for (int i = 0; ; i++) { + try { + writeBatchWrapper.put( + handle, + ("key:" + i).getBytes(), + ("value:" + i).getBytes()); + } catch (Exception e) { + // shouldn't have any errors + cancellationRequestedFuture.join(); + throw e; + } + // make sure that cancellation is triggered earlier than + // periodic flush but allow some delay of cancellation + // propagation + assertThat(i).isLessThan(cancellationCheckInterval * 2); + if (i == 0) { + // make sure the registry is closed at least after the first + // run, so that we can verify the cancellation check is + // validating correctly. + cancellationRequestedFuture.join(); + } + } + } + }) + .isInstanceOf(CancelTaskException.class); } @Test - public void basicTest() throws Exception { + void basicTest(@TempDir File folder) throws Exception { List> data = new ArrayList<>(10000); for (int i = 0; i < 10000; ++i) { data.add(new Tuple2<>(("key:" + i).getBytes(), ("value:" + i).getBytes())); } - try (RocksDB db = RocksDB.open(folder.newFolder().getAbsolutePath()); + try (RocksDB db = RocksDB.open(folder.getAbsolutePath()); WriteOptions options = new WriteOptions().setDisableWAL(true); ColumnFamilyHandle handle = db.createColumnFamily(new ColumnFamilyDescriptor("test".getBytes())); @@ -134,7 +141,7 @@ public void basicTest() throws Exception { // valid result for (Tuple2 item : data) { - Assert.assertArrayEquals(item.f1, db.get(handle, item.f0)); + assertThat(db.get(handle, item.f0)).isEqualTo(item.f1); } } } @@ -144,8 +151,8 @@ public void basicTest() throws Exception { * preconfigured value. */ @Test - public void testWriteBatchWrapperFlushAfterMemorySizeExceed() throws Exception { - try (RocksDB db = RocksDB.open(folder.newFolder().getAbsolutePath()); + void testWriteBatchWrapperFlushAfterMemorySizeExceed(@TempDir File folder) throws Exception { + try (RocksDB db = RocksDB.open(folder.getAbsolutePath()); WriteOptions options = new WriteOptions().setDisableWAL(true); ColumnFamilyHandle handle = db.createColumnFamily(new ColumnFamilyDescriptor("test".getBytes())); @@ -159,12 +166,12 @@ public void testWriteBatchWrapperFlushAfterMemorySizeExceed() throws Exception { // format is [handleType|kvType|keyLen|key|valueLen|value] // more information please ref write_batch.cc in RocksDB writeBatchWrapper.put(handle, dummy, dummy); - assertEquals(initBatchSize + 16, writeBatchWrapper.getDataSize()); + assertThat(writeBatchWrapper.getDataSize()).isEqualTo(initBatchSize + 16); writeBatchWrapper.put(handle, dummy, dummy); - assertEquals(initBatchSize + 32, writeBatchWrapper.getDataSize()); + assertThat(writeBatchWrapper.getDataSize()).isEqualTo(initBatchSize + 32); writeBatchWrapper.put(handle, dummy, dummy); // will flush all, then an empty write batch - assertEquals(initBatchSize, writeBatchWrapper.getDataSize()); + assertThat(writeBatchWrapper.getDataSize()).isEqualTo(initBatchSize); } } @@ -173,8 +180,8 @@ public void testWriteBatchWrapperFlushAfterMemorySizeExceed() throws Exception { * preconfigured value. */ @Test - public void testWriteBatchWrapperFlushAfterCountExceed() throws Exception { - try (RocksDB db = RocksDB.open(folder.newFolder().getAbsolutePath()); + void testWriteBatchWrapperFlushAfterCountExceed(@TempDir File folder) throws Exception { + try (RocksDB db = RocksDB.open(folder.getAbsolutePath()); WriteOptions options = new WriteOptions().setDisableWAL(true); ColumnFamilyHandle handle = db.createColumnFamily(new ColumnFamilyDescriptor("test".getBytes())); @@ -186,10 +193,10 @@ public void testWriteBatchWrapperFlushAfterCountExceed() throws Exception { for (int i = 1; i < 100; ++i) { writeBatchWrapper.put(handle, dummy, dummy); // each kv consumes 8 bytes - assertEquals(initBatchSize + 8 * i, writeBatchWrapper.getDataSize()); + assertThat(writeBatchWrapper.getDataSize()).isEqualTo(initBatchSize + 8 * i); } writeBatchWrapper.put(handle, dummy, dummy); - assertEquals(initBatchSize, writeBatchWrapper.getDataSize()); + assertThat(writeBatchWrapper.getDataSize()).isEqualTo(initBatchSize); } } @@ -198,16 +205,16 @@ public void testWriteBatchWrapperFlushAfterCountExceed() throws Exception { * WAL and closes them correctly. */ @Test - public void testDefaultWriteOptionsHaveDisabledWAL() throws Exception { + void testDefaultWriteOptionsHaveDisabledWAL(@TempDir File folder) throws Exception { WriteOptions options; - try (RocksDB db = RocksDB.open(folder.newFolder().getAbsolutePath()); + try (RocksDB db = RocksDB.open(folder.getAbsolutePath()); RocksDBWriteBatchWrapper writeBatchWrapper = new RocksDBWriteBatchWrapper(db, null, 200, 50)) { options = writeBatchWrapper.getOptions(); - assertTrue(options.isOwningHandle()); - assertTrue(options.disableWAL()); + assertThat(options.isOwningHandle()).isTrue(); + assertThat(options.disableWAL()).isTrue(); } - assertFalse(options.isOwningHandle()); + assertThat(options.isOwningHandle()).isFalse(); } /** @@ -215,16 +222,16 @@ public void testDefaultWriteOptionsHaveDisabledWAL() throws Exception { * not close them. */ @Test - public void testNotClosingPassedInWriteOption() throws Exception { + void testNotClosingPassedInWriteOption(@TempDir File folder) throws Exception { try (WriteOptions passInOption = new WriteOptions().setDisableWAL(false)) { - try (RocksDB db = RocksDB.open(folder.newFolder().getAbsolutePath()); + try (RocksDB db = RocksDB.open(folder.getAbsolutePath()); RocksDBWriteBatchWrapper writeBatchWrapper = new RocksDBWriteBatchWrapper(db, passInOption, 200, 50)) { WriteOptions options = writeBatchWrapper.getOptions(); - assertTrue(options.isOwningHandle()); - assertFalse(options.disableWAL()); + assertThat(options.isOwningHandle()).isTrue(); + assertThat(options.disableWAL()).isFalse(); } - assertTrue(passInOption.isOwningHandle()); + assertThat(passInOption.isOwningHandle()).isTrue(); } } } diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDbMultiClassLoaderTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDbMultiClassLoaderTest.java index de72cbbba3bad9..986e2791f67caa 100644 --- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDbMultiClassLoaderTest.java +++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDbMultiClassLoaderTest.java @@ -20,28 +20,26 @@ import org.apache.flink.util.FlinkUserCodeClassLoaders; -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 org.rocksdb.RocksDB; import java.lang.reflect.Method; import java.net.URL; +import java.nio.file.Path; 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 RocksDB JNI library loading works properly in the presence of the * RocksDB code being loaded dynamically via reflection. That can happen when RocksDB is in the user * code JAR, or in certain test setups. */ -public class RocksDbMultiClassLoaderTest { - - @Rule public final TemporaryFolder tmp = new TemporaryFolder(); +class RocksDbMultiClassLoaderTest { @Test - public void testTwoSeparateClassLoaders() throws Exception { + void testTwoSeparateClassLoaders(@TempDir Path tmp) throws Exception { // collect the libraries / class folders with RocksDB related code: the state backend and // RocksDB itself final URL codePath1 = @@ -71,13 +69,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 = tmp.toFile().getAbsolutePath(); final Method meth1 = clazz1.getDeclaredMethod("ensureRocksDBIsLoaded", String.class); final Method meth2 = clazz2.getDeclaredMethod("ensureRocksDBIsLoaded", String.class); diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksIncrementalCheckpointRescalingTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksIncrementalCheckpointRescalingTest.java index dec046d7ff9941..d422f577b7275e 100644 --- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksIncrementalCheckpointRescalingTest.java +++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksIncrementalCheckpointRescalingTest.java @@ -35,33 +35,23 @@ 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.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 java.util.Arrays; -import java.util.Collection; -import java.util.List; -/** Tests to guard rescaling from checkpoint. */ -@RunWith(Parameterized.class) -public class RocksIncrementalCheckpointRescalingTest extends TestLogger { +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.nio.file.Path; +import java.util.List; - @Rule public TemporaryFolder rootFolder = new TemporaryFolder(); +import static org.assertj.core.api.Assertions.assertThat; - @Parameterized.Parameters(name = "useIngestDbRestoreMode: {0}") - public static Collection parameters() { - return Arrays.asList(false, true); - } +/** Tests to guard rescaling from checkpoint. */ +class RocksIncrementalCheckpointRescalingTest { - @Parameterized.Parameter public boolean useIngestDbRestoreMode; + @TempDir private Path rootFolder; private final int maxParallelism = 10; @@ -69,73 +59,74 @@ 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)) + .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 + @ParameterizedTest(name = "useIngestDbRestoreMode: {0}") + @ValueSource(booleans = {false, true}) @SuppressWarnings("unchecked") - public void testScalingUp() throws Exception { + void testScalingUp(boolean useIngestDbRestoreMode) throws Exception { // -----------------------------------------> test with initial parallelism 1 // <--------------------------------------- @@ -144,10 +135,10 @@ public void testScalingUp() throws Exception { try (KeyedOneInputStreamOperatorTestHarness harness = getHarnessTest(keySelector, maxParallelism, 1, 0)) { - harness.setStateBackend(getStateBackend()); + harness.setStateBackend(getStateBackend(useIngestDbRestoreMode)); harness.setCheckpointStorage( new FileSystemCheckpointStorage( - "file://" + rootFolder.newFolder().getAbsolutePath())); + "file://" + TempDirUtils.newFolder(rootFolder).getAbsolutePath())); harness.open(); validHarnessResult(harness, 1, records); @@ -179,24 +170,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].setStateBackend(getStateBackend(useIngestDbRestoreMode)); 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].setStateBackend(getStateBackend(useIngestDbRestoreMode)); harness2[1].setCheckpointStorage( new FileSystemCheckpointStorage( - "file://" + rootFolder.newFolder().getAbsolutePath())); + "file://" + TempDirUtils.newFolder(rootFolder).getAbsolutePath())); harness2[1].setup(); harness2[1].initializeState(initState2); harness2[1].open(); @@ -242,36 +233,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].setStateBackend(getStateBackend(useIngestDbRestoreMode)); 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].setStateBackend(getStateBackend(useIngestDbRestoreMode)); 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].setStateBackend(getStateBackend(useIngestDbRestoreMode)); harness3[2].setCheckpointStorage( new FileSystemCheckpointStorage( - "file://" + rootFolder.newFolder().getAbsolutePath())); + "file://" + TempDirUtils.newFolder(rootFolder).getAbsolutePath())); harness3[2].setup(); harness3[2].initializeState(initState3); harness3[2].open(); @@ -284,9 +275,10 @@ public void testScalingUp() throws Exception { } } - @Test + @ParameterizedTest(name = "useIngestDbRestoreMode: {0}") + @ValueSource(booleans = {false, true}) @SuppressWarnings("unchecked") - public void testScalingDown() throws Exception { + void testScalingDown(boolean useIngestDbRestoreMode) throws Exception { // -----------------------------------------> test with initial parallelism 3 // <--------------------------------------- @@ -301,32 +293,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].setStateBackend(getStateBackend(useIngestDbRestoreMode)); 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].setStateBackend(getStateBackend(useIngestDbRestoreMode)); 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].setStateBackend(getStateBackend(useIngestDbRestoreMode)); 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]); @@ -369,12 +361,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].setStateBackend(getStateBackend(useIngestDbRestoreMode)); harness2[0].setCheckpointStorage( new FileSystemCheckpointStorage( - "file://" + rootFolder.newFolder().getAbsolutePath())); + "file://" + TempDirUtils.newFolder(rootFolder).getAbsolutePath())); harness2[0].setup(); harness2[0].initializeState(initState1); harness2[0].open(); @@ -382,12 +374,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].setStateBackend(getStateBackend(useIngestDbRestoreMode)); harness2[1].setCheckpointStorage( new FileSystemCheckpointStorage( - "file://" + rootFolder.newFolder().getAbsolutePath())); + "file://" + TempDirUtils.newFolder(rootFolder).getAbsolutePath())); harness2[1].setup(); harness2[1].initializeState(initState2); harness2[1].open(); @@ -418,10 +410,10 @@ public void testScalingDown() throws Exception { // this will choose the state handle generated by harness2[0] to init the target db // without any clipping. - harness.setStateBackend(getStateBackend()); + harness.setStateBackend(getStateBackend(useIngestDbRestoreMode)); harness.setCheckpointStorage( new FileSystemCheckpointStorage( - "file://" + rootFolder.newFolder().getAbsolutePath())); + "file://" + TempDirUtils.newFolder(rootFolder).getAbsolutePath())); harness.setup(); harness.initializeState(initState1); harness.open(); @@ -448,8 +440,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); } } @@ -468,7 +460,7 @@ private KeyedOneInputStreamOperatorTestHarness getHarne subtaskIdx); } - private StateBackend getStateBackend() throws Exception { + private StateBackend getStateBackend(boolean useIngestDbRestoreMode) throws Exception { EmbeddedRocksDBStateBackend rocksDBStateBackend = new EmbeddedRocksDBStateBackend(true); Configuration configuration = new Configuration(); configuration.set( diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksKeyGroupsRocksSingleStateIteratorTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksKeyGroupsRocksSingleStateIteratorTest.java index fa4a08f6b6619a..5caf4bdf122a66 100644 --- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksKeyGroupsRocksSingleStateIteratorTest.java +++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksKeyGroupsRocksSingleStateIteratorTest.java @@ -23,13 +23,12 @@ import org.apache.flink.core.fs.CloseableRegistry; import org.apache.flink.core.memory.ByteArrayOutputStreamWithPos; import org.apache.flink.state.rocksdb.iterator.RocksStatesPerKeyGroupMergeIterator; +import org.apache.flink.testutils.junit.utils.TempDirUtils; import org.apache.flink.util.IOUtils; -import org.junit.Assert; -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 org.rocksdb.ColumnFamilyDescriptor; import org.rocksdb.ColumnFamilyHandle; import org.rocksdb.NativeLibraryLoader; @@ -38,54 +37,58 @@ import java.io.DataOutputStream; import java.nio.ByteBuffer; +import java.nio.file.Path; import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.Random; +import static org.assertj.core.api.Assertions.assertThat; + /** Tests for the RocksStatesPerKeyGroupMergeIterator. */ -public class RocksKeyGroupsRocksSingleStateIteratorTest { +class RocksKeyGroupsRocksSingleStateIteratorTest { private static final int NUM_KEY_VAL_STATES = 50; private static final int MAX_NUM_KEYS = 20; - @Rule public TemporaryFolder tempFolder = new TemporaryFolder(); + @TempDir private Path tempFolder; - @Before - public void before() throws Exception { - NativeLibraryLoader.getInstance().loadLibrary(tempFolder.newFolder().getAbsolutePath()); + @BeforeEach + void before() throws Exception { + NativeLibraryLoader.getInstance() + .loadLibrary(TempDirUtils.newFolder(tempFolder).getAbsolutePath()); } @Test - public void testEmptyMergeIterator() throws Exception { + void testEmptyMergeIterator() throws Exception { RocksStatesPerKeyGroupMergeIterator emptyIterator = new RocksStatesPerKeyGroupMergeIterator( new CloseableRegistry(), Collections.emptyList(), Collections.emptyList(), 2); - Assert.assertFalse(emptyIterator.isValid()); + assertThat(emptyIterator.isValid()).isFalse(); } @Test - public void testMergeIteratorByte() throws Exception { - Assert.assertTrue(MAX_NUM_KEYS <= Byte.MAX_VALUE); + void testMergeIteratorByte() throws Exception { + assertThat(MAX_NUM_KEYS).isLessThanOrEqualTo(Byte.MAX_VALUE); testMergeIterator(Byte.MAX_VALUE); } @Test - public void testMergeIteratorShort() throws Exception { - Assert.assertTrue(MAX_NUM_KEYS <= Byte.MAX_VALUE); + void testMergeIteratorShort() throws Exception { + assertThat(MAX_NUM_KEYS).isLessThanOrEqualTo(Byte.MAX_VALUE); testMergeIterator(Short.MAX_VALUE); } - public void testMergeIterator(int maxParallelism) throws Exception { + void testMergeIterator(int maxParallelism) throws Exception { Random random = new Random(1234); try (ReadOptions readOptions = new ReadOptions(); - RocksDB rocksDB = RocksDB.open(tempFolder.getRoot().getAbsolutePath())) { + RocksDB rocksDB = RocksDB.open(tempFolder.toFile().getAbsolutePath())) { List> rocksIteratorsWithKVStateId = new ArrayList<>(); List> columnFamilyHandlesWithKeyCount = @@ -151,12 +154,11 @@ public void testMergeIterator(int maxParallelism) throws Exception { int keyGroup = maxParallelism > Byte.MAX_VALUE ? bb.getShort() : bb.get(); int key = bb.getInt(); - Assert.assertTrue(keyGroup >= prevKeyGroup); - Assert.assertTrue(key >= prevKey); - Assert.assertEquals(prevKeyGroup != keyGroup, mergeIterator.isNewKeyGroup()); - Assert.assertEquals( - prevKVState != mergeIterator.kvStateId(), - mergeIterator.isNewKeyValueState()); + assertThat(keyGroup).isGreaterThanOrEqualTo(prevKeyGroup); + assertThat(key).isGreaterThanOrEqualTo(prevKey); + assertThat(mergeIterator.isNewKeyGroup()).isEqualTo(prevKeyGroup != keyGroup); + assertThat(mergeIterator.isNewKeyValueState()) + .isEqualTo(prevKVState != mergeIterator.kvStateId()); prevKeyGroup = keyGroup; prevKVState = mergeIterator.kvStateId(); @@ -165,7 +167,7 @@ public void testMergeIterator(int maxParallelism) throws Exception { ++totalKeysActual; } - Assert.assertEquals(totalKeysExpected, totalKeysActual); + assertThat(totalKeysActual).isEqualTo(totalKeysExpected); } IOUtils.closeQuietly(rocksDB.getDefaultColumnFamily()); diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/restore/DistributeStateHandlerHelperTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/restore/DistributeStateHandlerHelperTest.java index 93daf0455e4586..70996b1208cb46 100644 --- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/restore/DistributeStateHandlerHelperTest.java +++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/restore/DistributeStateHandlerHelperTest.java @@ -32,7 +32,6 @@ import org.apache.flink.runtime.state.memory.ByteStreamStateHandle; import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot; import org.apache.flink.types.Either; -import org.apache.flink.util.TestLogger; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; @@ -60,7 +59,7 @@ import static org.assertj.core.api.Assertions.assertThat; /** Test class for {@link DistributeStateHandlerHelper}. */ -public class DistributeStateHandlerHelperTest extends TestLogger { +class DistributeStateHandlerHelperTest { private static final int NUM_KEY_GROUPS = 128; private static final KeyGroupRange KEY_GROUP_RANGE = new KeyGroupRange(0, NUM_KEY_GROUPS - 1); @@ -72,7 +71,7 @@ public class DistributeStateHandlerHelperTest extends TestLogger { /** Test whether sst files are exported when the key group all in range. */ @Test - public void testAutoCompactionIsDisabled() throws Exception { + void testAutoCompactionIsDisabled() throws Exception { Path rocksDir = tempDir.resolve("rocksdb_dir"); Path dbPath = rocksDir.resolve("db"); Path chkDir = rocksDir.resolve("chk");