This is an automated email from the ASF dual-hosted git repository. rong pushed a commit to branch pipe-con-issues in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 19c1669d99181a327189c0f198a4e1fd4472e649 Author: Steve Yurong Su <[email protected]> AuthorDate: Tue Aug 29 12:53:20 2023 +0800 Pipe: fix some minor concurrent issues --- .../iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java | 2 ++ .../iotdb/db/pipe/extractor/realtime/epoch/TsFileEpochManager.java | 6 +++--- .../iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java | 3 ++- 3 files changed, 7 insertions(+), 4 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java index 051e6d0437c..d1d1adf1d4e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java @@ -89,6 +89,8 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent implements TsFileIns }); } } + // check again after register close listener in case TsFile is closed during the process + isClosed.set(resource.isClosed()); } public void waitForTsFileClose() throws InterruptedException { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/epoch/TsFileEpochManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/epoch/TsFileEpochManager.java index cb94b725d10..cfa57760985 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/epoch/TsFileEpochManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/epoch/TsFileEpochManager.java @@ -29,8 +29,8 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.Collections; -import java.util.HashMap; -import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; import java.util.stream.Collectors; public class TsFileEpochManager { @@ -39,7 +39,7 @@ public class TsFileEpochManager { private static final String[] EMPTY_MEASUREMENT_ARRAY = new String[0]; - private final Map<String, TsFileEpoch> filePath2Epoch = new HashMap<>(); + private final ConcurrentMap<String, TsFileEpoch> filePath2Epoch = new ConcurrentHashMap<>(); public PipeRealtimeEvent bindPipeTsFileInsertionEvent( PipeTsFileInsertionEvent event, TsFileResource resource) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java index 45e755debb3..3cd09e73e92 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java @@ -84,6 +84,7 @@ import java.util.List; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentLinkedDeque; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock; @@ -161,7 +162,7 @@ public class TsFileProcessor { "{}: {} get flushQueryLock write lock released"; /** close file listener. */ - private final List<CloseFileListener> closeFileListeners = new ArrayList<>(); + private final List<CloseFileListener> closeFileListeners = new CopyOnWriteArrayList<>(); /** flush file listener. */ private final List<FlushListener> flushListeners = new ArrayList<>();
