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 2021/05/17 08:09:31 UTC
[iotdb] branch rel/0.12 updated: [To rel/0.12] Fix continuous
compaction run into dead loop when unseq compaction cannot select
candidates (#3205)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch rel/0.12
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/rel/0.12 by this push:
new 3b14591 [To rel/0.12] Fix continuous compaction run into dead loop when unseq compaction cannot select candidates (#3205)
3b14591 is described below
commit 3b145912eb6ceb1ad3ca1c916968df48cf2c654f
Author: zhanglingzhe0820 <44...@qq.com>
AuthorDate: Mon May 17 16:09:00 2021 +0800
[To rel/0.12] Fix continuous compaction run into dead loop when unseq compaction cannot select candidates (#3205)
Co-authored-by: zhanglingzhe <su...@foxmail.com>
---
.../db/engine/compaction/TsFileManagement.java | 15 +++++++-------
.../level/LevelCompactionTsFileManagement.java | 23 +++++++++++-----------
2 files changed, 20 insertions(+), 18 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/TsFileManagement.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/TsFileManagement.java
index ce813ac..4c71ca7 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/TsFileManagement.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/TsFileManagement.java
@@ -194,7 +194,7 @@ public abstract class TsFileManagement {
}
}
- public synchronized void merge(
+ public synchronized boolean merge(
boolean fullMerge,
List<TsFileResource> seqMergeList,
List<TsFileResource> unSeqMergeList,
@@ -206,7 +206,7 @@ public abstract class TsFileManagement {
storageGroupName,
(System.currentTimeMillis() - mergeStartTime));
}
- return;
+ return false;
}
// wait until seq merge has finished
while (isSeqMerging) {
@@ -215,7 +215,7 @@ public abstract class TsFileManagement {
} catch (InterruptedException e) {
logger.error("{} [Compaction] shutdown", storageGroupName, e);
Thread.currentThread().interrupt();
- return;
+ return false;
}
}
isUnseqMerging = true;
@@ -223,13 +223,13 @@ public abstract class TsFileManagement {
if (seqMergeList.isEmpty()) {
logger.info("{} no seq files to be merged", storageGroupName);
isUnseqMerging = false;
- return;
+ return false;
}
if (unSeqMergeList.isEmpty()) {
logger.info("{} no unseq files to be merged", storageGroupName);
isUnseqMerging = false;
- return;
+ return false;
}
if (unSeqMergeList.size() > maxOpenFileNumInEachUnseqCompaction) {
@@ -253,7 +253,7 @@ public abstract class TsFileManagement {
logger.info(
"{} cannot select merge candidates under the budget {}", storageGroupName, budget);
isUnseqMerging = false;
- return;
+ return false;
}
// avoid pending tasks holds the metadata and streams
mergeResource.clear();
@@ -290,9 +290,10 @@ public abstract class TsFileManagement {
mergeFiles[0].size(),
mergeFiles[1].size());
}
-
+ return true;
} catch (MergeException | IOException e) {
logger.error("{} cannot select file for merge", storageGroupName, e);
+ return false;
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/level/LevelCompactionTsFileManagement.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/level/LevelCompactionTsFileManagement.java
index 6bac52a..dc37e8e 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/level/LevelCompactionTsFileManagement.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/level/LevelCompactionTsFileManagement.java
@@ -600,12 +600,12 @@ public class LevelCompactionTsFileManagement extends TsFileManagement {
if (enableUnseqCompaction
&& unseqLevelNum <= 1
&& forkedUnSequenceTsFileResources.get(0).size() > 0) {
- isMergeExecutedInCurrentTask = true;
- merge(
- isForceFullMerge,
- getTsFileListByTimePartition(true, timePartition),
- forkedUnSequenceTsFileResources.get(0),
- Long.MAX_VALUE);
+ isMergeExecutedInCurrentTask =
+ merge(
+ isForceFullMerge,
+ getTsFileListByTimePartition(true, timePartition),
+ forkedUnSequenceTsFileResources.get(0),
+ Long.MAX_VALUE);
} else {
isMergeExecutedInCurrentTask =
merge(
@@ -651,11 +651,12 @@ public class LevelCompactionTsFileManagement extends TsFileManagement {
// do not merge current unseq file level to upper level and just merge all of them to
// seq file
isSeqMerging = false;
- merge(
- isForceFullMerge,
- getTsFileListByTimePartition(true, timePartition),
- mergeResources.get(i),
- Long.MAX_VALUE);
+ isMergeExecutedInCurrentTask =
+ merge(
+ isForceFullMerge,
+ getTsFileListByTimePartition(true, timePartition),
+ mergeResources.get(i),
+ Long.MAX_VALUE);
} else {
compactionLogger = new CompactionLogger(storageGroupDir, storageGroupName);
// log source file list and target file for recover