You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2024/06/17 05:50:49 UTC
(paimon) branch master updated: [test] Fix flaky test for ConcurrentModificationException with TraceableFileIO (#3525)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 9c41f35b5 [test] Fix flaky test for ConcurrentModificationException with TraceableFileIO (#3525)
9c41f35b5 is described below
commit 9c41f35b54db543f2f7c09f0a8859a8548b78383
Author: Zouxxyy <zo...@alibaba-inc.com>
AuthorDate: Mon Jun 17 13:50:43 2024 +0800
[test] Fix flaky test for ConcurrentModificationException with TraceableFileIO (#3525)
---
.../org/apache/paimon/utils/TraceableFileIO.java | 23 +++++++++++++++-------
1 file changed, 16 insertions(+), 7 deletions(-)
diff --git a/paimon-core/src/test/java/org/apache/paimon/utils/TraceableFileIO.java b/paimon-core/src/test/java/org/apache/paimon/utils/TraceableFileIO.java
index 623fe3147..eb616a9ab 100644
--- a/paimon-core/src/test/java/org/apache/paimon/utils/TraceableFileIO.java
+++ b/paimon-core/src/test/java/org/apache/paimon/utils/TraceableFileIO.java
@@ -32,7 +32,6 @@ import org.apache.paimon.fs.local.LocalFileIO;
import javax.annotation.concurrent.GuardedBy;
import java.io.IOException;
-import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -251,15 +250,25 @@ public class TraceableFileIO implements FileIO {
}
public static List<SeekableInputStream> openInputStreams(Predicate<Path> filter) {
- // copy out to avoid ConcurrentModificationException
- return new ArrayList<>(OPEN_INPUT_STREAMS)
- .stream().filter(s -> filter.test(s.file)).collect(Collectors.toList());
+ LOCK.lock();
+ try {
+ return OPEN_INPUT_STREAMS.stream()
+ .filter(s -> filter.test(s.file))
+ .collect(Collectors.toList());
+ } finally {
+ LOCK.unlock();
+ }
}
public static List<PositionOutputStream> openOutputStreams(Predicate<Path> filter) {
- // copy out to avoid ConcurrentModificationException
- return new ArrayList<>(OPEN_OUTPUT_STREAMS)
- .stream().filter(s -> filter.test(s.file)).collect(Collectors.toList());
+ LOCK.lock();
+ try {
+ return OPEN_OUTPUT_STREAMS.stream()
+ .filter(s -> filter.test(s.file))
+ .collect(Collectors.toList());
+ } finally {
+ LOCK.unlock();
+ }
}
/** Loader for {@link TraceableFileIO}. */