You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gu...@apache.org on 2023/05/25 08:30:51 UTC
[flink] 08/08: [FLINK-32161][test] Migrate and remove legacy WorkingDirectoryResource.
This is an automated email from the ASF dual-hosted git repository.
guoweijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit b783da3ead00dd9b7dc5fc1410ff5a63a1494e0e
Author: Weijie Guo <re...@163.com>
AuthorDate: Tue May 23 18:46:44 2023 +0800
[FLINK-32161][test] Migrate and remove legacy WorkingDirectoryResource.
---
.../TaskExecutorLocalStateStoresManagerTest.java | 143 ++++++++++----------
.../taskexecutor/TaskManagerRunnerStartupTest.java | 144 +++++++++++----------
.../testutils/WorkingDirectoryExtension.java | 12 +-
.../testutils/WorkingDirectoryResource.java | 45 -------
4 files changed, 160 insertions(+), 184 deletions(-)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/TaskExecutorLocalStateStoresManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/TaskExecutorLocalStateStoresManagerTest.java
index ef10986e28e..375b1dfd5af 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/TaskExecutorLocalStateStoresManagerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/TaskExecutorLocalStateStoresManagerTest.java
@@ -21,6 +21,7 @@ package org.apache.flink.runtime.state;
import org.apache.flink.api.common.JobID;
import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.testutils.AllCallbackWrapper;
import org.apache.flink.runtime.blob.VoidPermanentBlobService;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
@@ -30,18 +31,17 @@ import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
import org.apache.flink.runtime.taskexecutor.TaskExecutorResourceUtils;
import org.apache.flink.runtime.taskexecutor.TaskManagerServices;
import org.apache.flink.runtime.taskexecutor.TaskManagerServicesConfiguration;
-import org.apache.flink.runtime.testutils.WorkingDirectoryResource;
+import org.apache.flink.runtime.testutils.WorkingDirectoryExtension;
+import org.apache.flink.testutils.junit.utils.TempDirUtils;
import org.apache.flink.util.FileUtils;
import org.apache.flink.util.Reference;
-import org.apache.flink.util.TestLogger;
import org.apache.flink.util.concurrent.Executors;
import org.apache.flink.shaded.guava30.com.google.common.collect.Sets;
-import org.junit.Assert;
-import org.junit.ClassRule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
import java.io.File;
import java.io.IOException;
@@ -53,13 +53,14 @@ import java.util.Collection;
import static org.assertj.core.api.Assertions.assertThat;
/** Test for {@link TaskExecutorLocalStateStoresManager}. */
-public class TaskExecutorLocalStateStoresManagerTest extends TestLogger {
+class TaskExecutorLocalStateStoresManagerTest {
- @ClassRule public static TemporaryFolder temporaryFolder = new TemporaryFolder();
+ @TempDir public static File temporaryFolder;
- @ClassRule
- public static final WorkingDirectoryResource WORKING_DIRECTORY_RESOURCE =
- new WorkingDirectoryResource();
+ @RegisterExtension
+ public static final AllCallbackWrapper<WorkingDirectoryExtension>
+ WORKING_DIRECTORY_EXTENSION_WRAPPER =
+ new AllCallbackWrapper<>(new WorkingDirectoryExtension(() -> temporaryFolder));
/**
* This tests that the creation of {@link TaskManagerServices} correctly creates the local state
@@ -67,11 +68,11 @@ public class TaskExecutorLocalStateStoresManagerTest extends TestLogger {
* directory.
*/
@Test
- public void testCreationFromConfig() throws Exception {
+ void testCreationFromConfig() throws Exception {
final Configuration config = new Configuration();
- File newFolder = temporaryFolder.newFolder();
+ File newFolder = TempDirUtils.newFolder(temporaryFolder.toPath());
String tmpDir = newFolder.getAbsolutePath() + File.separator;
final String rootDirString =
"__localStateRoot1,__localStateRoot2,__localStateRoot3".replaceAll("__", tmpDir);
@@ -84,7 +85,9 @@ public class TaskExecutorLocalStateStoresManagerTest extends TestLogger {
config.setBoolean(CheckpointingOptions.LOCAL_RECOVERY, true);
final WorkingDirectory workingDirectory =
- WORKING_DIRECTORY_RESOURCE.createNewWorkingDirectory();
+ WORKING_DIRECTORY_EXTENSION_WRAPPER
+ .getCustomExtension()
+ .createNewWorkingDirectory();
TaskManagerServices taskManagerServices =
createTaskManagerServices(
createTaskManagerServiceConfiguration(config, workingDirectory),
@@ -102,7 +105,7 @@ public class TaskExecutorLocalStateStoresManagerTest extends TestLogger {
}
// verify local recovery mode
- Assert.assertTrue(taskStateManager.isLocalRecoveryEnabled());
+ assertThat(taskStateManager.isLocalRecoveryEnabled()).isTrue();
for (File rootDirectory : rootDirectories) {
FileUtils.deleteFileOrDirectory(rootDirectory);
@@ -117,12 +120,14 @@ public class TaskExecutorLocalStateStoresManagerTest extends TestLogger {
* tmp directory of the IOManager as default for the local state root directory.
*/
@Test
- public void testCreationFromConfigDefault() throws Exception {
+ void testCreationFromConfigDefault() throws Exception {
final Configuration config = new Configuration();
final WorkingDirectory workingDirectory =
- WORKING_DIRECTORY_RESOURCE.createNewWorkingDirectory();
+ WORKING_DIRECTORY_EXTENSION_WRAPPER
+ .getCustomExtension()
+ .createNewWorkingDirectory();
TaskManagerServicesConfiguration taskManagerServicesConfiguration =
createTaskManagerServiceConfiguration(config, workingDirectory);
@@ -136,25 +141,27 @@ public class TaskExecutorLocalStateStoresManagerTest extends TestLogger {
File[] localStateRootDirectories = taskStateManager.getLocalStateRootDirectories();
for (int i = 0; i < localStateRootDirectories.length; ++i) {
- Assert.assertEquals(
- workingDirectory.getLocalStateDirectory(), localStateRootDirectories[i]);
+ assertThat(localStateRootDirectories[i])
+ .isEqualTo(workingDirectory.getLocalStateDirectory());
}
- Assert.assertFalse(taskStateManager.isLocalRecoveryEnabled());
+ assertThat(taskStateManager.isLocalRecoveryEnabled()).isFalse();
} finally {
taskManagerServices.shutDown();
}
}
@Test
- public void testLocalStateNoCreateDirWhenDisabledLocalRecovery() throws Exception {
+ void testLocalStateNoCreateDirWhenDisabledLocalRecovery() throws Exception {
JobID jobID = new JobID();
JobVertexID jobVertexID = new JobVertexID();
AllocationID allocationID = new AllocationID();
int subtaskIdx = 23;
File[] rootDirs = {
- temporaryFolder.newFolder(), temporaryFolder.newFolder(), temporaryFolder.newFolder()
+ TempDirUtils.newFolder(temporaryFolder.toPath()),
+ TempDirUtils.newFolder(temporaryFolder.toPath()),
+ TempDirUtils.newFolder(temporaryFolder.toPath())
};
boolean localRecoveryEnabled = false;
@@ -173,15 +180,12 @@ public class TaskExecutorLocalStateStoresManagerTest extends TestLogger {
new Configuration(),
new Configuration());
- Assert.assertFalse(taskLocalStateStore.getLocalRecoveryConfig().isLocalRecoveryEnabled());
- Assert.assertNull(
- taskLocalStateStore
- .getLocalRecoveryConfig()
- .getLocalStateDirectoryProvider()
- .orElse(null));
+ assertThat(taskLocalStateStore.getLocalRecoveryConfig().isLocalRecoveryEnabled()).isFalse();
+ assertThat(taskLocalStateStore.getLocalRecoveryConfig().getLocalStateDirectoryProvider())
+ .isNotPresent();
for (File recoveryDir : rootDirs) {
- Assert.assertEquals(0, recoveryDir.listFiles().length);
+ assertThat(recoveryDir).isEmptyDirectory();
}
}
@@ -191,7 +195,7 @@ public class TaskExecutorLocalStateStoresManagerTest extends TestLogger {
* checks that subdirectories are correctly deleted on shutdown.
*/
@Test
- public void testSubtaskStateStoreDirectoryCreateAndDelete() throws Exception {
+ void testSubtaskStateStoreDirectoryCreateAndDelete() throws Exception {
JobID jobID = new JobID();
JobVertexID jobVertexID = new JobVertexID();
@@ -199,7 +203,9 @@ public class TaskExecutorLocalStateStoresManagerTest extends TestLogger {
int subtaskIdx = 23;
File[] rootDirs = {
- temporaryFolder.newFolder(), temporaryFolder.newFolder(), temporaryFolder.newFolder()
+ TempDirUtils.newFolder(temporaryFolder.toPath()),
+ TempDirUtils.newFolder(temporaryFolder.toPath()),
+ TempDirUtils.newFolder(temporaryFolder.toPath())
};
TaskExecutorLocalStateStoresManager storesManager =
new TaskExecutorLocalStateStoresManager(
@@ -221,44 +227,43 @@ public class TaskExecutorLocalStateStoresManagerTest extends TestLogger {
.orElseThrow(LocalRecoveryConfig.localRecoveryNotEnabled());
for (int i = 0; i < 10; ++i) {
- Assert.assertEquals(
- new File(
- rootDirs[(i & Integer.MAX_VALUE) % rootDirs.length],
- storesManager.allocationSubDirString(allocationID)),
- directoryProvider.allocationBaseDirectory(i));
+ assertThat(
+ new File(
+ rootDirs[(i & Integer.MAX_VALUE) % rootDirs.length],
+ storesManager.allocationSubDirString(allocationID)))
+ .isEqualTo(directoryProvider.allocationBaseDirectory(i));
}
long chkId = 42L;
File allocBaseDirChk42 = directoryProvider.allocationBaseDirectory(chkId);
File subtaskSpecificCheckpointDirectory =
directoryProvider.subtaskSpecificCheckpointDirectory(chkId);
- Assert.assertEquals(
- new File(
- allocBaseDirChk42,
- "jid_"
- + jobID
- + File.separator
- + "vtx_"
- + jobVertexID
- + "_"
- + "sti_"
- + subtaskIdx
- + File.separator
- + "chk_"
- + chkId),
- subtaskSpecificCheckpointDirectory);
-
- Assert.assertTrue(subtaskSpecificCheckpointDirectory.mkdirs());
+ assertThat(
+ new File(
+ allocBaseDirChk42,
+ "jid_"
+ + jobID
+ + File.separator
+ + "vtx_"
+ + jobVertexID
+ + "_"
+ + "sti_"
+ + subtaskIdx
+ + File.separator
+ + "chk_"
+ + chkId))
+ .isEqualTo(subtaskSpecificCheckpointDirectory);
+
+ assertThat(subtaskSpecificCheckpointDirectory.mkdirs()).isTrue();
File testFile = new File(subtaskSpecificCheckpointDirectory, "test");
- Assert.assertTrue(testFile.createNewFile());
+ assertThat(testFile.createNewFile()).isTrue();
// test that local recovery mode is forwarded to the created store
- Assert.assertEquals(
- storesManager.isLocalRecoveryEnabled(),
- taskLocalStateStore.getLocalRecoveryConfig().isLocalRecoveryEnabled());
+ assertThat(taskLocalStateStore.getLocalRecoveryConfig().isLocalRecoveryEnabled())
+ .isEqualTo(storesManager.isLocalRecoveryEnabled());
- Assert.assertTrue(testFile.exists());
+ assertThat(testFile).exists();
// check cleanup after releasing allocation id
storesManager.releaseLocalStateForAllocationId(allocationID);
@@ -282,9 +287,9 @@ public class TaskExecutorLocalStateStoresManagerTest extends TestLogger {
.orElseThrow(LocalRecoveryConfig.localRecoveryNotEnabled());
File chkDir = directoryProvider.subtaskSpecificCheckpointDirectory(23L);
- Assert.assertTrue(chkDir.mkdirs());
+ assertThat(chkDir.mkdirs()).isTrue();
testFile = new File(chkDir, "test");
- Assert.assertTrue(testFile.createNewFile());
+ assertThat(testFile.createNewFile()).isTrue();
// check cleanup after shutdown
storesManager.shutdown();
@@ -292,9 +297,9 @@ public class TaskExecutorLocalStateStoresManagerTest extends TestLogger {
}
@Test
- public void testOwnedLocalStateDirectoriesAreDeletedOnShutdown() throws IOException {
- final File localStateStoreA = temporaryFolder.newFolder();
- final File localStateStoreB = temporaryFolder.newFolder();
+ void testOwnedLocalStateDirectoriesAreDeletedOnShutdown() throws IOException {
+ final File localStateStoreA = TempDirUtils.newFolder(temporaryFolder.toPath());
+ final File localStateStoreB = TempDirUtils.newFolder(temporaryFolder.toPath());
final File[] localStateDirectories = {localStateStoreA, localStateStoreB};
@@ -314,9 +319,9 @@ public class TaskExecutorLocalStateStoresManagerTest extends TestLogger {
}
@Test
- public void testBorrowedLocalStateDirectoriesAreNotDeletedOnShutdown() throws IOException {
- final File localStateStoreA = temporaryFolder.newFolder();
- final File localStateStoreB = temporaryFolder.newFolder();
+ void testBorrowedLocalStateDirectoriesAreNotDeletedOnShutdown() throws IOException {
+ final File localStateStoreA = TempDirUtils.newFolder(temporaryFolder.toPath());
+ final File localStateStoreB = TempDirUtils.newFolder(temporaryFolder.toPath());
final File[] localStateDirectories = {localStateStoreA, localStateStoreB};
@@ -338,9 +343,9 @@ public class TaskExecutorLocalStateStoresManagerTest extends TestLogger {
}
@Test
- public void testRetainLocalStateForAllocationsDeletesUnretainedAllocationDirectories()
+ void testRetainLocalStateForAllocationsDeletesUnretainedAllocationDirectories()
throws IOException {
- final File localStateStore = temporaryFolder.newFolder();
+ final File localStateStore = TempDirUtils.newFolder(temporaryFolder.toPath());
final TaskExecutorLocalStateStoresManager taskExecutorLocalStateStoresManager =
new TaskExecutorLocalStateStoresManager(
true,
@@ -386,7 +391,7 @@ public class TaskExecutorLocalStateStoresManagerTest extends TestLogger {
for (File rootDir : rootDirs) {
File[] files = rootDir.listFiles();
if (files != null) {
- Assert.assertArrayEquals(new File[0], files);
+ assertThat(files).isEmpty();
}
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunnerStartupTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunnerStartupTest.java
index d2e113a4057..17980df1d03 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunnerStartupTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunnerStartupTest.java
@@ -24,6 +24,7 @@ import org.apache.flink.configuration.IllegalConfigurationException;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
import org.apache.flink.configuration.TaskManagerOptions;
+import org.apache.flink.core.testutils.AllCallbackWrapper;
import org.apache.flink.runtime.blob.NoOpTaskExecutorBlobService;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.entrypoint.WorkingDirectory;
@@ -37,27 +38,27 @@ import org.apache.flink.runtime.metrics.NoOpMetricRegistry;
import org.apache.flink.runtime.metrics.scope.ScopeFormats;
import org.apache.flink.runtime.metrics.util.TestingMetricRegistry;
import org.apache.flink.runtime.rpc.RpcService;
-import org.apache.flink.runtime.rpc.TestingRpcServiceResource;
+import org.apache.flink.runtime.rpc.TestingRpcServiceExtension;
import org.apache.flink.runtime.security.token.DelegationTokenReceiverRepository;
-import org.apache.flink.runtime.testutils.WorkingDirectoryResource;
+import org.apache.flink.runtime.testutils.WorkingDirectoryExtension;
+import org.apache.flink.testutils.junit.utils.TempDirUtils;
import org.apache.flink.util.IOUtils;
-import org.apache.flink.util.TestLogger;
import org.apache.flink.shaded.guava30.com.google.common.collect.Sets;
import org.apache.commons.io.FileUtils;
-import org.junit.After;
-import org.junit.Assume;
-import org.junit.Before;
-import org.junit.ClassRule;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assumptions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
import java.io.File;
import java.io.IOException;
import java.net.InetAddress;
import java.net.ServerSocket;
+import java.nio.file.Path;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
@@ -65,41 +66,45 @@ import java.util.regex.Pattern;
import java.util.stream.Collectors;
import java.util.stream.Stream;
-import static org.hamcrest.collection.IsIn.isIn;
-import static org.hamcrest.core.Every.everyItem;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.fail;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/**
* Tests that check how the {@link TaskManagerRunner} behaves when encountering startup problems.
*/
-public class TaskManagerRunnerStartupTest extends TestLogger {
+class TaskManagerRunnerStartupTest {
private static final String LOCAL_HOST = "localhost";
- @ClassRule
- public static final TestingRpcServiceResource RPC_SERVICE_RESOURCE =
- new TestingRpcServiceResource();
+ @RegisterExtension
+ public static final AllCallbackWrapper<TestingRpcServiceExtension>
+ RPC_SERVICE_EXTENSION_WRAPPER =
+ new AllCallbackWrapper<>(new TestingRpcServiceExtension());
- @Rule public final TemporaryFolder tempFolder = new TemporaryFolder();
+ @TempDir public Path tempFolder;
- @ClassRule
- public static final WorkingDirectoryResource WORKING_DIRECTORY_RESOURCE =
- new WorkingDirectoryResource();
+ @TempDir public static File workingDirectoryFolder;
- private final RpcService rpcService = RPC_SERVICE_RESOURCE.getTestingRpcService();
+ @RegisterExtension
+ public static final AllCallbackWrapper<WorkingDirectoryExtension>
+ WORKING_DIRECTORY_EXTENSION_WRAPPER =
+ new AllCallbackWrapper<>(
+ new WorkingDirectoryExtension(() -> workingDirectoryFolder));
+
+ private final RpcService rpcService =
+ RPC_SERVICE_EXTENSION_WRAPPER.getCustomExtension().getTestingRpcService();
private TestingHighAvailabilityServices highAvailabilityServices;
- @Before
- public void setupTest() {
+ @BeforeEach
+ void setupTest() {
highAvailabilityServices = new TestingHighAvailabilityServices();
highAvailabilityServices.setResourceManagerLeaderRetriever(
new SettableLeaderRetrievalService());
}
- @After
- public void tearDownTest() throws Exception {
+ @AfterEach
+ void tearDownTest() throws Exception {
highAvailabilityServices.closeAndCleanupAllData();
highAvailabilityServices = null;
}
@@ -109,28 +114,27 @@ public class TaskManagerRunnerStartupTest extends TestLogger {
* writable.
*/
@Test
- public void testIODirectoryNotWritable() throws Exception {
- File nonWritable = tempFolder.newFolder();
- Assume.assumeTrue(
- "Cannot create non-writable temporary file directory. Skipping test.",
- nonWritable.setWritable(false, false));
+ void testIODirectoryNotWritable() throws Exception {
+ File nonWritable = TempDirUtils.newFolder(tempFolder);
+ Assumptions.assumeTrue(
+ nonWritable.setWritable(false, false),
+ "Cannot create non-writable temporary file directory. Skipping test.");
try {
Configuration cfg = createFlinkConfiguration();
cfg.setString(CoreOptions.TMP_DIRS, nonWritable.getAbsolutePath());
- try {
-
- startTaskManager(
- cfg,
- rpcService,
- highAvailabilityServices,
- WORKING_DIRECTORY_RESOURCE.createNewWorkingDirectory());
-
- fail("Should fail synchronously with an IOException");
- } catch (IOException e) {
- // splendid!
- }
+ assertThatThrownBy(
+ () ->
+ startTaskManager(
+ cfg,
+ rpcService,
+ highAvailabilityServices,
+ WORKING_DIRECTORY_EXTENSION_WRAPPER
+ .getCustomExtension()
+ .createNewWorkingDirectory()),
+ "Should fail synchronously with an IOException")
+ .isInstanceOf(IOException.class);
} finally {
// noinspection ResultOfMethodCallIgnored
nonWritable.setWritable(true, false);
@@ -146,25 +150,30 @@ public class TaskManagerRunnerStartupTest extends TestLogger {
* Tests that the TaskManagerRunner startup fails synchronously when the memory configuration is
* wrong.
*/
- @Test(expected = IllegalConfigurationException.class)
- public void testMemoryConfigWrong() throws Exception {
+ @Test
+ void testMemoryConfigWrong() {
Configuration cfg = createFlinkConfiguration();
// something invalid
cfg.set(TaskManagerOptions.NETWORK_MEMORY_MIN, MemorySize.parse("100m"));
cfg.set(TaskManagerOptions.NETWORK_MEMORY_MAX, MemorySize.parse("10m"));
- startTaskManager(
- cfg,
- rpcService,
- highAvailabilityServices,
- WORKING_DIRECTORY_RESOURCE.createNewWorkingDirectory());
+ assertThatThrownBy(
+ () ->
+ startTaskManager(
+ cfg,
+ rpcService,
+ highAvailabilityServices,
+ WORKING_DIRECTORY_EXTENSION_WRAPPER
+ .getCustomExtension()
+ .createNewWorkingDirectory()))
+ .isInstanceOf(IllegalConfigurationException.class);
}
/**
* Tests that the TaskManagerRunner startup fails if the network stack cannot be initialized.
*/
@Test
- public void testStartupWhenNetworkStackFailsToInitialize() throws Exception {
+ void testStartupWhenNetworkStackFailsToInitialize() throws Exception {
final ServerSocket blocker = new ServerSocket(0, 50, InetAddress.getByName(LOCAL_HOST));
try {
@@ -172,15 +181,17 @@ public class TaskManagerRunnerStartupTest extends TestLogger {
cfg.setInteger(NettyShuffleEnvironmentOptions.DATA_PORT, blocker.getLocalPort());
cfg.setString(TaskManagerOptions.BIND_HOST, LOCAL_HOST);
- startTaskManager(
- cfg,
- rpcService,
- highAvailabilityServices,
- WORKING_DIRECTORY_RESOURCE.createNewWorkingDirectory());
-
- fail("Should throw IOException when the network stack cannot be initialized.");
- } catch (IOException e) {
- // ignored
+ assertThatThrownBy(
+ () ->
+ startTaskManager(
+ cfg,
+ rpcService,
+ highAvailabilityServices,
+ WORKING_DIRECTORY_EXTENSION_WRAPPER
+ .getCustomExtension()
+ .createNewWorkingDirectory()),
+ "Should throw IOException when the network stack cannot be initialized.")
+ .isInstanceOf(IOException.class);
} finally {
IOUtils.closeQuietly(blocker);
}
@@ -188,7 +199,7 @@ public class TaskManagerRunnerStartupTest extends TestLogger {
/** Checks that all expected metrics are initialized. */
@Test
- public void testMetricInitialization() throws Exception {
+ void testMetricInitialization() throws Exception {
Configuration cfg = createFlinkConfiguration();
List<String> registeredMetrics = new ArrayList<>();
@@ -196,7 +207,9 @@ public class TaskManagerRunnerStartupTest extends TestLogger {
cfg,
rpcService,
highAvailabilityServices,
- WORKING_DIRECTORY_RESOURCE.createNewWorkingDirectory(),
+ WORKING_DIRECTORY_EXTENSION_WRAPPER
+ .getCustomExtension()
+ .createNewWorkingDirectory(),
TestingMetricRegistry.builder()
.setRegisterConsumer(
(metric, metricName, group) ->
@@ -249,9 +262,8 @@ public class TaskManagerRunnerStartupTest extends TestLogger {
: Stream.empty())
.collect(Collectors.toSet());
- assertThat(
- expectedTaskManagerMetricsWithoutTaskManagerId,
- everyItem(isIn(registeredMetricsWithoutTaskManagerId)));
+ assertThat(expectedTaskManagerMetricsWithoutTaskManagerId)
+ .allSatisfy(ele -> assertThat(ele).isIn(registeredMetricsWithoutTaskManagerId));
}
// -----------------------------------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/WorkingDirectoryExtension.java b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/WorkingDirectoryExtension.java
index cf42face149..a9a9131df04 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/WorkingDirectoryExtension.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/WorkingDirectoryExtension.java
@@ -23,18 +23,22 @@ import org.apache.flink.runtime.entrypoint.WorkingDirectory;
import java.io.File;
import java.io.IOException;
+import java.util.function.Supplier;
/** Extension to generate {@link WorkingDirectory}. */
public class WorkingDirectoryExtension implements CustomExtension {
- private final File tmpDirectory;
+ private final Supplier<File> tmpDirectorySupplier;
private int counter = 0;
- public WorkingDirectoryExtension(File tmpDirectory) {
- this.tmpDirectory = tmpDirectory;
+ public WorkingDirectoryExtension(Supplier<File> tmpDirectorySupplier) {
+ // We must use supplier because @TempDir has not yet been injected when constructing this
+ // extension.
+ this.tmpDirectorySupplier = tmpDirectorySupplier;
}
public WorkingDirectory createNewWorkingDirectory() throws IOException {
- return WorkingDirectory.create(new File(tmpDirectory, "working_directory_" + counter++));
+ return WorkingDirectory.create(
+ new File(tmpDirectorySupplier.get(), "working_directory_" + counter++));
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/WorkingDirectoryResource.java b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/WorkingDirectoryResource.java
deleted file mode 100644
index ce873de6c58..00000000000
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/WorkingDirectoryResource.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.testutils;
-
-import org.apache.flink.runtime.entrypoint.WorkingDirectory;
-
-import org.junit.rules.ExternalResource;
-import org.junit.rules.TemporaryFolder;
-
-import java.io.IOException;
-
-/** External resource that can create temporary working directories. */
-public class WorkingDirectoryResource extends ExternalResource {
- private final TemporaryFolder temporaryFolder = new TemporaryFolder();
-
- @Override
- protected void before() throws Throwable {
- temporaryFolder.create();
- }
-
- @Override
- protected void after() {
- temporaryFolder.delete();
- }
-
- public WorkingDirectory createNewWorkingDirectory() throws IOException {
- return WorkingDirectory.create(temporaryFolder.newFolder());
- }
-}