This is an automated email from the ASF dual-hosted git repository.
fanrui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 3a15d1ce69a [FLINK-35601][test] Revert the junit5 migration of
InitOutputPathTest
3a15d1ce69a is described below
commit 3a15d1ce69ac21d619f60033ec45cae303489c8f
Author: Weijie Guo <[email protected]>
AuthorDate: Mon Jun 17 11:13:38 2024 +0800
[FLINK-35601][test] Revert the junit5 migration of InitOutputPathTest
PR#24881 introduce some issue during the migration to junit5, temporarily
revert the changes to `InitOutputPathTest` file.
---
.../apache/flink/core/fs/InitOutputPathTest.java | 114 +++++++++------------
1 file changed, 51 insertions(+), 63 deletions(-)
diff --git
a/flink-core/src/test/java/org/apache/flink/core/fs/InitOutputPathTest.java
b/flink-core/src/test/java/org/apache/flink/core/fs/InitOutputPathTest.java
index 71ca2a660d7..891cd123fb2 100644
--- a/flink-core/src/test/java/org/apache/flink/core/fs/InitOutputPathTest.java
+++ b/flink-core/src/test/java/org/apache/flink/core/fs/InitOutputPathTest.java
@@ -23,27 +23,31 @@ import org.apache.flink.core.fs.local.LocalDataOutputStream;
import org.apache.flink.core.fs.local.LocalFileSystem;
import org.apache.flink.core.testutils.CheckedThread;
import org.apache.flink.core.testutils.OneShotLatch;
-import org.apache.flink.testutils.junit.utils.TempDirUtils;
-import lombok.SneakyThrows;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.io.TempDir;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.lang.reflect.Field;
-import java.lang.reflect.Modifier;
-import java.nio.file.FileAlreadyExistsException;
import java.util.concurrent.locks.ReentrantLock;
-import static org.apache.flink.util.Preconditions.checkNotNull;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.Assert.fail;
+import static org.powermock.api.mockito.PowerMockito.whenNew;
/** A test validating that the initialization of local output paths is
properly synchronized. */
-class InitOutputPathTest {
+@RunWith(PowerMockRunner.class)
+@PrepareForTest(LocalFileSystem.class)
+public class InitOutputPathTest {
- @TempDir private static java.nio.file.Path tempFolder;
+ @Rule public final TemporaryFolder tempDir = new TemporaryFolder();
/**
* This test validates that this test case makes sense - that the error
can be produced in the
@@ -51,24 +55,27 @@ class InitOutputPathTest {
* latches.
*/
@Test
- void testErrorOccursUnSynchronized() throws Exception {
+ public void testErrorOccursUnSynchronized() throws Exception {
// deactivate the lock to produce the original un-synchronized state
Field lock =
FileSystem.class.getDeclaredField("OUTPUT_DIRECTORY_INIT_LOCK");
lock.setAccessible(true);
-
- Field modifiers = Field.class.getDeclaredField("modifiers");
- modifiers.setAccessible(true);
- modifiers.setInt(lock, lock.getModifiers() & ~Modifier.FINAL);
-
lock.set(null, new NoOpLock());
- // in the original un-synchronized state, we can force the race to
occur by using
- // the proper latch order to control the process of the concurrent
threads
- assertThatThrownBy(() ->
runTest(true)).isInstanceOf(FileNotFoundException.class);
- lock.set(null, new ReentrantLock(true));
+
+ try {
+ // in the original un-synchronized state, we can force the race to
occur by using
+ // the proper latch order to control the process of the concurrent
threads
+ runTest(true);
+ fail("should fail with an exception");
+ } catch (FileNotFoundException e) {
+ // expected
+ } finally {
+ // reset the proper value
+ lock.set(null, new ReentrantLock(true));
+ }
}
@Test
- void testProperSynchronized() throws Exception {
+ public void testProperSynchronized() throws Exception {
// in the synchronized variant, we cannot use the "await latches"
because not
// both threads can make process interleaved (due to the
synchronization)
// the test uses sleeps (rather than latches) to produce the same
interleaving.
@@ -80,7 +87,7 @@ class InitOutputPathTest {
}
private void runTest(final boolean useAwaits) throws Exception {
- final File tempFile = TempDirUtils.newFile(tempFolder);
+ final File tempFile = tempDir.newFile();
final Path path1 = new Path(tempFile.getAbsolutePath(), "1");
final Path path2 = new Path(tempFile.getAbsolutePath(), "2");
@@ -97,23 +104,32 @@ class InitOutputPathTest {
final OneShotLatch createAwaitLatch = new OneShotLatch();
final OneShotLatch createTriggerLatch = new OneShotLatch();
+ // this "new LocalDataOutputStream()" is in the end called by the
async threads
+ whenNew(LocalDataOutputStream.class)
+ .withAnyArguments()
+ .thenAnswer(
+ new Answer<LocalDataOutputStream>() {
+
+ @Override
+ public LocalDataOutputStream
answer(InvocationOnMock invocation)
+ throws Throwable {
+ createAwaitLatch.trigger();
+ createTriggerLatch.await();
+
+ final File file = (File)
invocation.getArguments()[0];
+ return new LocalDataOutputStream(file);
+ }
+ });
+
final LocalFileSystem fs1 =
new SyncedFileSystem(
- deleteAwaitLatch1,
- mkdirsAwaitLatch1,
- deleteTriggerLatch1,
- mkdirsTriggerLatch1,
- createAwaitLatch,
- createTriggerLatch);
+ deleteAwaitLatch1, mkdirsAwaitLatch1,
+ deleteTriggerLatch1, mkdirsTriggerLatch1);
final LocalFileSystem fs2 =
new SyncedFileSystem(
- deleteAwaitLatch2,
- mkdirsAwaitLatch2,
- deletetriggerLatch2,
- mkdirsTriggerLatch2,
- createAwaitLatch,
- createTriggerLatch);
+ deleteAwaitLatch2, mkdirsAwaitLatch2,
+ deletetriggerLatch2, mkdirsTriggerLatch2);
// start the concurrent file creators
FileCreator thread1 = new FileCreator(fs1, path1);
@@ -195,44 +211,16 @@ class InitOutputPathTest {
private final OneShotLatch deleteAwaitLatch;
private final OneShotLatch mkdirsAwaitLatch;
- private final OneShotLatch createAwaitLatch;
- private final OneShotLatch createTriggerLatch;
-
SyncedFileSystem(
OneShotLatch deleteTriggerLatch,
OneShotLatch mkdirsTriggerLatch,
OneShotLatch deleteAwaitLatch,
- OneShotLatch mkdirsAwaitLatch,
- OneShotLatch createAwaitLatch,
- OneShotLatch createTriggerLatch) {
+ OneShotLatch mkdirsAwaitLatch) {
this.deleteTriggerLatch = deleteTriggerLatch;
this.mkdirsTriggerLatch = mkdirsTriggerLatch;
this.deleteAwaitLatch = deleteAwaitLatch;
this.mkdirsAwaitLatch = mkdirsAwaitLatch;
- this.createAwaitLatch = createAwaitLatch;
- this.createTriggerLatch = createTriggerLatch;
- }
-
- @Override
- @SneakyThrows
- public FSDataOutputStream create(final Path filePath, final WriteMode
overwrite)
- throws IOException {
- checkNotNull(filePath, "filePath");
-
- if (exists(filePath) && overwrite == WriteMode.NO_OVERWRITE) {
- throw new FileAlreadyExistsException("File already exists: " +
filePath);
- }
-
- final Path parent = filePath.getParent();
- if (parent != null && !mkdirs(parent)) {
- throw new IOException("Mkdirs failed to create " + parent);
- }
-
- final File file = pathToFile(filePath);
- createAwaitLatch.trigger();
- createTriggerLatch.await();
- return new LocalDataOutputStream(file);
}
@Override
