You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ro...@apache.org on 2024/01/18 05:10:26 UTC
(iotdb) branch load-task-cleanup created (now c9fa43efc39)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a change to branch load-task-cleanup
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at c9fa43efc39 Load: Introduce LoadTsFileManager.CleanupTask to force close writer manager when exception occurs
This branch includes the following new commits:
new 5fc5d3988a7 refactor
new edbcab43bba Merge branch 'master' of https://github.com/apache/iotdb into load-task-cleanup
new c9fa43efc39 Load: Introduce LoadTsFileManager.CleanupTask to force close writer manager when exception occurs
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iotdb) 03/03: Load: Introduce LoadTsFileManager.CleanupTask to force close writer manager when exception occurs
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch load-task-cleanup
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit c9fa43efc39d4e4dc44fda775e7f0c57e18da1e1
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Thu Jan 18 13:10:18 2024 +0800
Load: Introduce LoadTsFileManager.CleanupTask to force close writer manager when exception occurs
---
.../execution/load/LoadTsFileManager.java | 155 ++++++++++++---------
1 file changed, 92 insertions(+), 63 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java
index 19239b0bfd7..1d5cbd0205f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java
@@ -19,21 +19,7 @@
package org.apache.iotdb.db.queryengine.execution.load;
-import java.io.File;
-import java.io.IOException;
-import java.nio.file.DirectoryNotEmptyException;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Objects;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.PriorityBlockingQueue;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
-import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.consensus.index.ProgressIndex;
import org.apache.iotdb.commons.file.SystemFileFactory;
@@ -44,6 +30,7 @@ import org.apache.iotdb.commons.utils.FileUtils;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.LoadFileException;
+import org.apache.iotdb.db.pipe.agent.PipeAgent;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFilePieceNode;
import org.apache.iotdb.db.queryengine.plan.scheduler.load.LoadTsFileScheduler;
import org.apache.iotdb.db.queryengine.plan.scheduler.load.LoadTsFileScheduler.LoadCommand;
@@ -54,9 +41,21 @@ import org.apache.iotdb.db.storageengine.dataregion.utils.TsFileResourceUtils;
import org.apache.iotdb.metrics.utils.MetricLevel;
import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
import org.apache.iotdb.tsfile.write.writer.TsFileIOWriter;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.DirectoryNotEmptyException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.PriorityBlockingQueue;
+
/**
* {@link LoadTsFileManager} is used for dealing with {@link LoadTsFilePieceNode} and {@link
* LoadCommand}. This class turn the content of a piece of loading TsFile into a new TsFile. When
@@ -75,22 +74,52 @@ public class LoadTsFileManager {
private final File loadDir;
- private final Map<String, TsFileWriterManager> uuid2WriterManager;
+ private final Map<String, TsFileWriterManager> uuid2WriterManager = new ConcurrentHashMap<>();
- private final ScheduledExecutorService cleanupExecutors;
- private final Map<String, ScheduledFuture<?>> uuid2Future;
- private final PriorityBlockingQueue<String> uuidQueue = new PriorityBlockingQueue<>();
+ private final Map<String, CleanupTask> uuid2CleanupTask = new ConcurrentHashMap<>();
+ private final PriorityBlockingQueue<CleanupTask> cleanupTaskQueue = new PriorityBlockingQueue<>();
public LoadTsFileManager() {
this.loadDir = SystemFileFactory.INSTANCE.getFile(CONFIG.getLoadTsFileDir());
- this.uuid2WriterManager = new ConcurrentHashMap<>();
- this.cleanupExecutors =
- IoTDBThreadPoolFactory.newScheduledThreadPool(1, LoadTsFileManager.class.getName());
- this.uuid2Future = new ConcurrentHashMap<>();
+ registerCleanupTaskExecutor();
recover();
}
+ private void registerCleanupTaskExecutor() {
+ PipeAgent.runtime()
+ .registerPeriodicalJob(
+ "LoadTsFileManager#cleanupTasks",
+ this::cleanupTasks,
+ LoadTsFileScheduler.LOAD_TASK_MAX_TIME_IN_SECOND);
+ }
+
+ private void cleanupTasks() {
+ while (!cleanupTaskQueue.isEmpty()) {
+ synchronized (uuid2CleanupTask) {
+ if (cleanupTaskQueue.isEmpty()) {
+ continue;
+ }
+
+ final CleanupTask cleanupTask = cleanupTaskQueue.peek();
+ if (cleanupTask.scheduledTime <= System.currentTimeMillis()) {
+ cleanupTask.run();
+
+ uuid2CleanupTask.remove(cleanupTask.uuid);
+ cleanupTaskQueue.poll();
+ } else {
+ final long waitTimeInMs = cleanupTask.scheduledTime - System.currentTimeMillis();
+ LOGGER.info(
+ "Next load cleanup task {} is not ready to run, wait for at least {} ms ({}s).",
+ cleanupTask.uuid,
+ waitTimeInMs,
+ waitTimeInMs / 1000.0);
+ return;
+ }
+ }
+ }
+ }
+
private void recover() {
if (!loadDir.exists()) {
return;
@@ -106,26 +135,28 @@ public class LoadTsFileManager {
uuid2WriterManager.put(uuid, writerManager);
writerManager.close();
- uuid2Future.put(
- uuid,
- cleanupExecutors.schedule(
- () -> forceCloseWriterManager(uuid),
- LoadTsFileScheduler.LOAD_TASK_MAX_TIME_IN_SECOND,
- TimeUnit.SECONDS));
+
+ synchronized (uuid2CleanupTask) {
+ final CleanupTask cleanupTask =
+ new CleanupTask(uuid, LoadTsFileScheduler.LOAD_TASK_MAX_TIME_IN_SECOND * 1000);
+ uuid2CleanupTask.put(uuid, cleanupTask);
+ cleanupTaskQueue.add(cleanupTask);
+ }
}
}
public void writeToDataRegion(DataRegion dataRegion, LoadTsFilePieceNode pieceNode, String uuid)
throws IOException {
if (!uuid2WriterManager.containsKey(uuid)) {
- uuid2Future.put(
- uuid,
- cleanupExecutors.schedule(
- () -> forceCloseWriterManager(uuid),
- LoadTsFileScheduler.LOAD_TASK_MAX_TIME_IN_SECOND,
- TimeUnit.SECONDS));
+ synchronized (uuid2CleanupTask) {
+ final CleanupTask cleanupTask =
+ new CleanupTask(uuid, LoadTsFileScheduler.LOAD_TASK_MAX_TIME_IN_SECOND * 1000);
+ uuid2CleanupTask.put(uuid, cleanupTask);
+ cleanupTaskQueue.add(cleanupTask);
+ }
}
- TsFileWriterManager writerManager =
+
+ final TsFileWriterManager writerManager =
uuid2WriterManager.computeIfAbsent(
uuid, o -> new TsFileWriterManager(SystemFileFactory.INSTANCE.getFile(loadDir, uuid)));
for (TsFileData tsFileData : pieceNode.getAllTsFileData()) {
@@ -158,29 +189,19 @@ public class LoadTsFileManager {
}
private void clean(String uuid) {
- uuid2WriterManager.get(uuid).close();
- uuid2WriterManager.remove(uuid);
- uuid2Future.get(uuid).cancel(true);
- uuid2Future.remove(uuid);
-
- final Path loadDirPath = loadDir.toPath();
- if (!Files.exists(loadDirPath)) {
- return;
- }
- try {
- Files.deleteIfExists(loadDirPath);
- LOGGER.info("Load dir {} was deleted.", loadDirPath);
- } catch (DirectoryNotEmptyException e) {
- LOGGER.info("Load dir {} is not empty, skip deleting.", loadDirPath);
- } catch (IOException e) {
- LOGGER.info(MESSAGE_DELETE_FAIL, loadDirPath);
+ synchronized (uuid2CleanupTask) {
+ final CleanupTask cleanupTask = uuid2CleanupTask.get(uuid);
+ if (cleanupTask != null) {
+ cleanupTask.cancel();
+ }
}
+
+ forceCloseWriterManager(uuid);
}
private void forceCloseWriterManager(String uuid) {
uuid2WriterManager.get(uuid).close();
uuid2WriterManager.remove(uuid);
- uuid2Future.remove(uuid);
final Path loadDirPath = loadDir.toPath();
if (!Files.exists(loadDirPath)) {
@@ -375,26 +396,38 @@ public class LoadTsFileManager {
}
}
- private static class TaskCleanupTask implements Runnable, Comparable<TaskCleanupTask> {
+ private class CleanupTask implements Runnable, Comparable<CleanupTask> {
private final String uuid;
- private final LoadTsFileManager loadTsFileManager;
-
private final long scheduledTime;
- private TaskCleanupTask(String uuid, LoadTsFileManager loadTsFileManager, long delayInMs) {
+ private volatile boolean isCanceled = false;
+
+ private CleanupTask(String uuid, long delayInMs) {
this.uuid = uuid;
- this.loadTsFileManager = loadTsFileManager;
scheduledTime = System.currentTimeMillis() + delayInMs;
}
+ public void cancel() {
+ isCanceled = true;
+ }
+
@Override
public void run() {
- loadTsFileManager.forceCloseWriterManager(uuid);
+ if (isCanceled) {
+ LOGGER.info("Load cleanup task {} is canceled.", uuid);
+ } else {
+ LOGGER.info("Load cleanup task {} starts.", uuid);
+ try {
+ forceCloseWriterManager(uuid);
+ } catch (Exception e) {
+ LOGGER.warn("Load cleanup task {} error.", uuid, e);
+ }
+ }
}
@Override
- public int compareTo(TaskCleanupTask that) {
+ public int compareTo(CleanupTask that) {
return Long.compare(this.scheduledTime, that.scheduledTime);
}
}
@@ -413,10 +446,6 @@ public class LoadTsFileManager {
return dataRegion;
}
- public TTimePartitionSlot getTimePartitionSlot() {
- return timePartitionSlot;
- }
-
@Override
public String toString() {
return String.join(
(iotdb) 02/03: Merge branch 'master' of https://github.com/apache/iotdb into load-task-cleanup
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch load-task-cleanup
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit edbcab43bbacb2c0b00e65b1c8c5cb08a3a468d0
Merge: 5fc5d3988a7 da790779d13
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Wed Jan 17 20:11:27 2024 +0800
Merge branch 'master' of https://github.com/apache/iotdb into load-task-cleanup
example/client-cpp-example/README.md | 2 +-
.../it/cluster/IoTDBClusterRestartIT.java | 70 +++
.../apache/iotdb/pipe/it/IoTDBPipeDataSinkIT.java | 46 --
.../apache/iotdb/pipe/it/IoTDBPipeNullValueIT.java | 320 +++++++++++
.../{collection-info.bat => collect-info.bat} | 206 +++----
.../tools/{collection-info.sh => collect-info.sh} | 212 ++++---
iotdb-client/client-cpp/README.md | 20 +-
...register-confignode.sh => daemon-confignode.sh} | 14 +-
.../iotdb/confignode/manager/ConfigManager.java | 79 ++-
.../pipe/coordinator/runtime/PipeMetaSyncer.java | 3 +
.../iotdb/confignode/persistence/AuthorInfo.java | 5 -
.../persistence/executor/ConfigPlanExecutor.java | 9 +-
.../partition/DatabasePartitionTable.java | 11 +
.../persistence/partition/PartitionInfo.java | 22 +-
.../persistence/partition/RegionGroup.java | 15 +
.../iotdb/confignode/service/ConfigNode.java | 2 +-
.../apache/iotdb/consensus/config/RatisConfig.java | 2 +-
.../iotdb/consensus/ratis/RatisConsensus.java | 4 +-
.../apache/iotdb/consensus/ratis/utils/Utils.java | 6 +-
.../iotdb/consensus/ratis/RatisConsensusTest.java | 50 +-
.../iotdb/consensus/ratis/RecoverReadTest.java | 4 +-
.../{register-datanode.sh => daemon-datanode.sh} | 14 +-
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 11 -
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 17 +-
.../agent/runtime/PipePeriodicalJobExecutor.java | 4 +
.../db/pipe/agent/runtime/PipeRuntimeAgent.java | 4 +-
.../db/pipe/agent/task/PipeTaskDataNodeAgent.java | 41 +-
.../thrift/async/IoTDBThriftAsyncConnector.java | 38 +-
.../apache/iotdb/db/pipe/event/EnrichedEvent.java | 41 +-
.../tablet/PipeInsertNodeTabletInsertionEvent.java | 13 +-
.../common/tablet/PipeRawTabletInsertionEvent.java | 11 +
.../tablet/TabletInsertionDataContainer.java | 152 +++--
.../common/tsfile/PipeTsFileInsertionEvent.java | 19 +-
.../pipe/extractor/IoTDBDataRegionExtractor.java | 23 +-
.../PipeHistoricalDataRegionTsFileExtractor.java | 54 +-
.../resource/tsfile/PipeTsFileResourceManager.java | 33 +-
.../pipe/resource/wal/PipeWALResourceManager.java | 41 +-
.../iotdb/db/pipe/task/PipeDataNodeTask.java | 30 +
.../subtask/connector/PipeConnectorSubtask.java | 16 +-
.../subtask/processor/PipeProcessorSubtask.java | 14 +-
.../common/header/ColumnHeaderConstant.java | 1 +
.../queryengine/execution/driver/DataDriver.java | 1 +
.../execution/driver/DataDriverContext.java | 16 +-
.../fragment/FragmentInstanceContext.java | 4 +
.../execution/load/LoadTsFileManager.java | 8 +-
.../process/join/LeftOuterTimeJoinOperator.java | 11 +-
.../queryengine/plan/execution/QueryExecution.java | 3 +
.../execution/config/metadata/ShowRegionTask.java | 7 +-
.../plan/planner/LocalExecutionPlanner.java | 7 +-
.../plan/planner/OperatorTreeGenerator.java | 4 +-
.../plan/planner/distribution/SourceRewriter.java | 8 +-
.../plan/planner/plan/LogicalQueryPlan.java | 6 +-
.../plan/planner/plan/PlanFragment.java | 5 +
.../scheduler/FragmentInstanceDispatcherImpl.java | 14 +-
.../mtree/impl/mem/mnode/info/LogicalViewInfo.java | 2 +-
.../mtree/impl/pbtree/flush/Scheduler.java | 192 +++----
.../impl/pbtree/memory/ReleaseFlushMonitor.java | 5 +-
.../mtree/impl/pbtree/schemafile/RecordUtils.java | 55 +-
.../mtree/impl/pbtree/schemafile/SchemaFile.java | 7 +-
.../impl/pbtree/schemafile/WrappedSegment.java | 13 +-
.../schemafile/pagemgr/BTreePageManager.java | 10 +-
.../pbtree/schemafile/pagemgr/IPageManager.java | 3 +
.../pbtree/schemafile/pagemgr/PageManager.java | 28 +-
.../db/service/metrics/DataNodeMetricsHelper.java | 2 +-
.../iotdb/db/service/metrics/WritingMetrics.java | 16 +-
.../db/storageengine/dataregion/DataRegion.java | 12 +-
.../impl/ReadChunkCompactionPerformer.java | 12 +-
.../execute/task/CompactionTaskSummary.java | 44 +-
.../execute/task/InnerSpaceCompactionTask.java | 2 +
.../task/InsertionCrossSpaceCompactionTask.java | 3 +-
.../execute/utils/MultiTsFileDeviceIterator.java | 4 +
.../execute/utils/executor/ModifiedStatus.java} | 31 +-
.../fast/AlignedSeriesCompactionExecutor.java | 1 +
.../fast/NonAlignedSeriesCompactionExecutor.java | 1 +
.../executor/fast/SeriesCompactionExecutor.java | 6 +-
.../ReadChunkAlignedSeriesCompactionExecutor.java | 472 +++++++++++++++
.../readchunk/SingleSeriesCompactionExecutor.java | 4 +-
.../executor/readchunk/loader/ChunkLoader.java | 104 ++++
.../readchunk/loader/InstantChunkLoader.java | 105 ++++
.../readchunk/loader/InstantPageLoader.java | 103 ++++
.../executor/readchunk/loader/PageLoader.java | 107 ++++
.../compaction/io/CompactionTsFileReader.java | 18 +
.../compaction/schedule/CompactionScheduler.java | 2 +-
.../compaction/schedule/CompactionTaskManager.java | 3 +-
.../compaction/schedule/CompactionWorker.java | 3 +
.../impl/SizeTieredCompactionSelector.java | 41 +-
.../dataregion/tsfile/TsFileResource.java | 26 +-
.../dataregion/utils/TsFileResourceUtils.java | 75 +--
.../dataregion/wal/buffer/WALBuffer.java | 2 -
.../db/tools/schema/PBTreeFileSketchTool.java | 2 +-
.../metadata/mtree/schemafile/SchemaFileTest.java | 2 +-
.../pipe/event/PipeTabletInsertionEventTest.java | 93 +--
.../join/LeftOuterTimeJoinOperatorTest.java | 290 +++++++++-
.../plan/planner/PipelineBuilderTest.java | 55 ++
.../distribution/AggregationDistributionTest.java | 33 +-
.../storageengine/dataregion/DataRegionTest.java | 6 -
.../compaction/CompactionValidationTest.java | 103 ++++
.../compaction/ReadChunkInnerCompactionTest.java | 144 +++++
.../InsertionCrossSpaceCompactionSelectorTest.java | 4 +-
.../cross/InsertionCrossSpaceCompactionTest.java | 6 +-
...nkCompactionPerformerWithAlignedSeriesTest.java | 631 +++++++++++++++++++++
.../compaction/utils/CompactionCheckerUtils.java | 8 +-
.../compaction/utils/CompactionTestFileWriter.java | 8 +-
.../resources/conf/iotdb-cluster.properties | 3 +
.../resources/conf/iotdb-common.properties | 8 +-
.../sbin/{clean-all.bat => destroy-all.bat} | 22 +-
.../sbin/{clean-all.sh => destroy-all.sh} | 59 +-
...clean-confignode.bat => destroy-confignode.bat} | 34 +-
.../{clean-confignode.sh => destroy-confignode.sh} | 7 +-
.../{clean-datanode.bat => destroy-datanode.bat} | 14 +-
.../{clean-datanode.sh => destroy-datanode.sh} | 7 +-
.../src/assembly/resources/sbin/start-all.sh | 9 +-
.../src/assembly/resources/sbin/stop-all.sh | 15 +-
.../commons/auth/authorizer/BasicAuthorizer.java | 1 -
.../iotdb/commons/auth/role/BasicRoleManager.java | 6 +
.../commons/auth/role/LocalFileRoleAccessor.java | 12 +-
.../commons/auth/role/LocalFileRoleManager.java | 6 +
.../iotdb/commons/auth/user/BasicUserManager.java | 1 +
.../commons/auth/user/LocalFileUserAccessor.java | 23 +-
.../commons/auth/user/LocalFileUserManager.java | 5 +
.../commons/pipe/agent/task/PipeTaskAgent.java | 86 ++-
.../commons/pipe/task/meta/PipeMetaKeeper.java | 4 +
.../commons/pipe/task/subtask/PipeSubtask.java | 15 +-
.../schema/view/viewExpression/ViewExpression.java | 18 +
.../service/metric/JvmGcMonitorMetrics.java | 10 +-
.../service/metric}/cpu/CpuUsageMetrics.java | 23 +-
.../org/apache/iotdb/commons/utils/FileUtils.java | 8 +-
.../file/metadata/AlignedTimeSeriesMetadata.java | 1 +
.../iotdb/tsfile/read/TsFileSequenceReader.java | 3 +-
.../tsfile/read/reader/page/AlignedPageReader.java | 7 +-
.../src/main/thrift/confignode.thrift | 1 +
pom.xml | 2 +-
132 files changed, 4132 insertions(+), 950 deletions(-)
(iotdb) 01/03: refactor
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch load-task-cleanup
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 5fc5d3988a7e343a45167faa203e417e0c71ac29
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Fri Jan 12 14:18:16 2024 +0800
refactor
---
.../execution/load/LoadTsFileManager.java | 52 ++++++++++++++++------
.../iotdb/db/storageengine/StorageEngine.java | 2 +-
2 files changed, 39 insertions(+), 15 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java
index 17907fd47f8..8455104dc12 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java
@@ -19,6 +19,19 @@
package org.apache.iotdb.db.queryengine.execution.load;
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.DirectoryNotEmptyException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.PriorityBlockingQueue;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
import org.apache.iotdb.commons.conf.IoTDBConstant;
@@ -41,23 +54,9 @@ import org.apache.iotdb.db.storageengine.dataregion.utils.TsFileResourceUtils;
import org.apache.iotdb.metrics.utils.MetricLevel;
import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
import org.apache.iotdb.tsfile.write.writer.TsFileIOWriter;
-
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.File;
-import java.io.IOException;
-import java.nio.file.DirectoryNotEmptyException;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Objects;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
-
/**
* {@link LoadTsFileManager} is used for dealing with {@link LoadTsFilePieceNode} and {@link
* LoadCommand}. This class turn the content of a piece of loading TsFile into a new TsFile. When
@@ -80,6 +79,7 @@ public class LoadTsFileManager {
private final ScheduledExecutorService cleanupExecutors;
private final Map<String, ScheduledFuture<?>> uuid2Future;
+ private final PriorityBlockingQueue<String> uuidQueue = new PriorityBlockingQueue<>();
public LoadTsFileManager() {
this.loadDir = SystemFileFactory.INSTANCE.getFile(CONFIG.getLoadTsFileDir());
@@ -375,6 +375,30 @@ public class LoadTsFileManager {
}
}
+ private static class TaskCleanupTask implements Runnable, Comparable<TaskCleanupTask> {
+
+ private final String uuid;
+ private final LoadTsFileManager loadTsFileManager;
+
+ private final long scheduledTime;
+
+ private TaskCleanupTask(String uuid, LoadTsFileManager loadTsFileManager, long delayInMs) {
+ this.uuid = uuid;
+ this.loadTsFileManager = loadTsFileManager;
+ scheduledTime = System.currentTimeMillis() + delayInMs;
+ }
+
+ @Override
+ public void run() {
+ loadTsFileManager.forceCloseWriterManager(uuid);
+ }
+
+ @Override
+ public int compareTo(TaskCleanupTask that) {
+ return Long.compare(this.scheduledTime, that.scheduledTime);
+ }
+ }
+
private static class DataPartitionInfo {
private final DataRegion dataRegion;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
index 3fe138d2dc1..f89247c26d3 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
@@ -143,7 +143,7 @@ public class StorageEngine implements IService {
private List<FlushListener> customFlushListeners = new ArrayList<>();
private int recoverDataRegionNum = 0;
- private LoadTsFileManager loadTsFileManager;
+ private volatile LoadTsFileManager loadTsFileManager;
private StorageEngine() {}