You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by xi...@apache.org on 2023/10/11 10:24:19 UTC
[iotdb] branch master updated: Refactor the CompactionWorker part of Compaction framework (#11242)
This is an automated email from the ASF dual-hosted git repository.
xingtanzjr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 9111d69d217 Refactor the CompactionWorker part of Compaction framework (#11242)
9111d69d217 is described below
commit 9111d69d217a594caef2090f37b8abc8e749a9df
Author: Zhang.Jinrui <xi...@gmail.com>
AuthorDate: Wed Oct 11 18:24:12 2023 +0800
Refactor the CompactionWorker part of Compaction framework (#11242)
---
.../FileCannotTransitToCompactingException.java | 49 +++++++++
.../execute/task/AbstractCompactionTask.java | 34 +++---
.../execute/task/CrossSpaceCompactionTask.java | 85 ++-------------
.../execute/task/InnerSpaceCompactionTask.java | 84 ++------------
.../compaction/schedule/CompactionScheduler.java | 1 -
.../compaction/schedule/CompactionTaskManager.java | 4 -
.../compaction/schedule/CompactionWorker.java | 67 +++++++++---
.../impl/SizeTieredCompactionSelector.java | 1 -
.../compaction/settle/SettleRequestHandler.java | 1 -
.../dataregion/tsfile/TsFileResource.java | 11 --
.../db/storageengine/rescon/memory/SystemInfo.java | 9 +-
.../storageengine/dataregion/DataRegionTest.java | 2 -
.../compaction/CompactionTaskComparatorTest.java | 61 +++--------
.../compaction/CompactionTaskManagerTest.java | 26 ++---
...yControlTest.java => CompactionWorkerTest.java} | 78 +++++++------
.../CrossSpaceCompactionWithUnusualCasesTest.java | 9 --
.../compaction/FastAlignedCrossCompactionTest.java | 47 --------
.../FastCompactionPerformerWithEmptyPageTest.java | 2 -
...InconsistentCompressionTypeAndEncodingTest.java | 7 --
.../FastCrossCompactionPerformerTest.java | 17 ++-
.../FastInnerCompactionPerformerTest.java | 121 +++------------------
.../FastNonAlignedCrossCompactionTest.java | 47 --------
.../compaction/ReadChunkInnerCompactionTest.java | 19 +---
.../ReadPointAlignedCrossCompactionTest.java | 47 --------
.../ReadPointNonAlignedCrossCompactionTest.java | 47 --------
.../cross/CrossSpaceCompactionSelectorTest.java | 92 +++++++++++-----
.../CrossSpaceCompactionWithFastPerformerTest.java | 4 -
...eCompactionWithFastPerformerValidationTest.java | 51 +--------
...sSpaceCompactionWithReadPointPerformerTest.java | 4 -
...actionWithReadPointPerformerValidationTest.java | 48 --------
...eCrossSpaceCompactionWithFastPerformerTest.java | 12 +-
...sSpaceCompactionWithReadPointPerformerTest.java | 12 +-
.../inner/InnerCompactionEmptyTsFileTest.java | 2 -
.../InnerSeqCompactionWithFastPerformerTest.java | 14 +--
...nerSeqCompactionWithReadChunkPerformerTest.java | 7 +-
.../inner/InnerSpaceCompactionSelectorTest.java | 54 ++++++---
...eCrossSpaceCompactionRecoverCompatibleTest.java | 4 +-
.../SizeTieredCompactionRecoverCompatibleTest.java | 4 +-
.../utils/CompactionUpdateFileCountTest.java | 18 +--
.../utils/MultiTsFileDeviceIteratorTest.java | 7 --
40 files changed, 363 insertions(+), 846 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/exception/FileCannotTransitToCompactingException.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/exception/FileCannotTransitToCompactingException.java
new file mode 100644
index 00000000000..d14ae4a44db
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/exception/FileCannotTransitToCompactingException.java
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.storageengine.dataregion.compaction.execute.exception;
+
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+
+public class FileCannotTransitToCompactingException extends Exception {
+
+ public FileCannotTransitToCompactingException(TsFileResource f) {
+ super(
+ String.format("TsFile %s cannot transit to COMPACTING. its status: %s", f, f.getStatus()));
+ }
+
+ public FileCannotTransitToCompactingException() {}
+
+ public FileCannotTransitToCompactingException(String message) {
+ super(message);
+ }
+
+ public FileCannotTransitToCompactingException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public FileCannotTransitToCompactingException(Throwable cause) {
+ super(cause);
+ }
+
+ public FileCannotTransitToCompactingException(
+ String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
+ super(message, cause, enableSuppression, writableStackTrace);
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/AbstractCompactionTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/AbstractCompactionTask.java
index 40504650338..02412723c48 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/AbstractCompactionTask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/AbstractCompactionTask.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task;
import org.apache.iotdb.db.service.metrics.CompactionMetrics;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.exception.FileCannotTransitToCompactingException;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.ICompactionPerformer;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.CompactionUtils;
import org.apache.iotdb.db.storageengine.dataregion.compaction.schedule.CompactionTaskManager;
@@ -27,8 +28,8 @@ import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileManager;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
+import java.io.IOException;
import java.util.List;
-import java.util.concurrent.atomic.AtomicInteger;
/**
* AbstractCompactionTask is the base class for all compaction task, it carries out the execution of
@@ -41,7 +42,6 @@ public abstract class AbstractCompactionTask {
protected String dataRegionId;
protected String storageGroupName;
protected long timePartition;
- protected final AtomicInteger currentTaskNum;
protected final TsFileManager tsFileManager;
protected ICompactionPerformer performer;
protected int hashCode = -1;
@@ -59,14 +59,12 @@ public abstract class AbstractCompactionTask {
String dataRegionId,
long timePartition,
TsFileManager tsFileManager,
- AtomicInteger currentTaskNum,
long serialId) {
this(
storageGroupName,
dataRegionId,
timePartition,
tsFileManager,
- currentTaskNum,
serialId,
CompactionTaskType.NORMAL);
}
@@ -76,14 +74,12 @@ public abstract class AbstractCompactionTask {
String dataRegionId,
long timePartition,
TsFileManager tsFileManager,
- AtomicInteger currentTaskNum,
long serialId,
CompactionTaskType compactionTaskType) {
this.storageGroupName = storageGroupName;
this.dataRegionId = dataRegionId;
this.timePartition = timePartition;
this.tsFileManager = tsFileManager;
- this.currentTaskNum = currentTaskNum;
this.serialId = serialId;
this.compactionTaskType = compactionTaskType;
}
@@ -113,13 +109,11 @@ public abstract class AbstractCompactionTask {
protected abstract boolean doCompaction();
public boolean start() {
- currentTaskNum.incrementAndGet();
boolean isSuccess = false;
+ summary.start();
try {
- summary.start();
isSuccess = doCompaction();
} finally {
- this.currentTaskNum.decrementAndGet();
summary.finish(isSuccess);
CompactionTaskManager.getInstance().removeRunningTaskFuture(this);
CompactionMetrics.getInstance()
@@ -142,13 +136,21 @@ public abstract class AbstractCompactionTask {
public abstract boolean equalsOtherTask(AbstractCompactionTask otherTask);
- /**
- * Check if the compaction task is valid (selected files are not merging, closed and exist). If
- * the task is valid, then set the merging status of selected files to true.
- *
- * @return true if the task is valid else false
- */
- public abstract boolean checkValidAndSetMerging();
+ public void transitSourceFilesToMerging() throws FileCannotTransitToCompactingException {
+ for (TsFileResource f : getAllSourceTsFiles()) {
+ if (!f.setStatus(TsFileResourceStatus.COMPACTING)) {
+ throw new FileCannotTransitToCompactingException(f);
+ }
+ }
+ }
+
+ public abstract long getEstimatedMemoryCost() throws IOException;
+
+ public abstract int getProcessedFileNum();
+
+ public boolean isCompactionAllowed() {
+ return tsFileManager.isAllowCompaction();
+ }
@Override
public int hashCode() {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/CrossSpaceCompactionTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/CrossSpaceCompactionTask.java
index 3331293e3a2..c2e13051c77 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/CrossSpaceCompactionTask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/CrossSpaceCompactionTask.java
@@ -23,8 +23,6 @@ import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.db.service.metrics.CompactionMetrics;
import org.apache.iotdb.db.service.metrics.FileMetrics;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.exception.CompactionExceptionHandler;
-import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.exception.CompactionFileCountExceededException;
-import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.exception.CompactionMemoryNotEnoughException;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.exception.CompactionValidationFailedException;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.ICrossCompactionPerformer;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.FastCompactionPerformer;
@@ -37,7 +35,6 @@ import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceList;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.generator.TsFileNameGenerator;
-import org.apache.iotdb.db.storageengine.rescon.memory.SystemInfo;
import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
@@ -46,7 +43,6 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.util.ArrayList;
import java.util.List;
-import java.util.concurrent.atomic.AtomicInteger;
public class CrossSpaceCompactionTask extends AbstractCompactionTask {
private static final Logger LOGGER =
@@ -57,7 +53,6 @@ public class CrossSpaceCompactionTask extends AbstractCompactionTask {
protected TsFileResourceList unseqTsFileResourceList;
private File logFile;
protected List<TsFileResource> targetTsfileResourceList;
- protected List<TsFileResource> holdReadLockList = new ArrayList<>();
protected List<TsFileResource> holdWriteLockList = new ArrayList<>();
protected double selectedSeqFileSize = 0;
protected double selectedUnseqFileSize = 0;
@@ -69,7 +64,6 @@ public class CrossSpaceCompactionTask extends AbstractCompactionTask {
List<TsFileResource> selectedSequenceFiles,
List<TsFileResource> selectedUnsequenceFiles,
ICrossCompactionPerformer performer,
- AtomicInteger currentTaskNum,
long memoryCost,
long serialId) {
super(
@@ -77,7 +71,6 @@ public class CrossSpaceCompactionTask extends AbstractCompactionTask {
tsFileManager.getDataRegionId(),
timePartition,
tsFileManager,
- currentTaskNum,
serialId);
this.selectedSequenceFiles = selectedSequenceFiles;
this.selectedUnsequenceFiles = selectedUnsequenceFiles;
@@ -194,8 +187,8 @@ public class CrossSpaceCompactionTask extends AbstractCompactionTask {
throw new CompactionValidationFailedException("Failed to pass compaction validation");
}
- releaseReadAndLockWrite(selectedSequenceFiles);
- releaseReadAndLockWrite(selectedUnsequenceFiles);
+ lockWrite(selectedSequenceFiles);
+ lockWrite(selectedUnsequenceFiles);
for (TsFileResource sequenceResource : selectedSequenceFiles) {
if (sequenceResource.getModFile().exists()) {
@@ -280,11 +273,7 @@ public class CrossSpaceCompactionTask extends AbstractCompactionTask {
false,
true);
} finally {
- SystemInfo.getInstance().resetCompactionMemoryCost(memoryCost);
- SystemInfo.getInstance()
- .decreaseCompactionFileNumCost(
- selectedSequenceFiles.size() + selectedUnsequenceFiles.size());
- releaseAllLocksAndResetStatus();
+ releaseAllLocks();
}
return isSuccess;
}
@@ -300,15 +289,10 @@ public class CrossSpaceCompactionTask extends AbstractCompactionTask {
&& this.performer.getClass().isInstance(otherCrossCompactionTask.performer);
}
- private void releaseAllLocksAndResetStatus() {
- resetCompactionCandidateStatusForAllSourceFiles();
- for (TsFileResource tsFileResource : holdReadLockList) {
- tsFileResource.readUnlock();
- }
+ private void releaseAllLocks() {
for (TsFileResource tsFileResource : holdWriteLockList) {
tsFileResource.writeUnlock();
}
- holdReadLockList.clear();
holdWriteLockList.clear();
}
@@ -355,70 +339,21 @@ public class CrossSpaceCompactionTask extends AbstractCompactionTask {
return equalsOtherTask((CrossSpaceCompactionTask) other);
}
- private void releaseReadAndLockWrite(List<TsFileResource> tsFileResourceList) {
+ private void lockWrite(List<TsFileResource> tsFileResourceList) {
for (TsFileResource tsFileResource : tsFileResourceList) {
- tsFileResource.readUnlock();
- holdReadLockList.remove(tsFileResource);
tsFileResource.writeLock();
holdWriteLockList.add(tsFileResource);
}
}
@Override
- public boolean checkValidAndSetMerging() {
- if (!tsFileManager.isAllowCompaction()) {
- resetCompactionCandidateStatusForAllSourceFiles();
- return false;
- }
- if (!isDiskSpaceCheckPassed()) {
- LOGGER.debug(
- "cross compaction task start check failed because disk free ratio is less than disk_space_warning_threshold");
- return false;
- }
- try {
- SystemInfo.getInstance().addCompactionMemoryCost(memoryCost, 60);
- SystemInfo.getInstance()
- .addCompactionFileNum(selectedSequenceFiles.size() + selectedUnsequenceFiles.size(), 60);
- } catch (Exception e) {
- if (e instanceof InterruptedException) {
- LOGGER.warn("Interrupted when allocating memory for compaction", e);
- Thread.currentThread().interrupt();
- } else if (e instanceof CompactionMemoryNotEnoughException) {
- LOGGER.info("No enough memory for current compaction task {}", this, e);
- } else if (e instanceof CompactionFileCountExceededException) {
- LOGGER.info("No enough file num for current compaction task {}", this, e);
- SystemInfo.getInstance().resetCompactionMemoryCost(memoryCost);
- }
- resetCompactionCandidateStatusForAllSourceFiles();
- return false;
- }
-
- boolean addReadLockSuccess =
- addReadLock(selectedSequenceFiles) && addReadLock(selectedUnsequenceFiles);
- if (!addReadLockSuccess) {
- SystemInfo.getInstance().resetCompactionMemoryCost(memoryCost);
- SystemInfo.getInstance()
- .decreaseCompactionFileNumCost(
- selectedSequenceFiles.size() + selectedUnsequenceFiles.size());
- }
- return addReadLockSuccess;
+ public long getEstimatedMemoryCost() {
+ return memoryCost;
}
- private boolean addReadLock(List<TsFileResource> tsFileResourceList) {
- try {
- for (TsFileResource tsFileResource : tsFileResourceList) {
- tsFileResource.readLock();
- holdReadLockList.add(tsFileResource);
- if (!tsFileResource.setStatus(TsFileResourceStatus.COMPACTING)) {
- releaseAllLocksAndResetStatus();
- return false;
- }
- }
- } catch (Exception e) {
- releaseAllLocksAndResetStatus();
- throw e;
- }
- return true;
+ @Override
+ public int getProcessedFileNum() {
+ return selectedSequenceFiles.size() + selectedUnsequenceFiles.size();
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/InnerSpaceCompactionTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/InnerSpaceCompactionTask.java
index 5659aaf57a0..d3bd3dbb746 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/InnerSpaceCompactionTask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/InnerSpaceCompactionTask.java
@@ -24,8 +24,6 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.service.metrics.CompactionMetrics;
import org.apache.iotdb.db.service.metrics.FileMetrics;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.exception.CompactionExceptionHandler;
-import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.exception.CompactionFileCountExceededException;
-import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.exception.CompactionMemoryNotEnoughException;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.exception.CompactionValidationFailedException;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.ICompactionPerformer;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.FastCompactionPerformer;
@@ -42,7 +40,6 @@ import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceList;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.generator.TsFileNameGenerator;
-import org.apache.iotdb.db.storageengine.rescon.memory.SystemInfo;
import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
import org.apache.iotdb.tsfile.exception.write.TsFileNotCompleteException;
@@ -56,7 +53,6 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
-import java.util.concurrent.atomic.AtomicInteger;
public class InnerSpaceCompactionTask extends AbstractCompactionTask {
private static final Logger LOGGER =
@@ -73,7 +69,6 @@ public class InnerSpaceCompactionTask extends AbstractCompactionTask {
protected TsFileResourceList tsFileResourceList;
protected List<TsFileResource> targetTsFileList;
- protected boolean[] isHoldingReadLock;
protected boolean[] isHoldingWriteLock;
protected long maxModsFileSize;
@@ -86,7 +81,6 @@ public class InnerSpaceCompactionTask extends AbstractCompactionTask {
List<TsFileResource> selectedTsFileResourceList,
boolean sequence,
ICompactionPerformer performer,
- AtomicInteger currentTaskNum,
long serialId) {
this(
timePartition,
@@ -94,7 +88,6 @@ public class InnerSpaceCompactionTask extends AbstractCompactionTask {
selectedTsFileResourceList,
sequence,
performer,
- currentTaskNum,
serialId,
CompactionTaskType.NORMAL);
}
@@ -105,7 +98,6 @@ public class InnerSpaceCompactionTask extends AbstractCompactionTask {
List<TsFileResource> selectedTsFileResourceList,
boolean sequence,
ICompactionPerformer performer,
- AtomicInteger currentTaskNum,
long serialId,
CompactionTaskType compactionTaskType) {
super(
@@ -113,7 +105,6 @@ public class InnerSpaceCompactionTask extends AbstractCompactionTask {
tsFileManager.getDataRegionId(),
timePartition,
tsFileManager,
- currentTaskNum,
serialId,
compactionTaskType);
this.selectedTsFileResourceList = selectedTsFileResourceList;
@@ -122,15 +113,13 @@ public class InnerSpaceCompactionTask extends AbstractCompactionTask {
if (IoTDBDescriptor.getInstance().getConfig().isEnableCompactionMemControl()) {
if (this.performer instanceof ReadChunkCompactionPerformer) {
innerSpaceEstimator = new ReadChunkInnerCompactionEstimator();
- } else if (!sequence && this.performer instanceof FastCompactionInnerCompactionEstimator) {
+ } else if (!sequence && this.performer instanceof FastCompactionPerformer) {
innerSpaceEstimator = new FastCompactionInnerCompactionEstimator();
}
}
- isHoldingReadLock = new boolean[selectedTsFileResourceList.size()];
isHoldingWriteLock = new boolean[selectedTsFileResourceList.size()];
for (int i = 0; i < selectedTsFileResourceList.size(); ++i) {
isHoldingWriteLock[i] = false;
- isHoldingReadLock[i] = false;
}
if (sequence) {
tsFileResourceList = tsFileManager.getOrCreateSequenceListByTimePartition(timePartition);
@@ -147,10 +136,6 @@ public class InnerSpaceCompactionTask extends AbstractCompactionTask {
@Override
@SuppressWarnings({"squid:S6541", "squid:S3776", "squid:S2142"})
protected boolean doCompaction() {
- if (!tsFileManager.isAllowCompaction()) {
- return true;
- }
-
long startTime = System.currentTimeMillis();
// get resource of target file
String dataDirectory = selectedTsFileResourceList.get(0).getTsFile().getParent();
@@ -250,8 +235,6 @@ public class InnerSpaceCompactionTask extends AbstractCompactionTask {
// release the read lock of all source files, and get the write lock of them to delete them
for (int i = 0; i < selectedTsFileResourceList.size(); ++i) {
- selectedTsFileResourceList.get(i).readUnlock();
- isHoldingReadLock[i] = false;
selectedTsFileResourceList.get(i).writeLock();
isHoldingWriteLock[i] = true;
}
@@ -350,9 +333,7 @@ public class InnerSpaceCompactionTask extends AbstractCompactionTask {
isSequence());
}
} finally {
- SystemInfo.getInstance().resetCompactionMemoryCost(memoryCost);
- SystemInfo.getInstance().decreaseCompactionFileNumCost(selectedTsFileResourceList.size());
- releaseAllLocksAndResetStatus();
+ releaseAllLocks();
}
return isSuccess;
}
@@ -459,13 +440,9 @@ public class InnerSpaceCompactionTask extends AbstractCompactionTask {
* release the read lock and write lock of files if it is held, and set the merging status of
* selected files to false.
*/
- private void releaseAllLocksAndResetStatus() {
- resetCompactionCandidateStatusForAllSourceFiles();
+ private void releaseAllLocks() {
for (int i = 0; i < selectedTsFileResourceList.size(); ++i) {
TsFileResource resource = selectedTsFileResourceList.get(i);
- if (isHoldingReadLock[i]) {
- resource.readUnlock();
- }
if (isHoldingWriteLock[i]) {
resource.writeUnlock();
}
@@ -473,53 +450,16 @@ public class InnerSpaceCompactionTask extends AbstractCompactionTask {
}
@Override
- public boolean checkValidAndSetMerging() {
- if (!tsFileManager.isAllowCompaction()) {
- resetCompactionCandidateStatusForAllSourceFiles();
- return false;
- }
- if (!isDiskSpaceCheckPassed()) {
- LOGGER.debug(
- "inner compaction task start check failed because disk free ratio is less than disk_space_warning_threshold");
- return false;
- }
- try {
- for (int i = 0; i < selectedTsFileResourceList.size(); ++i) {
- TsFileResource resource = selectedTsFileResourceList.get(i);
- resource.readLock();
- isHoldingReadLock[i] = true;
- if (!resource.setStatus(TsFileResourceStatus.COMPACTING)) {
- releaseAllLocksAndResetStatus();
- return false;
- }
- }
- if (innerSpaceEstimator != null) {
- memoryCost = innerSpaceEstimator.estimateInnerCompactionMemory(selectedTsFileResourceList);
- }
- SystemInfo.getInstance().addCompactionMemoryCost(memoryCost, 60);
- SystemInfo.getInstance().addCompactionFileNum(selectedTsFileResourceList.size(), 60);
- } catch (Exception e) {
- if (e instanceof InterruptedException) {
- LOGGER.warn("Interrupted when allocating memory for compaction", e);
- Thread.currentThread().interrupt();
- } else if (e instanceof CompactionMemoryNotEnoughException) {
- LOGGER.warn("No enough memory for current compaction task {}", this, e);
- } else if (e instanceof CompactionFileCountExceededException) {
- LOGGER.warn("No enough file num for current compaction task {}", this, e);
- SystemInfo.getInstance().resetCompactionMemoryCost(memoryCost);
- }
- releaseAllLocksAndResetStatus();
- return false;
- } finally {
- try {
- if (innerSpaceEstimator != null) {
- innerSpaceEstimator.close();
- }
- } catch (IOException e) {
- LOGGER.warn("Failed to close InnerSpaceCompactionMemoryEstimator");
- }
+ public long getEstimatedMemoryCost() throws IOException {
+ if (innerSpaceEstimator != null && memoryCost == 0L) {
+ memoryCost = innerSpaceEstimator.estimateInnerCompactionMemory(selectedTsFileResourceList);
}
- return true;
+ return memoryCost;
+ }
+
+ @Override
+ public int getProcessedFileNum() {
+ return selectedTsFileResourceList.size();
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduler.java
index 0ee232c49cd..1098ef6be80 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduler.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduler.java
@@ -150,7 +150,6 @@ public class CompactionScheduler {
.getConfig()
.getCrossCompactionPerformer()
.createInstance(),
- CompactionTaskManager.currentTaskNum,
memoryCost.get(i),
tsFileManager.getNextCompactionTaskId()))) {
trySubmitCount++;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionTaskManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionTaskManager.java
index 74f3a99f483..d586bad38ca 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionTaskManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionTaskManager.java
@@ -71,8 +71,6 @@ public class CompactionTaskManager implements IService {
// The thread pool that executes the sub compaction task.
private WrappedThreadPoolExecutor subCompactionTaskExecutionPool;
- public static final AtomicInteger currentTaskNum = new AtomicInteger(0);
-
private final FixedPriorityBlockingQueue<AbstractCompactionTask> candidateCompactionTaskQueue =
new FixedPriorityBlockingQueue<>(
config.getCandidateCompactionTaskQueueSize(), new DefaultCompactionTaskComparatorImpl());
@@ -98,7 +96,6 @@ public class CompactionTaskManager implements IService {
|| config.isEnableUnseqSpaceCompaction()
|| config.isEnableCrossSpaceCompaction())) {
initThreadPool();
- currentTaskNum.set(0);
candidateCompactionTaskQueue.regsitPollLastHook(
AbstractCompactionTask::resetCompactionCandidateStatusForAllSourceFiles);
init = true;
@@ -424,7 +421,6 @@ public class CompactionTaskManager implements IService {
candidateCompactionTaskQueue.clear();
init = true;
}
- currentTaskNum.set(0);
init = true;
logger.info("Compaction task manager started.");
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionWorker.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionWorker.java
index f5bb18b5b54..36421205502 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionWorker.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionWorker.java
@@ -19,8 +19,13 @@
package org.apache.iotdb.db.storageengine.dataregion.compaction.schedule;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.exception.CompactionFileCountExceededException;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.exception.CompactionMemoryNotEnoughException;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.exception.FileCannotTransitToCompactingException;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.AbstractCompactionTask;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.CompactionTaskSummary;
+import org.apache.iotdb.db.storageengine.rescon.memory.SystemInfo;
import org.apache.iotdb.db.utils.datastructure.FixedPriorityBlockingQueue;
import org.slf4j.Logger;
@@ -28,6 +33,7 @@ import org.slf4j.LoggerFactory;
import javax.validation.constraints.NotNull;
+import java.io.IOException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
@@ -48,22 +54,55 @@ public class CompactionWorker implements Runnable {
@Override
public void run() {
while (!Thread.currentThread().isInterrupted()) {
- AbstractCompactionTask task;
- try {
- task = compactionTaskQueue.take();
- } catch (InterruptedException e) {
- log.warn("CompactionThread-{} terminates because interruption", threadId);
+ processOneCompactionTask();
+ }
+ }
+
+ private void processOneCompactionTask() {
+ AbstractCompactionTask task;
+ try {
+ task = compactionTaskQueue.take();
+ } catch (InterruptedException e) {
+ log.warn("CompactionThread-{} terminates because interruption", threadId);
+ Thread.currentThread().interrupt();
+ return;
+ }
+ long estimatedMemoryCost = 0L;
+ boolean memoryAcquired = false;
+ boolean fileHandleAcquired = false;
+ try {
+ if (task == null || !task.isCompactionAllowed()) {
+ log.info("Compaction task is not allowed to be executed by TsFileManager. Task {}", task);
return;
}
- try {
- if (task != null && task.checkValidAndSetMerging()) {
- CompactionTaskSummary summary = task.getSummary();
- CompactionTaskFuture future = new CompactionTaskFuture(summary);
- CompactionTaskManager.getInstance().recordTask(task, future);
- task.start();
- }
- } catch (Exception e) {
- log.error("CompactionWorker.run(), Exception.", e);
+ task.transitSourceFilesToMerging();
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableCompactionMemControl()) {
+ estimatedMemoryCost = task.getEstimatedMemoryCost();
+ memoryAcquired = SystemInfo.getInstance().addCompactionMemoryCost(estimatedMemoryCost, 60);
+ }
+ fileHandleAcquired =
+ SystemInfo.getInstance().addCompactionFileNum(task.getProcessedFileNum(), 60);
+ CompactionTaskSummary summary = task.getSummary();
+ CompactionTaskFuture future = new CompactionTaskFuture(summary);
+ CompactionTaskManager.getInstance().recordTask(task, future);
+ task.start();
+ } catch (FileCannotTransitToCompactingException
+ | IOException
+ | CompactionMemoryNotEnoughException
+ | CompactionFileCountExceededException e) {
+ log.info("CompactionTask {} cannot be executed. Reason: {}", task, e);
+ } catch (InterruptedException e) {
+ log.warn("InterruptedException occurred when preparing compaction task. {}", task, e);
+ Thread.currentThread().interrupt();
+ } finally {
+ if (task != null) {
+ task.resetCompactionCandidateStatusForAllSourceFiles();
+ }
+ if (memoryAcquired) {
+ SystemInfo.getInstance().resetCompactionMemoryCost(estimatedMemoryCost);
+ }
+ if (fileHandleAcquired) {
+ SystemInfo.getInstance().decreaseCompactionFileNumCost(task.getProcessedFileNum());
}
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/impl/SizeTieredCompactionSelector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/impl/SizeTieredCompactionSelector.java
index 4835e6b4b10..ded95c9332f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/impl/SizeTieredCompactionSelector.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/impl/SizeTieredCompactionSelector.java
@@ -249,7 +249,6 @@ public class SizeTieredCompactionSelector
fileResources,
sequence,
createCompactionPerformer(),
- CompactionTaskManager.currentTaskNum,
tsFileManager.getNextCompactionTaskId(),
compactionTaskType);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/settle/SettleRequestHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/settle/SettleRequestHandler.java
index dc95c5ed8cb..60652442dad 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/settle/SettleRequestHandler.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/settle/SettleRequestHandler.java
@@ -275,7 +275,6 @@ public class SettleRequestHandler {
tsFileResources,
hasSeqFiles,
performer,
- CompactionTaskManager.currentTaskNum,
tsFileManager.getNextCompactionTaskId());
try {
CompactionTaskManager.getInstance().addTaskToWaitingQueue(task);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java
index bbd3a17d017..ac29a91e959 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java
@@ -912,17 +912,6 @@ public class TsFileResource {
return ramSize;
}
- public void delete() throws IOException {
- forceMarkDeleted();
- if (file.exists()) {
- Files.delete(file.toPath());
- Files.delete(
- FSFactoryProducer.getFSFactory()
- .getFile(file.toPath() + TsFileResource.RESOURCE_SUFFIX)
- .toPath());
- }
- }
-
public long getMaxPlanIndex() {
return maxPlanIndex;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/memory/SystemInfo.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/memory/SystemInfo.java
index 50cbe144f61..1ff1244ca18 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/memory/SystemInfo.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/memory/SystemInfo.java
@@ -188,7 +188,7 @@ public class SystemInfo {
this.flushingMemTablesCost -= flushingMemTableCost;
}
- public void addCompactionFileNum(int fileNum, long timeOutInSecond)
+ public boolean addCompactionFileNum(int fileNum, long timeOutInSecond)
throws InterruptedException, CompactionFileCountExceededException {
if (fileNum > totalFileLimitForCrossTask) {
// source file num is greater than the max file num for compaction
@@ -210,13 +210,11 @@ public class SystemInfo {
Thread.sleep(100);
originFileNum = this.compactionFileNumCost.get();
}
+ return true;
}
- public void addCompactionMemoryCost(long memoryCost, long timeOutInSecond)
+ public boolean addCompactionMemoryCost(long memoryCost, long timeOutInSecond)
throws InterruptedException, CompactionMemoryNotEnoughException {
- if (!config.isEnableCompactionMemControl()) {
- return;
- }
if (memoryCost > memorySizeForCompaction) {
// required memory cost is greater than the total memory budget for compaction
throw new CompactionMemoryNotEnoughException(
@@ -239,6 +237,7 @@ public class SystemInfo {
Thread.sleep(100);
originSize = this.compactionMemoryCost.get();
}
+ return true;
}
public synchronized void resetCompactionMemoryCost(long compactionMemoryCost) {
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java
index 83a5fb700a0..4e7b678298c 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java
@@ -75,7 +75,6 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
-import java.util.concurrent.atomic.AtomicInteger;
public class DataRegionTest {
private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
@@ -777,7 +776,6 @@ public class DataRegionTest {
dataRegion.getSequenceFileList(),
true,
performer,
- new AtomicInteger(0),
0);
CompactionTaskManager.getInstance().addTaskToWaitingQueue(task);
Thread.sleep(20);
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionTaskComparatorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionTaskComparatorTest.java
index 8890cc2e488..10083d7def8 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionTaskComparatorTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionTaskComparatorTest.java
@@ -58,7 +58,6 @@ import static org.junit.Assert.assertTrue;
public class CompactionTaskComparatorTest {
private final Logger LOGGER = LoggerFactory.getLogger(CompactionTaskComparatorTest.class);
- private final AtomicInteger taskNum = new AtomicInteger(0);
private FixedPriorityBlockingQueue<AbstractCompactionTask> compactionTaskQueue =
new FixedPriorityBlockingQueue<>(1024, new DefaultCompactionTaskComparatorImpl());
private TsFileManager tsFileManager = new TsFileManager("fakeSg", "0", "/");
@@ -84,8 +83,7 @@ public class CompactionTaskComparatorTest {
new FakedTsFileResource(new File(String.format("%d-%d-0-0.tsfile", i + j, i + j)), j));
}
compactionTasks[i] =
- new FakedInnerSpaceCompactionTask(
- "fakeSg", 0, tsFileManager, taskNum, true, resources, 0);
+ new FakedInnerSpaceCompactionTask("fakeSg", 0, tsFileManager, true, resources, 0);
compactionTaskQueue.put(compactionTasks[i]);
}
@@ -107,8 +105,7 @@ public class CompactionTaskComparatorTest {
new File(String.format("%d-%d-0-0.tsfile", i + j, i + j)), j - i + 101));
}
compactionTasks[i] =
- new FakedInnerSpaceCompactionTask(
- "fakeSg", 0, tsFileManager, taskNum, true, resources, 0);
+ new FakedInnerSpaceCompactionTask("fakeSg", 0, tsFileManager, true, resources, 0);
compactionTaskQueue.put(compactionTasks[i]);
}
@@ -130,8 +127,7 @@ public class CompactionTaskComparatorTest {
new File(String.format("%d-%d-%d-0.tsfile", i + j, i + j, j - i + 101)), 1));
}
compactionTasks[i] =
- new FakedInnerSpaceCompactionTask(
- "fakeSg", 0, tsFileManager, taskNum, true, resources, 0);
+ new FakedInnerSpaceCompactionTask("fakeSg", 0, tsFileManager, true, resources, 0);
compactionTaskQueue.put(compactionTasks[i]);
}
@@ -156,8 +152,7 @@ public class CompactionTaskComparatorTest {
new File(String.format("%d-%d-%d-0.tsfile", i + j, i + j, j - i + 101)), 1));
}
compactionTasks[i] =
- new FakedInnerSpaceCompactionTask(
- "fakeSg", 0, tsFileManager, taskNum, true, resources, 0);
+ new FakedInnerSpaceCompactionTask("fakeSg", 0, tsFileManager, true, resources, 0);
limitQueue.add(compactionTasks[i]);
}
@@ -179,8 +174,7 @@ public class CompactionTaskComparatorTest {
new File(String.format("%d-%d-0-0.tsfile", i + j, i + j, j - i + 101)), 1));
}
compactionTasks[i] =
- new FakedInnerSpaceCompactionTask(
- "fakeSg", 0, tsFileManager, taskNum, true, resources, 0);
+ new FakedInnerSpaceCompactionTask("fakeSg", 0, tsFileManager, true, resources, 0);
compactionTaskQueue.put(compactionTasks[i]);
}
@@ -203,8 +197,7 @@ public class CompactionTaskComparatorTest {
new FakedTsFileResource(new File(String.format("%d-%d-0-0.tsfile", i + j, i + j)), j));
}
innerCompactionTasks[i] =
- new FakedInnerSpaceCompactionTask(
- "fakeSg", 0, tsFileManager, taskNum, true, resources, 0);
+ new FakedInnerSpaceCompactionTask("fakeSg", 0, tsFileManager, true, resources, 0);
}
for (int i = 0; i < 100; ++i) {
@@ -220,7 +213,7 @@ public class CompactionTaskComparatorTest {
}
crossCompactionTasks[i] =
new FakeCrossSpaceCompactionTask(
- "fakeSg", 0, tsFileManager, taskNum, sequenceResources, unsequenceResources, 0);
+ "fakeSg", 0, tsFileManager, sequenceResources, unsequenceResources, 0);
}
for (int i = 0; i < 100; i++) {
@@ -257,7 +250,7 @@ public class CompactionTaskComparatorTest {
}
crossCompactionTasks[i] =
new FakeCrossSpaceCompactionTask(
- "fakeSg", 0, tsFileManager, taskNum, sequenceResources, unsequenceResources, 0);
+ "fakeSg", 0, tsFileManager, sequenceResources, unsequenceResources, 0);
compactionTaskQueue.put(crossCompactionTasks[i]);
}
for (int i = 100; i < 200; ++i) {
@@ -273,7 +266,7 @@ public class CompactionTaskComparatorTest {
}
crossCompactionTasks[i] =
new FakeCrossSpaceCompactionTask(
- "fakeSg", 0, tsFileManager, taskNum, sequenceResources, unsequenceResources, 0);
+ "fakeSg", 0, tsFileManager, sequenceResources, unsequenceResources, 0);
compactionTaskQueue.put(crossCompactionTasks[i]);
}
@@ -299,7 +292,7 @@ public class CompactionTaskComparatorTest {
}
compactionTaskQueue.put(
new FakedInnerSpaceCompactionTask(
- "fakeSg" + i, 0, tsFileManagers[i], taskNum, true, resources, j));
+ "fakeSg" + i, 0, tsFileManagers[i], true, resources, j));
}
}
Map<String, AtomicInteger> taskCount = new HashMap<>();
@@ -332,8 +325,7 @@ public class CompactionTaskComparatorTest {
new FakedTsFileResource(new File(String.format("%d-%d-0-0.tsfile", i + j, i + j)), j));
}
FakedInnerSpaceCompactionTask innerTask =
- new FakedInnerSpaceCompactionTask(
- "fakeSg", 0, tsFileManager, taskNum, true, resources, 0);
+ new FakedInnerSpaceCompactionTask("fakeSg", 0, tsFileManager, true, resources, 0);
compactionTaskQueue.put(innerTask);
}
@@ -343,13 +335,7 @@ public class CompactionTaskComparatorTest {
fakedTsFileResource.getModFile().write(new Deletion(new PartialPath("root.test.d1"), 1, 1));
compactionTaskQueue.put(
new FakedInnerSpaceCompactionTask(
- "fakeSg",
- 0,
- tsFileManager,
- taskNum,
- true,
- Collections.singletonList(fakedTsFileResource),
- 0));
+ "fakeSg", 0, tsFileManager, true, Collections.singletonList(fakedTsFileResource), 0));
FakedInnerSpaceCompactionTask task = (FakedInnerSpaceCompactionTask) compactionTaskQueue.take();
Assert.assertEquals(
targetFileName, task.getSelectedTsFileResourceList().get(0).getTsFile().getName());
@@ -369,8 +355,7 @@ public class CompactionTaskComparatorTest {
new DefaultCompactionTaskComparatorImpl());
for (int i = 0; i < 10; i++) {
FakedInnerSpaceCompactionTask task =
- new FakedInnerSpaceCompactionTask(
- "fakeSg", i, tsFileManager, taskNum, true, resources1, 0);
+ new FakedInnerSpaceCompactionTask("fakeSg", i, tsFileManager, true, resources1, 0);
candidateCompactionTaskQueue.put(task);
}
@@ -403,7 +388,6 @@ public class CompactionTaskComparatorTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- taskNum,
0,
0);
candidateCompactionTaskQueue.put(task);
@@ -426,7 +410,6 @@ public class CompactionTaskComparatorTest {
"fakeSg",
1,
tsFileManager,
- taskNum,
true,
Collections.singletonList(
new FakedTsFileResource(new File(String.format("%d-%d-0-0.tsfile", 1, 1)), 1)),
@@ -437,7 +420,6 @@ public class CompactionTaskComparatorTest {
"fakeSg",
1,
tsFileManager,
- taskNum,
true,
Collections.singletonList(
new FakedTsFileResource(new File(String.format("%d-%d-0-0.tsfile", 1, 2)), 1)),
@@ -448,7 +430,6 @@ public class CompactionTaskComparatorTest {
"fakeSg",
1,
tsFileManager,
- taskNum,
true,
Collections.singletonList(
new FakedTsFileResource(new File(String.format("%d-%d-0-0.tsfile", 1, 3)), 1)),
@@ -469,7 +450,6 @@ public class CompactionTaskComparatorTest {
String storageGroupName,
long timePartition,
TsFileManager tsFileManager,
- AtomicInteger currentTaskNum,
boolean sequence,
List<TsFileResource> selectedTsFileResourceList,
long serialId) {
@@ -479,7 +459,6 @@ public class CompactionTaskComparatorTest {
selectedTsFileResourceList,
sequence,
new FastCompactionPerformer(false),
- currentTaskNum,
serialId);
}
@@ -487,7 +466,6 @@ public class CompactionTaskComparatorTest {
String storageGroupName,
long timePartition,
TsFileManager tsFileManager,
- AtomicInteger currentTaskNum,
boolean sequence,
List<TsFileResource> selectedTsFileResourceList,
long serialId,
@@ -498,7 +476,6 @@ public class CompactionTaskComparatorTest {
selectedTsFileResourceList,
sequence,
new FastCompactionPerformer(false),
- currentTaskNum,
serialId,
compactionTaskType);
}
@@ -512,11 +489,6 @@ public class CompactionTaskComparatorTest {
public boolean equalsOtherTask(AbstractCompactionTask other) {
return false;
}
-
- @Override
- public boolean checkValidAndSetMerging() {
- return true;
- }
}
private static class FakeCrossSpaceCompactionTask extends CrossSpaceCompactionTask {
@@ -525,7 +497,6 @@ public class CompactionTaskComparatorTest {
String fullStorageGroupName,
long timePartition,
TsFileManager tsFileManager,
- AtomicInteger currentTaskNum,
List<TsFileResource> selectedSequenceFiles,
List<TsFileResource> selectedUnsequenceFiles,
long serialId) {
@@ -535,7 +506,6 @@ public class CompactionTaskComparatorTest {
selectedSequenceFiles,
selectedUnsequenceFiles,
new ReadPointCompactionPerformer(),
- currentTaskNum,
0,
serialId);
}
@@ -549,11 +519,6 @@ public class CompactionTaskComparatorTest {
public boolean equalsOtherTask(AbstractCompactionTask other) {
return false;
}
-
- @Override
- public boolean checkValidAndSetMerging() {
- return true;
- }
}
private static class FakedTsFileResource extends TsFileResource {
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionTaskManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionTaskManagerTest.java
index 4013483d8e5..5781886f62e 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionTaskManagerTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionTaskManagerTest.java
@@ -43,7 +43,6 @@ import java.io.File;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.Future;
-import java.util.concurrent.atomic.AtomicInteger;
public class CompactionTaskManagerTest extends InnerCompactionTest {
static final Logger logger = LoggerFactory.getLogger(CompactionTaskManagerTest.class);
@@ -78,11 +77,9 @@ public class CompactionTaskManagerTest extends InnerCompactionTest {
new TsFileManager("root.compactionTest", "0", tempSGDir.getAbsolutePath());
tsFileManager.addAll(seqResources, true);
InnerSpaceCompactionTask task1 =
- new InnerSpaceCompactionTask(
- 0, tsFileManager, seqResources, true, performer, new AtomicInteger(0), 0);
+ new InnerSpaceCompactionTask(0, tsFileManager, seqResources, true, performer, 0);
InnerSpaceCompactionTask task2 =
- new InnerSpaceCompactionTask(
- 0, tsFileManager, seqResources, true, performer, new AtomicInteger(0), 0);
+ new InnerSpaceCompactionTask(0, tsFileManager, seqResources, true, performer, 0);
seqResources.get(0).readLock();
CompactionTaskManager manager = CompactionTaskManager.getInstance();
Future<CompactionTaskSummary> summaryFuture = null;
@@ -122,11 +119,9 @@ public class CompactionTaskManagerTest extends InnerCompactionTest {
new TsFileManager("root.compactionTest", "0", tempSGDir.getAbsolutePath());
tsFileManager.addAll(seqResources, true);
InnerSpaceCompactionTask task1 =
- new InnerSpaceCompactionTask(
- 0, tsFileManager, seqResources, true, performer, new AtomicInteger(0), 0);
+ new InnerSpaceCompactionTask(0, tsFileManager, seqResources, true, performer, 0);
InnerSpaceCompactionTask task2 =
- new InnerSpaceCompactionTask(
- 0, tsFileManager, seqResources, true, performer, new AtomicInteger(0), 0);
+ new InnerSpaceCompactionTask(0, tsFileManager, seqResources, true, performer, 0);
seqResources.get(0).readLock();
Future<CompactionTaskSummary> summaryFuture = null;
try {
@@ -164,11 +159,9 @@ public class CompactionTaskManagerTest extends InnerCompactionTest {
new TsFileManager("root.compactionTest", "0", tempSGDir.getAbsolutePath());
tsFileManager.addAll(seqResources, true);
InnerSpaceCompactionTask task1 =
- new InnerSpaceCompactionTask(
- 0, tsFileManager, seqResources, true, performer, new AtomicInteger(0), 0);
+ new InnerSpaceCompactionTask(0, tsFileManager, seqResources, true, performer, 0);
InnerSpaceCompactionTask task2 =
- new InnerSpaceCompactionTask(
- 0, tsFileManager, seqResources, true, performer, new AtomicInteger(0), 0);
+ new InnerSpaceCompactionTask(0, tsFileManager, seqResources, true, performer, 0);
CompactionTaskManager manager = CompactionTaskManager.getInstance();
seqResources.get(0).readLock();
Assert.assertTrue(manager.addTaskToWaitingQueue(task1));
@@ -192,8 +185,7 @@ public class CompactionTaskManagerTest extends InnerCompactionTest {
new TsFileManager("root.compactionTest", "0", tempSGDir.getAbsolutePath());
tsFileManager.addAll(seqResources, true);
InnerSpaceCompactionTask task1 =
- new InnerSpaceCompactionTask(
- 0, tsFileManager, seqResources, true, performer, new AtomicInteger(0), 0);
+ new InnerSpaceCompactionTask(0, tsFileManager, seqResources, true, performer, 0);
CompactionTaskManager manager = CompactionTaskManager.getInstance();
manager.restart();
seqResources.get(0).readLock();
@@ -222,8 +214,7 @@ public class CompactionTaskManagerTest extends InnerCompactionTest {
new TsFileManager("root.compactionTest", "0", tempSGDir.getAbsolutePath());
tsFileManager.addAll(seqResources, true);
InnerSpaceCompactionTask task =
- new InnerSpaceCompactionTask(
- 0, tsFileManager, seqResources, true, performer, new AtomicInteger(0), 0);
+ new InnerSpaceCompactionTask(0, tsFileManager, seqResources, true, performer, 0);
seqResources.get(0).readLock();
CompactionTaskManager.getInstance().addTaskToWaitingQueue(task);
@@ -253,7 +244,6 @@ public class CompactionTaskManagerTest extends InnerCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/MemoryControlTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionWorkerTest.java
similarity index 76%
rename from iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/MemoryControlTest.java
rename to iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionWorkerTest.java
index 26ed3625919..c3b887b209f 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/MemoryControlTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionWorkerTest.java
@@ -20,12 +20,15 @@
package org.apache.iotdb.db.storageengine.dataregion.compaction;
import org.apache.iotdb.commons.exception.MetadataException;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.AbstractCompactionTask;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.CrossSpaceCompactionTask;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.InnerSpaceCompactionTask;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.schedule.CompactionWorker;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileManager;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
import org.apache.iotdb.db.storageengine.rescon.memory.SystemInfo;
+import org.apache.iotdb.db.utils.datastructure.FixedPriorityBlockingQueue;
import org.apache.iotdb.tsfile.exception.write.WriteProcessException;
import org.junit.Assert;
@@ -37,9 +40,8 @@ import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
-import java.util.concurrent.atomic.AtomicInteger;
-public class MemoryControlTest {
+public class CompactionWorkerTest {
@Before
public void setUp()
throws IOException, WriteProcessException, MetadataException, InterruptedException {
@@ -74,11 +76,15 @@ public class MemoryControlTest {
sequenceFiles,
unsequenceFiles,
null,
- new AtomicInteger(0),
1024L * 1024L * 1024L * 50L,
0);
- boolean success = task.checkValidAndSetMerging();
- Assert.assertFalse(success);
+ CrossSpaceCompactionTask taskMock = Mockito.spy(task);
+ Mockito.doReturn(true).when(taskMock).start();
+ FixedPriorityBlockingQueue<AbstractCompactionTask> mockQueue =
+ Mockito.mock(FixedPriorityBlockingQueue.class);
+ Mockito.when(mockQueue.take()).thenReturn(taskMock).thenThrow(new InterruptedException());
+ CompactionWorker worker = new CompactionWorker(0, mockQueue);
+ worker.run();
Assert.assertEquals(0, SystemInfo.getInstance().getCompactionMemoryCost().get());
Assert.assertEquals(0, SystemInfo.getInstance().getCompactionFileNumCost().get());
for (TsFileResource tsFileResource : sequenceFiles) {
@@ -92,7 +98,7 @@ public class MemoryControlTest {
}
@Test
- public void testFailedToAllocateFileNumInCrossTask() {
+ public void testFailedToAllocateFileNumInCrossTask() throws InterruptedException {
int oldMaxCrossCompactionCandidateFileNum =
SystemInfo.getInstance().getTotalFileLimitForCrossTask();
SystemInfo.getInstance().setTotalFileLimitForCrossTask(2);
@@ -117,16 +123,14 @@ public class MemoryControlTest {
tsFileManager.addAll(unsequenceFiles, false);
CrossSpaceCompactionTask task =
new CrossSpaceCompactionTask(
- 0L,
- tsFileManager,
- sequenceFiles,
- unsequenceFiles,
- null,
- new AtomicInteger(0),
- 1000,
- 0);
- boolean success = task.checkValidAndSetMerging();
- Assert.assertFalse(success);
+ 0L, tsFileManager, sequenceFiles, unsequenceFiles, null, 1000, 0);
+ CrossSpaceCompactionTask taskMock = Mockito.spy(task);
+ Mockito.doReturn(true).when(taskMock).start();
+ FixedPriorityBlockingQueue<AbstractCompactionTask> mockQueue =
+ Mockito.mock(FixedPriorityBlockingQueue.class);
+ Mockito.when(mockQueue.take()).thenReturn(taskMock).thenThrow(new InterruptedException());
+ CompactionWorker worker = new CompactionWorker(0, mockQueue);
+ worker.run();
Assert.assertEquals(0, SystemInfo.getInstance().getCompactionMemoryCost().get());
Assert.assertEquals(0, SystemInfo.getInstance().getCompactionFileNumCost().get());
for (TsFileResource tsFileResource : sequenceFiles) {
@@ -142,13 +146,9 @@ public class MemoryControlTest {
}
}
- /**
- * AllowCompaction is false.
- *
- * @throws Exception
- */
+ /** AllowCompaction is false. */
@Test
- public void testFailedToCheckValidInCrossTask() {
+ public void testFailedToCheckValidInCrossTask() throws InterruptedException {
List<TsFileResource> sequenceFiles = new ArrayList<>();
for (int i = 1; i <= 10; i++) {
sequenceFiles.add(
@@ -164,15 +164,16 @@ public class MemoryControlTest {
TsFileResourceStatus.COMPACTION_CANDIDATE));
}
TsFileManager tsFileManager = Mockito.mock(TsFileManager.class);
- Mockito.when(tsFileManager.getStorageGroupName()).thenReturn("root.sg");
- Mockito.when(tsFileManager.getDataRegionId()).thenReturn("1");
-
+ Mockito.when(tsFileManager.isAllowCompaction()).thenReturn(false);
// fail to check valid when tsfile manager is not allowed to compaction in cross task
CrossSpaceCompactionTask task =
new CrossSpaceCompactionTask(
- 0L, tsFileManager, sequenceFiles, unsequenceFiles, null, new AtomicInteger(0), 1000, 0);
- boolean success = task.checkValidAndSetMerging();
- Assert.assertFalse(success);
+ 0L, tsFileManager, sequenceFiles, unsequenceFiles, null, 1000, 0);
+ FixedPriorityBlockingQueue<AbstractCompactionTask> mockQueue =
+ Mockito.mock(FixedPriorityBlockingQueue.class);
+ Mockito.when(mockQueue.take()).thenReturn(task).thenThrow(new InterruptedException());
+ CompactionWorker worker = new CompactionWorker(0, mockQueue);
+ worker.run();
Assert.assertEquals(0, SystemInfo.getInstance().getCompactionMemoryCost().get());
Assert.assertEquals(0, SystemInfo.getInstance().getCompactionFileNumCost().get());
for (TsFileResource tsFileResource : sequenceFiles) {
@@ -185,13 +186,9 @@ public class MemoryControlTest {
}
}
- /**
- * AllowCompaction is false.
- *
- * @throws Exception
- */
+ /** AllowCompaction is false. */
@Test
- public void testFailedToCheckValidInInnerTask() {
+ public void testFailedToCheckValidInInnerTask() throws InterruptedException {
List<TsFileResource> sequenceFiles = new ArrayList<>();
for (int i = 1; i <= 10; i++) {
sequenceFiles.add(
@@ -200,15 +197,16 @@ public class MemoryControlTest {
TsFileResourceStatus.COMPACTION_CANDIDATE));
}
TsFileManager tsFileManager = Mockito.mock(TsFileManager.class);
- Mockito.when(tsFileManager.getStorageGroupName()).thenReturn("root.sg");
- Mockito.when(tsFileManager.getDataRegionId()).thenReturn("1");
+ Mockito.when(tsFileManager.isAllowCompaction()).thenReturn(false);
// fail to check valid when tsfile manager is not allowed to compaction in inner task
InnerSpaceCompactionTask innerTask =
- new InnerSpaceCompactionTask(
- 0L, tsFileManager, sequenceFiles, true, null, new AtomicInteger(0), 0L);
- boolean success = innerTask.checkValidAndSetMerging();
- Assert.assertFalse(success);
+ new InnerSpaceCompactionTask(0L, tsFileManager, sequenceFiles, true, null, 0L);
+ FixedPriorityBlockingQueue<AbstractCompactionTask> mockQueue =
+ Mockito.mock(FixedPriorityBlockingQueue.class);
+ Mockito.when(mockQueue.take()).thenReturn(innerTask).thenThrow(new InterruptedException());
+ CompactionWorker worker = new CompactionWorker(0, mockQueue);
+ worker.run();
Assert.assertEquals(0, SystemInfo.getInstance().getCompactionMemoryCost().get());
Assert.assertEquals(0, SystemInfo.getInstance().getCompactionFileNumCost().get());
for (TsFileResource tsFileResource : sequenceFiles) {
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CrossSpaceCompactionWithUnusualCasesTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CrossSpaceCompactionWithUnusualCasesTest.java
index 6c981ea045f..5728e28b9fe 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CrossSpaceCompactionWithUnusualCasesTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CrossSpaceCompactionWithUnusualCasesTest.java
@@ -52,7 +52,6 @@ import org.junit.Test;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
-import java.util.concurrent.atomic.AtomicInteger;
/**
* If an unsequence file is generated by a load operation or some files in sequence space is
@@ -135,7 +134,6 @@ public class CrossSpaceCompactionWithUnusualCasesTest extends AbstractCompaction
result.get(0).getSeqFiles(),
result.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -208,7 +206,6 @@ public class CrossSpaceCompactionWithUnusualCasesTest extends AbstractCompaction
result.get(0).getSeqFiles(),
result.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -292,7 +289,6 @@ public class CrossSpaceCompactionWithUnusualCasesTest extends AbstractCompaction
result.get(0).getSeqFiles(),
result.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -386,7 +382,6 @@ public class CrossSpaceCompactionWithUnusualCasesTest extends AbstractCompaction
result.get(0).getSeqFiles(),
result.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -482,7 +477,6 @@ public class CrossSpaceCompactionWithUnusualCasesTest extends AbstractCompaction
result.get(0).getSeqFiles(),
result.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -544,7 +538,6 @@ public class CrossSpaceCompactionWithUnusualCasesTest extends AbstractCompaction
result.get(0).getSeqFiles(),
result.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -648,7 +641,6 @@ public class CrossSpaceCompactionWithUnusualCasesTest extends AbstractCompaction
result.get(0).getSeqFiles(),
result.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -891,7 +883,6 @@ public class CrossSpaceCompactionWithUnusualCasesTest extends AbstractCompaction
result.get(0).getSeqFiles(),
result.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/FastAlignedCrossCompactionTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/FastAlignedCrossCompactionTest.java
index c70f03454c5..6cd909c840f 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/FastAlignedCrossCompactionTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/FastAlignedCrossCompactionTest.java
@@ -49,7 +49,6 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.iotdb.db.storageengine.dataregion.compaction.utils.TsFileGeneratorUtils.createChunkWriter;
import static org.apache.iotdb.db.storageengine.dataregion.compaction.utils.TsFileGeneratorUtils.createCompressionType;
@@ -256,7 +255,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -272,7 +270,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -523,7 +520,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -539,7 +535,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -789,7 +784,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -805,7 +799,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -1072,7 +1065,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -1088,7 +1080,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -1454,7 +1445,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -1470,7 +1460,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -1844,7 +1833,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -1860,7 +1848,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -2244,7 +2231,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -2260,7 +2246,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -2628,7 +2613,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -2644,7 +2628,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -3013,7 +2996,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -3029,7 +3011,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -3409,7 +3390,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -3425,7 +3405,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -3859,7 +3838,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -3875,7 +3853,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -4312,7 +4289,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -4328,7 +4304,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -4765,7 +4740,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -4781,7 +4755,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -5270,7 +5243,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -5286,7 +5258,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -5720,7 +5691,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -5736,7 +5706,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -5951,7 +5920,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -5967,7 +5935,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -6244,7 +6211,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -6260,7 +6226,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -6664,7 +6629,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -6680,7 +6644,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -7049,7 +7012,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -7065,7 +7027,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -7316,7 +7277,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -7332,7 +7292,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -7469,7 +7428,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -7485,7 +7443,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -7673,7 +7630,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -7689,7 +7645,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -7914,7 +7869,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -7930,7 +7884,6 @@ public class FastAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/FastCompactionPerformerWithEmptyPageTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/FastCompactionPerformerWithEmptyPageTest.java
index 815105aff1a..356cf458527 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/FastCompactionPerformerWithEmptyPageTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/FastCompactionPerformerWithEmptyPageTest.java
@@ -43,7 +43,6 @@ import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.atomic.AtomicInteger;
public class FastCompactionPerformerWithEmptyPageTest extends AbstractCompactionTest {
@@ -106,7 +105,6 @@ public class FastCompactionPerformerWithEmptyPageTest extends AbstractCompaction
Arrays.asList(seqFile1),
Arrays.asList(unseqFile1),
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
try {
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/FastCompactionPerformerWithInconsistentCompressionTypeAndEncodingTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/FastCompactionPerformerWithInconsistentCompressionTypeAndEncodingTest.java
index ce2d95737d4..00249ca8095 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/FastCompactionPerformerWithInconsistentCompressionTypeAndEncodingTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/FastCompactionPerformerWithInconsistentCompressionTypeAndEncodingTest.java
@@ -54,7 +54,6 @@ import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.atomic.AtomicInteger;
public class FastCompactionPerformerWithInconsistentCompressionTypeAndEncodingTest
extends AbstractCompactionTest {
@@ -113,7 +112,6 @@ public class FastCompactionPerformerWithInconsistentCompressionTypeAndEncodingTe
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
@@ -192,7 +190,6 @@ public class FastCompactionPerformerWithInconsistentCompressionTypeAndEncodingTe
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
@@ -267,7 +264,6 @@ public class FastCompactionPerformerWithInconsistentCompressionTypeAndEncodingTe
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
@@ -319,7 +315,6 @@ public class FastCompactionPerformerWithInconsistentCompressionTypeAndEncodingTe
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
@@ -398,7 +393,6 @@ public class FastCompactionPerformerWithInconsistentCompressionTypeAndEncodingTe
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
@@ -473,7 +467,6 @@ public class FastCompactionPerformerWithInconsistentCompressionTypeAndEncodingTe
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/FastCrossCompactionPerformerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/FastCrossCompactionPerformerTest.java
index 849b31790e3..d67bf6be322 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/FastCrossCompactionPerformerTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/FastCrossCompactionPerformerTest.java
@@ -27,17 +27,20 @@ import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.ICompactionPerformer;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.FastCompactionPerformer;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.AbstractCompactionTask;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.CrossSpaceCompactionTask;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.subtask.FastCompactionTaskSummary;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.CompactionUtils;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.reader.IDataBlockReader;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.reader.SeriesDataBlockReader;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.schedule.CompactionWorker;
import org.apache.iotdb.db.storageengine.dataregion.compaction.utils.CompactionFileGeneratorUtils;
import org.apache.iotdb.db.storageengine.dataregion.read.control.FileReaderManager;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import org.apache.iotdb.db.storageengine.rescon.memory.SystemInfo;
import org.apache.iotdb.db.tools.validate.TsFileValidationTool;
import org.apache.iotdb.db.utils.EnvironmentUtils;
+import org.apache.iotdb.db.utils.datastructure.FixedPriorityBlockingQueue;
import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
import org.apache.iotdb.tsfile.exception.write.WriteProcessException;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
@@ -52,6 +55,7 @@ import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
+import org.mockito.Mockito;
import java.io.File;
import java.io.IOException;
@@ -60,7 +64,6 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.iotdb.commons.conf.IoTDBConstant.PATH_SEPARATOR;
import static org.apache.iotdb.db.utils.EnvironmentUtils.TEST_QUERY_JOB_ID;
@@ -4030,7 +4033,7 @@ public class FastCrossCompactionPerformerTest extends AbstractCompactionTest {
@Test
public void testReleaseFileNumAndMemoryAfterCrossTask()
- throws IOException, MetadataException, WriteProcessException {
+ throws IOException, MetadataException, WriteProcessException, InterruptedException {
int oldMaxCrossCompactionCandidateFileNum =
SystemInfo.getInstance().getTotalFileLimitForCrossTask();
SystemInfo.getInstance().setTotalFileLimitForCrossTask(15);
@@ -4048,13 +4051,15 @@ public class FastCrossCompactionPerformerTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
1000,
0);
Assert.assertTrue(task.setSourceFilesToCompactionCandidate());
- boolean success = task.checkValidAndSetMerging();
- Assert.assertTrue(success);
- Assert.assertTrue(task.start());
+
+ FixedPriorityBlockingQueue<AbstractCompactionTask> mockQueue =
+ Mockito.mock(FixedPriorityBlockingQueue.class);
+ Mockito.when(mockQueue.take()).thenReturn(task).thenThrow(new InterruptedException());
+ CompactionWorker worker = new CompactionWorker(0, mockQueue);
+ worker.run();
Assert.assertEquals(0, SystemInfo.getInstance().getCompactionFileNumCost().get());
Assert.assertEquals(0, SystemInfo.getInstance().getCompactionMemoryCost().get());
} finally {
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/FastInnerCompactionPerformerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/FastInnerCompactionPerformerTest.java
index e264b2cdc72..aafe6b52f56 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/FastInnerCompactionPerformerTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/FastInnerCompactionPerformerTest.java
@@ -55,7 +55,6 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.iotdb.commons.conf.IoTDBConstant.PATH_SEPARATOR;
import static org.junit.Assert.assertEquals;
@@ -121,13 +120,7 @@ public class FastInnerCompactionPerformerTest extends AbstractCompactionTest {
tsFileManager.addAll(unseqResources, false);
InnerSpaceCompactionTask task =
new InnerSpaceCompactionTask(
- 0,
- tsFileManager,
- seqResources,
- true,
- new FastCompactionPerformer(false),
- new AtomicInteger(0),
- 0);
+ 0, tsFileManager, seqResources, true, new FastCompactionPerformer(false), 0);
Assert.assertTrue(task.start());
Assert.assertEquals(0, FileReaderManager.getInstance().getClosedFileReaderMap().size());
Assert.assertEquals(0, FileReaderManager.getInstance().getUnclosedFileReaderMap().size());
@@ -219,13 +212,7 @@ public class FastInnerCompactionPerformerTest extends AbstractCompactionTest {
tsFileManager.addAll(unseqResources, false);
InnerSpaceCompactionTask task =
new InnerSpaceCompactionTask(
- 0,
- tsFileManager,
- seqResources,
- true,
- new FastCompactionPerformer(false),
- new AtomicInteger(0),
- 0);
+ 0, tsFileManager, seqResources, true, new FastCompactionPerformer(false), 0);
Assert.assertTrue(task.start());
Assert.assertEquals(0, FileReaderManager.getInstance().getClosedFileReaderMap().size());
Assert.assertEquals(0, FileReaderManager.getInstance().getUnclosedFileReaderMap().size());
@@ -360,13 +347,7 @@ public class FastInnerCompactionPerformerTest extends AbstractCompactionTest {
tsFileManager.addAll(unseqResources, false);
InnerSpaceCompactionTask task =
new InnerSpaceCompactionTask(
- 0,
- tsFileManager,
- seqResources,
- true,
- new FastCompactionPerformer(false),
- new AtomicInteger(0),
- 0);
+ 0, tsFileManager, seqResources, true, new FastCompactionPerformer(false), 0);
Assert.assertTrue(task.start());
Assert.assertEquals(0, FileReaderManager.getInstance().getClosedFileReaderMap().size());
Assert.assertEquals(0, FileReaderManager.getInstance().getUnclosedFileReaderMap().size());
@@ -474,13 +455,7 @@ public class FastInnerCompactionPerformerTest extends AbstractCompactionTest {
tsFileManager.addAll(unseqResources, false);
InnerSpaceCompactionTask task =
new InnerSpaceCompactionTask(
- 0,
- tsFileManager,
- unseqResources,
- false,
- new FastCompactionPerformer(false),
- new AtomicInteger(0),
- 0);
+ 0, tsFileManager, unseqResources, false, new FastCompactionPerformer(false), 0);
Assert.assertTrue(task.start());
Assert.assertEquals(0, FileReaderManager.getInstance().getClosedFileReaderMap().size());
Assert.assertEquals(0, FileReaderManager.getInstance().getUnclosedFileReaderMap().size());
@@ -589,13 +564,7 @@ public class FastInnerCompactionPerformerTest extends AbstractCompactionTest {
tsFileManager.addAll(unseqResources, false);
InnerSpaceCompactionTask task =
new InnerSpaceCompactionTask(
- 0,
- tsFileManager,
- unseqResources,
- false,
- new FastCompactionPerformer(false),
- new AtomicInteger(0),
- 0);
+ 0, tsFileManager, unseqResources, false, new FastCompactionPerformer(false), 0);
Assert.assertTrue(task.start());
Assert.assertEquals(0, FileReaderManager.getInstance().getClosedFileReaderMap().size());
Assert.assertEquals(0, FileReaderManager.getInstance().getUnclosedFileReaderMap().size());
@@ -735,13 +704,7 @@ public class FastInnerCompactionPerformerTest extends AbstractCompactionTest {
tsFileManager.addAll(unseqResources, false);
InnerSpaceCompactionTask task =
new InnerSpaceCompactionTask(
- 0,
- tsFileManager,
- unseqResources,
- false,
- new FastCompactionPerformer(false),
- new AtomicInteger(0),
- 0);
+ 0, tsFileManager, unseqResources, false, new FastCompactionPerformer(false), 0);
Assert.assertTrue(task.start());
Assert.assertEquals(0, FileReaderManager.getInstance().getClosedFileReaderMap().size());
Assert.assertEquals(0, FileReaderManager.getInstance().getUnclosedFileReaderMap().size());
@@ -870,13 +833,7 @@ public class FastInnerCompactionPerformerTest extends AbstractCompactionTest {
tsFileManager.addAll(unseqResources, false);
InnerSpaceCompactionTask task =
new InnerSpaceCompactionTask(
- 0,
- tsFileManager,
- unseqResources,
- false,
- new FastCompactionPerformer(false),
- new AtomicInteger(0),
- 0);
+ 0, tsFileManager, unseqResources, false, new FastCompactionPerformer(false), 0);
Assert.assertTrue(task.start());
Assert.assertEquals(0, FileReaderManager.getInstance().getClosedFileReaderMap().size());
Assert.assertEquals(0, FileReaderManager.getInstance().getUnclosedFileReaderMap().size());
@@ -991,13 +948,7 @@ public class FastInnerCompactionPerformerTest extends AbstractCompactionTest {
tsFileManager.addAll(unseqResources, false);
InnerSpaceCompactionTask task =
new InnerSpaceCompactionTask(
- 0,
- tsFileManager,
- unseqResources,
- false,
- new FastCompactionPerformer(false),
- new AtomicInteger(0),
- 0);
+ 0, tsFileManager, unseqResources, false, new FastCompactionPerformer(false), 0);
Assert.assertTrue(task.start());
Assert.assertEquals(0, FileReaderManager.getInstance().getClosedFileReaderMap().size());
Assert.assertEquals(0, FileReaderManager.getInstance().getUnclosedFileReaderMap().size());
@@ -1082,13 +1033,7 @@ public class FastInnerCompactionPerformerTest extends AbstractCompactionTest {
tsFileManager.addAll(unseqResources, false);
InnerSpaceCompactionTask task =
new InnerSpaceCompactionTask(
- 0,
- tsFileManager,
- seqResources,
- true,
- new FastCompactionPerformer(false),
- new AtomicInteger(0),
- 0);
+ 0, tsFileManager, seqResources, true, new FastCompactionPerformer(false), 0);
Assert.assertTrue(task.start());
Assert.assertEquals(0, FileReaderManager.getInstance().getClosedFileReaderMap().size());
Assert.assertEquals(0, FileReaderManager.getInstance().getUnclosedFileReaderMap().size());
@@ -1202,13 +1147,7 @@ public class FastInnerCompactionPerformerTest extends AbstractCompactionTest {
tsFileManager.addAll(unseqResources, false);
InnerSpaceCompactionTask task =
new InnerSpaceCompactionTask(
- 0,
- tsFileManager,
- seqResources,
- true,
- new FastCompactionPerformer(false),
- new AtomicInteger(0),
- 0);
+ 0, tsFileManager, seqResources, true, new FastCompactionPerformer(false), 0);
Assert.assertTrue(task.start());
Assert.assertEquals(0, FileReaderManager.getInstance().getClosedFileReaderMap().size());
Assert.assertEquals(0, FileReaderManager.getInstance().getUnclosedFileReaderMap().size());
@@ -1331,13 +1270,7 @@ public class FastInnerCompactionPerformerTest extends AbstractCompactionTest {
tsFileManager.addAll(unseqResources, false);
InnerSpaceCompactionTask task =
new InnerSpaceCompactionTask(
- 0,
- tsFileManager,
- seqResources,
- true,
- new FastCompactionPerformer(false),
- new AtomicInteger(0),
- 0);
+ 0, tsFileManager, seqResources, true, new FastCompactionPerformer(false), 0);
Assert.assertTrue(task.start());
Assert.assertEquals(0, FileReaderManager.getInstance().getClosedFileReaderMap().size());
Assert.assertEquals(0, FileReaderManager.getInstance().getUnclosedFileReaderMap().size());
@@ -1465,13 +1398,7 @@ public class FastInnerCompactionPerformerTest extends AbstractCompactionTest {
tsFileManager.addAll(unseqResources, false);
InnerSpaceCompactionTask task =
new InnerSpaceCompactionTask(
- 0,
- tsFileManager,
- unseqResources,
- false,
- new FastCompactionPerformer(false),
- new AtomicInteger(0),
- 0);
+ 0, tsFileManager, unseqResources, false, new FastCompactionPerformer(false), 0);
Assert.assertTrue(task.start());
Assert.assertEquals(0, FileReaderManager.getInstance().getClosedFileReaderMap().size());
Assert.assertEquals(0, FileReaderManager.getInstance().getUnclosedFileReaderMap().size());
@@ -1649,13 +1576,7 @@ public class FastInnerCompactionPerformerTest extends AbstractCompactionTest {
tsFileManager.addAll(unseqResources, false);
InnerSpaceCompactionTask task =
new InnerSpaceCompactionTask(
- 0,
- tsFileManager,
- unseqResources,
- false,
- new FastCompactionPerformer(false),
- new AtomicInteger(0),
- 0);
+ 0, tsFileManager, unseqResources, false, new FastCompactionPerformer(false), 0);
Assert.assertTrue(task.start());
Assert.assertEquals(0, FileReaderManager.getInstance().getClosedFileReaderMap().size());
Assert.assertEquals(0, FileReaderManager.getInstance().getUnclosedFileReaderMap().size());
@@ -1813,13 +1734,7 @@ public class FastInnerCompactionPerformerTest extends AbstractCompactionTest {
tsFileManager.addAll(unseqResources, false);
InnerSpaceCompactionTask task =
new InnerSpaceCompactionTask(
- 0,
- tsFileManager,
- unseqResources,
- false,
- new FastCompactionPerformer(false),
- new AtomicInteger(0),
- 0);
+ 0, tsFileManager, unseqResources, false, new FastCompactionPerformer(false), 0);
Assert.assertTrue(task.start());
Assert.assertEquals(0, FileReaderManager.getInstance().getClosedFileReaderMap().size());
Assert.assertEquals(0, FileReaderManager.getInstance().getUnclosedFileReaderMap().size());
@@ -1929,13 +1844,7 @@ public class FastInnerCompactionPerformerTest extends AbstractCompactionTest {
tsFileManager.addAll(unseqResources, false);
InnerSpaceCompactionTask task =
new InnerSpaceCompactionTask(
- 0,
- tsFileManager,
- unseqResources,
- false,
- new FastCompactionPerformer(false),
- new AtomicInteger(0),
- 0);
+ 0, tsFileManager, unseqResources, false, new FastCompactionPerformer(false), 0);
Assert.assertTrue(task.start());
Assert.assertEquals(0, FileReaderManager.getInstance().getClosedFileReaderMap().size());
Assert.assertEquals(0, FileReaderManager.getInstance().getUnclosedFileReaderMap().size());
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/FastNonAlignedCrossCompactionTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/FastNonAlignedCrossCompactionTest.java
index 3436efb614f..1f56106664e 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/FastNonAlignedCrossCompactionTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/FastNonAlignedCrossCompactionTest.java
@@ -51,7 +51,6 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.iotdb.db.storageengine.dataregion.compaction.utils.TsFileGeneratorUtils.createChunkWriter;
import static org.apache.iotdb.db.storageengine.dataregion.compaction.utils.TsFileGeneratorUtils.createCompressionType;
@@ -279,7 +278,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -301,7 +299,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -548,7 +545,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -570,7 +566,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -816,7 +811,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -844,7 +838,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -1108,7 +1101,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -1130,7 +1122,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -1493,7 +1484,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -1515,7 +1505,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -1885,7 +1874,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -1907,7 +1895,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -2287,7 +2274,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -2309,7 +2295,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -2674,7 +2659,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -2696,7 +2680,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -3062,7 +3045,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -3084,7 +3066,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -3460,7 +3441,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -3482,7 +3462,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -3913,7 +3892,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -3935,7 +3913,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -4369,7 +4346,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -4391,7 +4367,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -4825,7 +4800,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -4847,7 +4821,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -5333,7 +5306,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -5355,7 +5327,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -5786,7 +5757,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -5808,7 +5778,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -6023,7 +5992,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -6045,7 +6013,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -6322,7 +6289,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -6344,7 +6310,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -6748,7 +6713,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -6770,7 +6734,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -7136,7 +7099,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -7158,7 +7120,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -7409,7 +7370,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -7431,7 +7391,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -7568,7 +7527,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -7590,7 +7548,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -7778,7 +7735,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -7800,7 +7756,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -8018,7 +7973,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -8040,7 +7994,6 @@ public class FastNonAlignedCrossCompactionTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/ReadChunkInnerCompactionTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/ReadChunkInnerCompactionTest.java
index 40e14ea792b..fc9ce3a259c 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/ReadChunkInnerCompactionTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/ReadChunkInnerCompactionTest.java
@@ -47,7 +47,6 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.iotdb.db.storageengine.dataregion.compaction.utils.TsFileGeneratorUtils.createChunkWriter;
import static org.apache.iotdb.db.storageengine.dataregion.compaction.utils.TsFileGeneratorUtils.createCompressionType;
@@ -183,13 +182,8 @@ public class ReadChunkInnerCompactionTest extends AbstractCompactionTest {
readSourceFiles(createTimeseries(maxDeviceNum, maxMeasurementNum, false), tsDataTypes);
InnerSpaceCompactionTask task =
new InnerSpaceCompactionTask(
- 0,
- tsFileManager,
- seqResources,
- true,
- new ReadChunkCompactionPerformer(),
- new AtomicInteger(0),
- 0);
+ 0, tsFileManager, seqResources, true, new ReadChunkCompactionPerformer(), 0);
+
task.start();
validateSeqFiles(true);
@@ -361,13 +355,8 @@ public class ReadChunkInnerCompactionTest extends AbstractCompactionTest {
readSourceFiles(createTimeseries(maxDeviceNum, maxMeasurementNum, true), tsDataTypes);
InnerSpaceCompactionTask task =
new InnerSpaceCompactionTask(
- 0,
- tsFileManager,
- seqResources,
- true,
- new ReadChunkCompactionPerformer(),
- new AtomicInteger(0),
- 0);
+ 0, tsFileManager, seqResources, true, new ReadChunkCompactionPerformer(), 0);
+
task.start();
validateSeqFiles(true);
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/ReadPointAlignedCrossCompactionTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/ReadPointAlignedCrossCompactionTest.java
index 80c2d95f1c9..6b6498f5372 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/ReadPointAlignedCrossCompactionTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/ReadPointAlignedCrossCompactionTest.java
@@ -49,7 +49,6 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.iotdb.db.storageengine.dataregion.compaction.utils.TsFileGeneratorUtils.createChunkWriter;
import static org.apache.iotdb.db.storageengine.dataregion.compaction.utils.TsFileGeneratorUtils.createCompressionType;
@@ -259,7 +258,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -275,7 +273,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -526,7 +523,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -542,7 +538,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -792,7 +787,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -808,7 +802,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -1075,7 +1068,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -1091,7 +1083,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -1457,7 +1448,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -1473,7 +1463,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -1847,7 +1836,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -1863,7 +1851,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -2247,7 +2234,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -2263,7 +2249,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -2631,7 +2616,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -2647,7 +2631,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -3016,7 +2999,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -3032,7 +3014,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -3412,7 +3393,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -3428,7 +3408,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -3862,7 +3841,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -3878,7 +3856,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -4315,7 +4292,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -4331,7 +4307,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -4768,7 +4743,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -4784,7 +4758,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -5273,7 +5246,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -5289,7 +5261,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -5723,7 +5694,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -5739,7 +5709,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -5954,7 +5923,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -5970,7 +5938,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -6247,7 +6214,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -6263,7 +6229,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -6667,7 +6632,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -6683,7 +6647,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -7052,7 +7015,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -7068,7 +7030,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -7319,7 +7280,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -7335,7 +7295,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -7472,7 +7431,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -7488,7 +7446,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -7676,7 +7633,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -7692,7 +7648,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -7917,7 +7872,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -7933,7 +7887,6 @@ public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/ReadPointNonAlignedCrossCompactionTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/ReadPointNonAlignedCrossCompactionTest.java
index 14fb6ca01ea..0db27da3866 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/ReadPointNonAlignedCrossCompactionTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/ReadPointNonAlignedCrossCompactionTest.java
@@ -50,7 +50,6 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.iotdb.db.storageengine.dataregion.compaction.utils.TsFileGeneratorUtils.createChunkWriter;
import static org.apache.iotdb.db.storageengine.dataregion.compaction.utils.TsFileGeneratorUtils.createCompressionType;
@@ -281,7 +280,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -297,7 +295,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -544,7 +541,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -560,7 +556,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -806,7 +801,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -822,7 +816,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -1086,7 +1079,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -1102,7 +1094,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -1465,7 +1456,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -1481,7 +1471,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -1851,7 +1840,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -1867,7 +1855,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -2247,7 +2234,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -2263,7 +2249,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -2628,7 +2613,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -2644,7 +2628,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -3010,7 +2993,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -3026,7 +3008,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -3402,7 +3383,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -3418,7 +3398,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -3849,7 +3828,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -3865,7 +3843,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -4299,7 +4276,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -4315,7 +4291,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -4749,7 +4724,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -4765,7 +4739,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -5251,7 +5224,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -5267,7 +5239,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -5698,7 +5669,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -5714,7 +5684,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -5929,7 +5898,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -5945,7 +5913,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -6222,7 +6189,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -6238,7 +6204,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -6642,7 +6607,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -6658,7 +6622,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -7024,7 +6987,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -7040,7 +7002,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -7291,7 +7252,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -7307,7 +7267,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -7444,7 +7403,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -7460,7 +7418,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -7648,7 +7605,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -7664,7 +7620,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
@@ -7882,7 +7837,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
@@ -7898,7 +7852,6 @@ public class ReadPointNonAlignedCrossCompactionTest extends AbstractCompactionTe
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
Assert.assertTrue(innerTask.start());
validateSeqFiles(true);
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/CrossSpaceCompactionSelectorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/CrossSpaceCompactionSelectorTest.java
index 62f3a4b0ae3..f72e1b5eb35 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/CrossSpaceCompactionSelectorTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/CrossSpaceCompactionSelectorTest.java
@@ -22,8 +22,10 @@ import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.storageengine.dataregion.compaction.AbstractCompactionTest;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.exception.FileCannotTransitToCompactingException;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.AbstractCompactionTask;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.CrossSpaceCompactionTask;
-import org.apache.iotdb.db.storageengine.dataregion.compaction.schedule.CompactionTaskManager;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.schedule.CompactionWorker;
import org.apache.iotdb.db.storageengine.dataregion.compaction.selector.impl.RewriteCrossSpaceCompactionSelector;
import org.apache.iotdb.db.storageengine.dataregion.compaction.selector.utils.CrossCompactionTaskResource;
import org.apache.iotdb.db.storageengine.dataregion.compaction.selector.utils.CrossSpaceCompactionCandidate;
@@ -31,12 +33,14 @@ import org.apache.iotdb.db.storageengine.dataregion.read.control.FileReaderManag
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
import org.apache.iotdb.db.storageengine.rescon.memory.SystemInfo;
+import org.apache.iotdb.db.utils.datastructure.FixedPriorityBlockingQueue;
import org.apache.iotdb.tsfile.exception.write.WriteProcessException;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
+import org.mockito.Mockito;
import java.io.IOException;
import java.util.List;
@@ -177,7 +181,7 @@ public class CrossSpaceCompactionSelectorTest extends AbstractCompactionTest {
@Test
public void testSelectWithTooManySourceFiles()
- throws IOException, MetadataException, WriteProcessException {
+ throws IOException, MetadataException, WriteProcessException, InterruptedException {
int oldMaxFileNumForCompaction = SystemInfo.getInstance().getTotalFileLimitForCrossTask();
SystemInfo.getInstance().setTotalFileLimitForCrossTask(1);
SystemInfo.getInstance().getCompactionFileNumCost().set(0);
@@ -203,13 +207,19 @@ public class CrossSpaceCompactionSelectorTest extends AbstractCompactionTest {
.getConfig()
.getCrossCompactionPerformer()
.createInstance(),
- CompactionTaskManager.currentTaskNum,
1000,
tsFileManager.getNextCompactionTaskId());
// set file status to COMPACTION_CANDIDATE
Assert.assertTrue(crossSpaceCompactionTask.setSourceFilesToCompactionCandidate());
- Assert.assertFalse(crossSpaceCompactionTask.checkValidAndSetMerging());
+ FixedPriorityBlockingQueue<AbstractCompactionTask> mockQueue =
+ Mockito.mock(FixedPriorityBlockingQueue.class);
+ Mockito.when(mockQueue.take())
+ .thenReturn(crossSpaceCompactionTask)
+ .thenThrow(new InterruptedException());
+ CompactionWorker worker = new CompactionWorker(0, mockQueue);
+ worker.run();
+
Assert.assertEquals(0, SystemInfo.getInstance().getCompactionMemoryCost().get());
Assert.assertEquals(0, SystemInfo.getInstance().getCompactionFileNumCost().get());
for (TsFileResource resource : seqResources) {
@@ -341,7 +351,6 @@ public class CrossSpaceCompactionSelectorTest extends AbstractCompactionTest {
.getConfig()
.getCrossCompactionPerformer()
.createInstance(),
- CompactionTaskManager.currentTaskNum,
crossCompactionTaskResource.getTotalMemoryCost(),
tsFileManager.getNextCompactionTaskId());
// set file status to COMPACTION_CANDIDATE
@@ -455,7 +464,6 @@ public class CrossSpaceCompactionSelectorTest extends AbstractCompactionTest {
.getConfig()
.getCrossCompactionPerformer()
.createInstance(),
- CompactionTaskManager.currentTaskNum,
crossCompactionTaskResource.getTotalMemoryCost(),
tsFileManager.getNextCompactionTaskId());
// set file status to COMPACTION_CANDIDATE
@@ -556,7 +564,6 @@ public class CrossSpaceCompactionSelectorTest extends AbstractCompactionTest {
.getConfig()
.getCrossCompactionPerformer()
.createInstance(),
- CompactionTaskManager.currentTaskNum,
crossCompactionTaskResource.getTotalMemoryCost(),
tsFileManager.getNextCompactionTaskId());
// set file status to COMPACTION_CANDIDATE and add into queue
@@ -569,9 +576,20 @@ public class CrossSpaceCompactionSelectorTest extends AbstractCompactionTest {
cd1.countDown();
cd2.await();
- if (crossSpaceCompactionTask.checkValidAndSetMerging()) {
- throw new RuntimeException("cross space compaction task should be invalid.");
- }
+ boolean exceptionCaught = false;
+
+ Assert.assertThrows(
+ "cross space compaction task should be invalid.",
+ FileCannotTransitToCompactingException.class,
+ crossSpaceCompactionTask::transitSourceFilesToMerging);
+
+ FixedPriorityBlockingQueue<AbstractCompactionTask> mockQueue =
+ Mockito.mock(FixedPriorityBlockingQueue.class);
+ Mockito.when(mockQueue.take())
+ .thenReturn(crossSpaceCompactionTask)
+ .thenThrow(new InterruptedException());
+ CompactionWorker worker = new CompactionWorker(0, mockQueue);
+ worker.run();
for (int i = 0; i < seqResources.size(); i++) {
TsFileResource resource = seqResources.get(i);
@@ -675,7 +693,6 @@ public class CrossSpaceCompactionSelectorTest extends AbstractCompactionTest {
.getConfig()
.getCrossCompactionPerformer()
.createInstance(),
- CompactionTaskManager.currentTaskNum,
crossCompactionTaskResource.getTotalMemoryCost(),
tsFileManager.getNextCompactionTaskId());
// set file status to COMPACTION_CANDIDATE
@@ -793,7 +810,6 @@ public class CrossSpaceCompactionSelectorTest extends AbstractCompactionTest {
.getConfig()
.getCrossCompactionPerformer()
.createInstance(),
- CompactionTaskManager.currentTaskNum,
crossCompactionTaskResource.getTotalMemoryCost(),
tsFileManager.getNextCompactionTaskId());
// set file status to COMPACTION_CANDIDATE
@@ -901,7 +917,6 @@ public class CrossSpaceCompactionSelectorTest extends AbstractCompactionTest {
.getConfig()
.getCrossCompactionPerformer()
.createInstance(),
- CompactionTaskManager.currentTaskNum,
crossCompactionTaskResource.getTotalMemoryCost(),
tsFileManager.getNextCompactionTaskId());
// set file status to COMPACTION_CANDIDATE and add into queue
@@ -914,9 +929,18 @@ public class CrossSpaceCompactionSelectorTest extends AbstractCompactionTest {
cd1.countDown();
cd2.await();
- if (crossSpaceCompactionTask.checkValidAndSetMerging()) {
- throw new RuntimeException("cross space compaction task should be invalid.");
- }
+ Assert.assertThrows(
+ "cross space compaction task should be invalid.",
+ FileCannotTransitToCompactingException.class,
+ crossSpaceCompactionTask::transitSourceFilesToMerging);
+
+ FixedPriorityBlockingQueue<AbstractCompactionTask> mockQueue =
+ Mockito.mock(FixedPriorityBlockingQueue.class);
+ Mockito.when(mockQueue.take())
+ .thenReturn(crossSpaceCompactionTask)
+ .thenThrow(new InterruptedException());
+ CompactionWorker worker = new CompactionWorker(0, mockQueue);
+ worker.run();
for (int i = 0; i < seqResources.size(); i++) {
TsFileResource resource = seqResources.get(i);
@@ -1161,7 +1185,6 @@ public class CrossSpaceCompactionSelectorTest extends AbstractCompactionTest {
.getConfig()
.getCrossCompactionPerformer()
.createInstance(),
- CompactionTaskManager.currentTaskNum,
crossCompactionTaskResource.getTotalMemoryCost(),
tsFileManager.getNextCompactionTaskId());
// set file status to COMPACTION_CANDIDATE
@@ -1275,7 +1298,6 @@ public class CrossSpaceCompactionSelectorTest extends AbstractCompactionTest {
.getConfig()
.getCrossCompactionPerformer()
.createInstance(),
- CompactionTaskManager.currentTaskNum,
crossCompactionTaskResource.getTotalMemoryCost(),
tsFileManager.getNextCompactionTaskId());
// set file status to COMPACTION_CANDIDATE
@@ -1380,7 +1402,6 @@ public class CrossSpaceCompactionSelectorTest extends AbstractCompactionTest {
.getConfig()
.getCrossCompactionPerformer()
.createInstance(),
- CompactionTaskManager.currentTaskNum,
crossCompactionTaskResource.getTotalMemoryCost(),
tsFileManager.getNextCompactionTaskId());
// set file status to COMPACTION_CANDIDATE and add into queue
@@ -1393,9 +1414,18 @@ public class CrossSpaceCompactionSelectorTest extends AbstractCompactionTest {
cd1.countDown();
cd2.await();
- if (crossSpaceCompactionTask.checkValidAndSetMerging()) {
- throw new RuntimeException("cross space compaction task should be invalid.");
- }
+ Assert.assertThrows(
+ "cross space compaction task should be invalid.",
+ FileCannotTransitToCompactingException.class,
+ crossSpaceCompactionTask::transitSourceFilesToMerging);
+
+ FixedPriorityBlockingQueue<AbstractCompactionTask> mockQueue =
+ Mockito.mock(FixedPriorityBlockingQueue.class);
+ Mockito.when(mockQueue.take())
+ .thenReturn(crossSpaceCompactionTask)
+ .thenThrow(new InterruptedException());
+ CompactionWorker worker = new CompactionWorker(0, mockQueue);
+ worker.run();
for (int i = 0; i < unseqResources.size(); i++) {
TsFileResource resource = unseqResources.get(i);
@@ -1498,7 +1528,6 @@ public class CrossSpaceCompactionSelectorTest extends AbstractCompactionTest {
.getConfig()
.getCrossCompactionPerformer()
.createInstance(),
- CompactionTaskManager.currentTaskNum,
crossCompactionTaskResource.getTotalMemoryCost(),
tsFileManager.getNextCompactionTaskId());
// set file status to COMPACTION_CANDIDATE
@@ -1616,7 +1645,6 @@ public class CrossSpaceCompactionSelectorTest extends AbstractCompactionTest {
.getConfig()
.getCrossCompactionPerformer()
.createInstance(),
- CompactionTaskManager.currentTaskNum,
crossCompactionTaskResource.getTotalMemoryCost(),
tsFileManager.getNextCompactionTaskId());
// set file status to COMPACTION_CANDIDATE
@@ -1724,7 +1752,6 @@ public class CrossSpaceCompactionSelectorTest extends AbstractCompactionTest {
.getConfig()
.getCrossCompactionPerformer()
.createInstance(),
- CompactionTaskManager.currentTaskNum,
crossCompactionTaskResource.getTotalMemoryCost(),
tsFileManager.getNextCompactionTaskId());
// set file status to COMPACTION_CANDIDATE and add into queue
@@ -1737,9 +1764,18 @@ public class CrossSpaceCompactionSelectorTest extends AbstractCompactionTest {
cd1.countDown();
cd2.await();
- if (crossSpaceCompactionTask.checkValidAndSetMerging()) {
- throw new RuntimeException("cross space compaction task should be invalid.");
- }
+ Assert.assertThrows(
+ "cross space compaction task should be invalid.",
+ FileCannotTransitToCompactingException.class,
+ crossSpaceCompactionTask::transitSourceFilesToMerging);
+
+ FixedPriorityBlockingQueue<AbstractCompactionTask> mockQueue =
+ Mockito.mock(FixedPriorityBlockingQueue.class);
+ Mockito.when(mockQueue.take())
+ .thenReturn(crossSpaceCompactionTask)
+ .thenThrow(new InterruptedException());
+ CompactionWorker worker = new CompactionWorker(0, mockQueue);
+ worker.run();
for (int i = 0; i < unseqResources.size(); i++) {
TsFileResource resource = unseqResources.get(i);
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/CrossSpaceCompactionWithFastPerformerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/CrossSpaceCompactionWithFastPerformerTest.java
index a3231e65607..90116dced7a 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/CrossSpaceCompactionWithFastPerformerTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/CrossSpaceCompactionWithFastPerformerTest.java
@@ -60,7 +60,6 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import java.util.concurrent.atomic.AtomicInteger;
import static org.junit.Assert.fail;
@@ -427,7 +426,6 @@ public class CrossSpaceCompactionWithFastPerformerTest {
mergeResource.getSeqFiles(),
mergeResource.getUnseqFiles(),
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(compactionTask.start());
@@ -731,7 +729,6 @@ public class CrossSpaceCompactionWithFastPerformerTest {
mergeResource.getSeqFiles(),
mergeResource.getUnseqFiles(),
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(compactionTask.start());
@@ -1034,7 +1031,6 @@ public class CrossSpaceCompactionWithFastPerformerTest {
mergeResource.getSeqFiles(),
mergeResource.getUnseqFiles(),
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(compactionTask.start());
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/CrossSpaceCompactionWithFastPerformerValidationTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/CrossSpaceCompactionWithFastPerformerValidationTest.java
index e7010e8787d..8da3fd685e0 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/CrossSpaceCompactionWithFastPerformerValidationTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/CrossSpaceCompactionWithFastPerformerValidationTest.java
@@ -65,7 +65,6 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.iotdb.commons.conf.IoTDBConstant.PATH_SEPARATOR;
import static org.apache.iotdb.tsfile.utils.TsFileGeneratorUtils.alignDeviceOffset;
@@ -129,7 +128,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -174,7 +172,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -219,7 +216,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -272,7 +268,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -322,7 +317,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -370,7 +364,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -421,7 +414,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -471,7 +463,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -521,7 +512,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -572,7 +562,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -623,7 +612,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -674,7 +662,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -726,7 +713,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -779,7 +765,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -832,7 +817,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -886,7 +870,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -941,7 +924,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -996,7 +978,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1051,7 +1032,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1106,7 +1086,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1160,7 +1139,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1210,7 +1188,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1261,7 +1238,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1312,7 +1288,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1363,7 +1338,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1415,7 +1389,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1468,7 +1441,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1521,7 +1493,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1575,7 +1546,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1630,7 +1600,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1685,7 +1654,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1740,7 +1708,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1795,7 +1762,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1849,7 +1815,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1901,7 +1866,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1949,7 +1913,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -2002,7 +1965,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -2056,7 +2018,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -2110,7 +2071,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -2211,8 +2171,7 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
FastCompactionPerformer performer = new FastCompactionPerformer(false);
performer.setSourceFiles(sourceFiles);
InnerSpaceCompactionTask innerSpaceCompactionTask =
- new InnerSpaceCompactionTask(
- 0, tsFileManager, sourceFiles, true, performer, new AtomicInteger(0), 0);
+ new InnerSpaceCompactionTask(0, tsFileManager, sourceFiles, true, performer, 0);
Assert.assertTrue(
CompactionUtils.validateTsFileResources(tsFileManager, COMPACTION_TEST_SG, 0));
innerSpaceCompactionTask.start();
@@ -2376,7 +2335,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
sourceFiles.getSeqFiles(),
sourceFiles.getUnseqFiles(),
new FastCompactionPerformer(true),
- new AtomicInteger(0),
sourceFiles.getTotalMemoryCost(),
0)
.start());
@@ -2448,7 +2406,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
sourceFiles.getSeqFiles(),
sourceFiles.getUnseqFiles(),
new FastCompactionPerformer(true),
- new AtomicInteger(0),
sourceFiles.getTotalMemoryCost(),
0)
.start();
@@ -2521,7 +2478,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
sourceFiles.getSeqFiles(),
sourceFiles.getUnseqFiles(),
new FastCompactionPerformer(true),
- new AtomicInteger(0),
sourceFiles.getTotalMemoryCost(),
0)
.start();
@@ -2596,7 +2552,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
sourceFiles.getSeqFiles(),
sourceFiles.getUnseqFiles(),
new FastCompactionPerformer(true),
- new AtomicInteger(0),
sourceFiles.getTotalMemoryCost(),
0)
.start());
@@ -2676,7 +2631,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
sourceFiles.getSeqFiles(),
sourceFiles.getUnseqFiles(),
new FastCompactionPerformer(true),
- new AtomicInteger(0),
sourceFiles.getTotalMemoryCost(),
0)
.start();
@@ -2755,7 +2709,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
sourceFiles.getSeqFiles(),
sourceFiles.getUnseqFiles(),
new FastCompactionPerformer(true),
- new AtomicInteger(0),
sourceFiles.getTotalMemoryCost(),
0)
.start();
@@ -2835,7 +2788,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
sourceFiles.getSeqFiles(),
sourceFiles.getUnseqFiles(),
new FastCompactionPerformer(true),
- new AtomicInteger(0),
sourceFiles.getTotalMemoryCost(),
0)
.start();
@@ -2916,7 +2868,6 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
sourceFiles.getSeqFiles(),
sourceFiles.getUnseqFiles(),
new FastCompactionPerformer(true),
- new AtomicInteger(0),
sourceFiles.getTotalMemoryCost(),
0)
.start();
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/CrossSpaceCompactionWithReadPointPerformerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/CrossSpaceCompactionWithReadPointPerformerTest.java
index a6615a8f2ba..59a273d3f4d 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/CrossSpaceCompactionWithReadPointPerformerTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/CrossSpaceCompactionWithReadPointPerformerTest.java
@@ -59,7 +59,6 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import java.util.concurrent.atomic.AtomicInteger;
import static org.junit.Assert.fail;
@@ -426,7 +425,6 @@ public class CrossSpaceCompactionWithReadPointPerformerTest {
mergeResource.getSeqFiles(),
mergeResource.getUnseqFiles(),
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
compactionTask.start();
@@ -730,7 +728,6 @@ public class CrossSpaceCompactionWithReadPointPerformerTest {
mergeResource.getSeqFiles(),
mergeResource.getUnseqFiles(),
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
compactionTask.start();
@@ -1033,7 +1030,6 @@ public class CrossSpaceCompactionWithReadPointPerformerTest {
mergeResource.getSeqFiles(),
mergeResource.getUnseqFiles(),
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
compactionTask.start();
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/CrossSpaceCompactionWithReadPointPerformerValidationTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/CrossSpaceCompactionWithReadPointPerformerValidationTest.java
index 7176e4771a6..67dfb9294cc 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/CrossSpaceCompactionWithReadPointPerformerValidationTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/CrossSpaceCompactionWithReadPointPerformerValidationTest.java
@@ -64,7 +64,6 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.iotdb.commons.conf.IoTDBConstant.PATH_SEPARATOR;
import static org.apache.iotdb.tsfile.utils.TsFileGeneratorUtils.alignDeviceOffset;
@@ -129,7 +128,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -174,7 +172,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -219,7 +216,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -272,7 +268,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -322,7 +317,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -370,7 +364,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -421,7 +414,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -471,7 +463,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -521,7 +512,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -572,7 +562,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -623,7 +612,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -674,7 +662,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -726,7 +713,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -779,7 +765,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -832,7 +817,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -886,7 +870,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -941,7 +924,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -996,7 +978,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1051,7 +1032,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1106,7 +1086,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1160,7 +1139,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1210,7 +1188,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1261,7 +1238,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1312,7 +1288,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1363,7 +1338,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1415,7 +1389,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1468,7 +1441,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1521,7 +1493,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1575,7 +1546,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1630,7 +1600,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1685,7 +1654,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1740,7 +1708,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1795,7 +1762,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1849,7 +1815,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1901,7 +1866,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -1949,7 +1913,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -2002,7 +1965,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -2056,7 +2018,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -2110,7 +2071,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
selected.get(0).getSeqFiles(),
selected.get(0).getUnseqFiles(),
performer,
- new AtomicInteger(0),
0,
tsFileManager.getNextCompactionTaskId())
.doCompaction();
@@ -2228,7 +2188,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
sourceFiles.getSeqFiles(),
sourceFiles.getUnseqFiles(),
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
sourceFiles.getTotalMemoryCost(),
0)
.start());
@@ -2301,7 +2260,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
sourceFiles.getSeqFiles(),
sourceFiles.getUnseqFiles(),
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
sourceFiles.getTotalMemoryCost(),
0)
.start());
@@ -2375,7 +2333,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
sourceFiles.getSeqFiles(),
sourceFiles.getUnseqFiles(),
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
sourceFiles.getTotalMemoryCost(),
0)
.start());
@@ -2450,7 +2407,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
sourceFiles.getSeqFiles(),
sourceFiles.getUnseqFiles(),
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
sourceFiles.getTotalMemoryCost(),
0)
.start());
@@ -2531,7 +2487,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
sourceFiles.getSeqFiles(),
sourceFiles.getUnseqFiles(),
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
sourceFiles.getTotalMemoryCost(),
0)
.start());
@@ -2611,7 +2566,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
sourceFiles.getSeqFiles(),
sourceFiles.getUnseqFiles(),
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
sourceFiles.getTotalMemoryCost(),
0)
.start());
@@ -2692,7 +2646,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
sourceFiles.getSeqFiles(),
sourceFiles.getUnseqFiles(),
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
sourceFiles.getTotalMemoryCost(),
0)
.start());
@@ -2774,7 +2727,6 @@ public class CrossSpaceCompactionWithReadPointPerformerValidationTest
sourceFiles.getSeqFiles(),
sourceFiles.getUnseqFiles(),
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
sourceFiles.getTotalMemoryCost(),
0)
.start());
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/RewriteCrossSpaceCompactionWithFastPerformerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/RewriteCrossSpaceCompactionWithFastPerformerTest.java
index 6cb1bf85250..5d9f43868ce 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/RewriteCrossSpaceCompactionWithFastPerformerTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/RewriteCrossSpaceCompactionWithFastPerformerTest.java
@@ -37,6 +37,7 @@ import org.apache.iotdb.db.storageengine.dataregion.flush.TsFileFlushPolicy;
import org.apache.iotdb.db.storageengine.dataregion.read.control.FileReaderManager;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileManager;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.generator.TsFileNameGenerator;
import org.apache.iotdb.db.storageengine.dataregion.wal.recover.WALRecoverManager;
import org.apache.iotdb.db.utils.EnvironmentUtils;
@@ -62,7 +63,6 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.iotdb.commons.conf.IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX;
import static org.apache.iotdb.commons.conf.IoTDBConstant.PATH_SEPARATOR;
@@ -226,7 +226,6 @@ public class RewriteCrossSpaceCompactionWithFastPerformerTest extends AbstractCo
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
task.start();
@@ -454,7 +453,6 @@ public class RewriteCrossSpaceCompactionWithFastPerformerTest extends AbstractCo
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
task.start();
@@ -596,11 +594,11 @@ public class RewriteCrossSpaceCompactionWithFastPerformerTest extends AbstractCo
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
task.setSourceFilesToCompactionCandidate();
- task.checkValidAndSetMerging();
+ seqResources.forEach(f -> f.setStatus(TsFileResourceStatus.COMPACTING));
+ unseqResources.forEach(f -> f.setStatus(TsFileResourceStatus.COMPACTING));
// delete data in source file during compaction
vsgp.deleteByDevice(
new PartialPath(
@@ -715,11 +713,11 @@ public class RewriteCrossSpaceCompactionWithFastPerformerTest extends AbstractCo
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
task.setSourceFilesToCompactionCandidate();
- task.checkValidAndSetMerging();
+ seqResources.forEach(f -> f.setStatus(TsFileResourceStatus.COMPACTING));
+ unseqResources.forEach(f -> f.setStatus(TsFileResourceStatus.COMPACTING));
// delete data in source file during compaction
vsgp.deleteByDevice(
new PartialPath(
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/RewriteCrossSpaceCompactionWithReadPointPerformerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/RewriteCrossSpaceCompactionWithReadPointPerformerTest.java
index b066cdbc935..24bf2305eff 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/RewriteCrossSpaceCompactionWithReadPointPerformerTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/RewriteCrossSpaceCompactionWithReadPointPerformerTest.java
@@ -37,6 +37,7 @@ import org.apache.iotdb.db.storageengine.dataregion.flush.TsFileFlushPolicy;
import org.apache.iotdb.db.storageengine.dataregion.read.control.FileReaderManager;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileManager;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.generator.TsFileNameGenerator;
import org.apache.iotdb.db.storageengine.dataregion.wal.recover.WALRecoverManager;
import org.apache.iotdb.db.utils.EnvironmentUtils;
@@ -62,7 +63,6 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.iotdb.commons.conf.IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX;
import static org.apache.iotdb.commons.conf.IoTDBConstant.PATH_SEPARATOR;
@@ -226,7 +226,6 @@ public class RewriteCrossSpaceCompactionWithReadPointPerformerTest extends Abstr
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
task.start();
@@ -454,7 +453,6 @@ public class RewriteCrossSpaceCompactionWithReadPointPerformerTest extends Abstr
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
task.start();
@@ -596,11 +594,11 @@ public class RewriteCrossSpaceCompactionWithReadPointPerformerTest extends Abstr
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
task.setSourceFilesToCompactionCandidate();
- task.checkValidAndSetMerging();
+ seqResources.forEach(f -> f.setStatus(TsFileResourceStatus.COMPACTING));
+ unseqResources.forEach(f -> f.setStatus(TsFileResourceStatus.COMPACTING));
// delete data in source file during compaction
vsgp.deleteByDevice(
new PartialPath(
@@ -715,11 +713,11 @@ public class RewriteCrossSpaceCompactionWithReadPointPerformerTest extends Abstr
seqResources,
unseqResources,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0,
0);
task.setSourceFilesToCompactionCandidate();
- task.checkValidAndSetMerging();
+ seqResources.forEach(f -> f.setStatus(TsFileResourceStatus.COMPACTING));
+ unseqResources.forEach(f -> f.setStatus(TsFileResourceStatus.COMPACTING));
// delete data in source file during compaction
vsgp.deleteByDevice(
new PartialPath(
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/inner/InnerCompactionEmptyTsFileTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/inner/InnerCompactionEmptyTsFileTest.java
index 647d6edc363..86dbeddf3a5 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/inner/InnerCompactionEmptyTsFileTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/inner/InnerCompactionEmptyTsFileTest.java
@@ -37,7 +37,6 @@ import org.junit.Test;
import java.io.File;
import java.io.IOException;
import java.util.concurrent.Future;
-import java.util.concurrent.atomic.AtomicInteger;
public class InnerCompactionEmptyTsFileTest extends InnerCompactionTest {
@@ -84,7 +83,6 @@ public class InnerCompactionEmptyTsFileTest extends InnerCompactionTest {
unseqResources.subList(0, 3),
false,
new ReadPointCompactionPerformer(),
- new AtomicInteger(0),
0);
unseqResources.get(0).readLock();
CompactionTaskManager.getInstance().addTaskToWaitingQueue(task);
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/inner/InnerSeqCompactionWithFastPerformerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/inner/InnerSeqCompactionWithFastPerformerTest.java
index dec41bd7848..e8c01e2b0e1 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/inner/InnerSeqCompactionWithFastPerformerTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/inner/InnerSeqCompactionWithFastPerformerTest.java
@@ -39,6 +39,7 @@ import org.apache.iotdb.db.storageengine.dataregion.compaction.utils.CompactionF
import org.apache.iotdb.db.storageengine.dataregion.compaction.utils.CompactionTimeseriesType;
import org.apache.iotdb.db.storageengine.dataregion.flush.TsFileFlushPolicy;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.generator.TsFileNameGenerator;
import org.apache.iotdb.db.utils.EnvironmentUtils;
import org.apache.iotdb.db.utils.constant.TestConstant;
@@ -61,7 +62,6 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;
-import java.util.concurrent.atomic.AtomicInteger;
public class InnerSeqCompactionWithFastPerformerTest {
static final String COMPACTION_TEST_SG = "root.compactionTest";
@@ -1134,15 +1134,11 @@ public class InnerSeqCompactionWithFastPerformerTest {
ICompactionPerformer performer = new FastCompactionPerformer(false);
InnerSpaceCompactionTask task =
new InnerSpaceCompactionTask(
- 0,
- vsgp.getTsFileResourceManager(),
- sourceResources,
- true,
- performer,
- new AtomicInteger(0),
- 0);
+ 0, vsgp.getTsFileResourceManager(), sourceResources, true, performer, 0);
+
task.setSourceFilesToCompactionCandidate();
- task.checkValidAndSetMerging();
+ // set the source files to COMPACTING manually to simulate the concurrent scenario
+ sourceResources.forEach(f -> f.setStatus(TsFileResourceStatus.COMPACTING));
// delete data during compaction
vsgp.deleteByDevice(new PartialPath(fullPaths[0]), 0, 1200, 0);
vsgp.deleteByDevice(new PartialPath(fullPaths[0]), 0, 1800, 0);
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/inner/InnerSeqCompactionWithReadChunkPerformerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/inner/InnerSeqCompactionWithReadChunkPerformerTest.java
index d56b0946808..83b5933c60e 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/inner/InnerSeqCompactionWithReadChunkPerformerTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/inner/InnerSeqCompactionWithReadChunkPerformerTest.java
@@ -39,6 +39,7 @@ import org.apache.iotdb.db.storageengine.dataregion.compaction.utils.CompactionF
import org.apache.iotdb.db.storageengine.dataregion.compaction.utils.CompactionTimeseriesType;
import org.apache.iotdb.db.storageengine.dataregion.flush.TsFileFlushPolicy;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.generator.TsFileNameGenerator;
import org.apache.iotdb.db.utils.EnvironmentUtils;
import org.apache.iotdb.db.utils.constant.TestConstant;
@@ -60,7 +61,6 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;
-import java.util.concurrent.atomic.AtomicInteger;
public class InnerSeqCompactionWithReadChunkPerformerTest {
static final String COMPACTION_TEST_SG = "root.compactionTest";
@@ -1060,7 +1060,7 @@ public class InnerSeqCompactionWithReadChunkPerformerTest {
@Test
public void testCompactionWithDeletionsDuringCompactions()
- throws MetadataException, IOException, DataRegionException {
+ throws MetadataException, IOException, DataRegionException, InterruptedException {
// create source seq files
List<TsFileResource> sourceResources = new ArrayList<>();
List<List<Long>> chunkPagePointsNum = new ArrayList<>();
@@ -1101,10 +1101,9 @@ public class InnerSeqCompactionWithReadChunkPerformerTest {
sourceResources,
true,
new ReadChunkCompactionPerformer(),
- new AtomicInteger(0),
0);
task.setSourceFilesToCompactionCandidate();
- task.checkValidAndSetMerging();
+ sourceResources.forEach(f -> f.setStatus(TsFileResourceStatus.COMPACTING));
// delete data during compaction
vsgp.deleteByDevice(new PartialPath(fullPaths[0]), 0, 1200, 0);
vsgp.deleteByDevice(new PartialPath(fullPaths[0]), 0, 1800, 0);
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/inner/InnerSpaceCompactionSelectorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/inner/InnerSpaceCompactionSelectorTest.java
index 0f286a3d4ec..27d390df298 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/inner/InnerSpaceCompactionSelectorTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/inner/InnerSpaceCompactionSelectorTest.java
@@ -23,21 +23,25 @@ import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.storageengine.dataregion.compaction.AbstractCompactionTest;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.exception.FileCannotTransitToCompactingException;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.FastCompactionPerformer;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.AbstractCompactionTask;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.InnerSpaceCompactionTask;
-import org.apache.iotdb.db.storageengine.dataregion.compaction.schedule.CompactionTaskManager;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.schedule.CompactionWorker;
import org.apache.iotdb.db.storageengine.dataregion.compaction.selector.impl.SizeTieredCompactionSelector;
import org.apache.iotdb.db.storageengine.dataregion.modification.Deletion;
import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile;
import org.apache.iotdb.db.storageengine.dataregion.read.control.FileReaderManager;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
+import org.apache.iotdb.db.utils.datastructure.FixedPriorityBlockingQueue;
import org.apache.iotdb.tsfile.exception.write.WriteProcessException;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
+import org.mockito.Mockito;
import java.io.IOException;
import java.util.List;
@@ -174,7 +178,6 @@ public class InnerSpaceCompactionSelectorTest extends AbstractCompactionTest {
task,
true,
new FastCompactionPerformer(false),
- CompactionTaskManager.currentTaskNum,
tsFileManager.getNextCompactionTaskId());
// set file status to COMPACTION_CANDIDATE
if (idx == 0) {
@@ -275,7 +278,6 @@ public class InnerSpaceCompactionSelectorTest extends AbstractCompactionTest {
task,
true,
new FastCompactionPerformer(false),
- CompactionTaskManager.currentTaskNum,
tsFileManager.getNextCompactionTaskId());
if (!innerSpaceCompactionTask.setSourceFilesToCompactionCandidate()) {
@@ -289,9 +291,19 @@ public class InnerSpaceCompactionSelectorTest extends AbstractCompactionTest {
cd1.countDown();
cd2.await();
- if (innerSpaceCompactionTask.checkValidAndSetMerging()) {
- throw new RuntimeException("cross space compaction task should be invalid.");
- }
+ Assert.assertThrows(
+ "inner space compaction task should be invalid.",
+ FileCannotTransitToCompactingException.class,
+ innerSpaceCompactionTask::transitSourceFilesToMerging);
+
+ FixedPriorityBlockingQueue<AbstractCompactionTask> mockQueue =
+ Mockito.mock(FixedPriorityBlockingQueue.class);
+ Mockito.when(mockQueue.take())
+ .thenReturn(innerSpaceCompactionTask)
+ .thenThrow(new InterruptedException());
+ CompactionWorker worker = new CompactionWorker(0, mockQueue);
+ worker.run();
+
for (int i = 0; i < task.size(); i++) {
TsFileResource resource = task.get(i);
if (i == 1) {
@@ -303,8 +315,10 @@ public class InnerSpaceCompactionSelectorTest extends AbstractCompactionTest {
}
}
} else {
- if (!innerSpaceCompactionTask.checkValidAndSetMerging()) {
- throw new RuntimeException("cross space compaction task should be valid.");
+ try {
+ innerSpaceCompactionTask.transitSourceFilesToMerging();
+ } catch (FileCannotTransitToCompactingException e) {
+ Assert.fail("inner space compaction task should be valid.");
}
for (int i = 0; i < task.size(); i++) {
TsFileResource resource = task.get(i);
@@ -465,7 +479,6 @@ public class InnerSpaceCompactionSelectorTest extends AbstractCompactionTest {
task,
true,
new FastCompactionPerformer(false),
- CompactionTaskManager.currentTaskNum,
tsFileManager.getNextCompactionTaskId());
// set file status to COMPACTION_CANDIDATE
if (idx == 0) {
@@ -570,7 +583,6 @@ public class InnerSpaceCompactionSelectorTest extends AbstractCompactionTest {
task,
true,
new FastCompactionPerformer(false),
- CompactionTaskManager.currentTaskNum,
tsFileManager.getNextCompactionTaskId());
if (!innerSpaceCompactionTask.setSourceFilesToCompactionCandidate()) {
@@ -584,9 +596,19 @@ public class InnerSpaceCompactionSelectorTest extends AbstractCompactionTest {
cd1.countDown();
cd2.await();
- if (innerSpaceCompactionTask.checkValidAndSetMerging()) {
- throw new RuntimeException("cross space compaction task should be invalid.");
- }
+ Assert.assertThrows(
+ "inner space compaction task should be invalid.",
+ FileCannotTransitToCompactingException.class,
+ innerSpaceCompactionTask::transitSourceFilesToMerging);
+
+ FixedPriorityBlockingQueue<AbstractCompactionTask> mockQueue =
+ Mockito.mock(FixedPriorityBlockingQueue.class);
+ Mockito.when(mockQueue.take())
+ .thenReturn(innerSpaceCompactionTask)
+ .thenThrow(new InterruptedException());
+ CompactionWorker worker = new CompactionWorker(0, mockQueue);
+ worker.run();
+
for (int i = 0; i < task.size(); i++) {
TsFileResource resource = task.get(i);
if (i == 1) {
@@ -598,8 +620,10 @@ public class InnerSpaceCompactionSelectorTest extends AbstractCompactionTest {
}
}
} else {
- if (!innerSpaceCompactionTask.checkValidAndSetMerging()) {
- throw new RuntimeException("cross space compaction task should be valid.");
+ try {
+ innerSpaceCompactionTask.transitSourceFilesToMerging();
+ } catch (FileCannotTransitToCompactingException e) {
+ Assert.fail("inner space compaction task should be valid.");
}
for (int i = 0; i < task.size(); i++) {
TsFileResource resource = task.get(i);
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/recover/RewriteCrossSpaceCompactionRecoverCompatibleTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/recover/RewriteCrossSpaceCompactionRecoverCompatibleTest.java
index f1e8ab7e402..ed90707cd3c 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/recover/RewriteCrossSpaceCompactionRecoverCompatibleTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/recover/RewriteCrossSpaceCompactionRecoverCompatibleTest.java
@@ -180,7 +180,7 @@ public class RewriteCrossSpaceCompactionRecoverCompatibleTest extends AbstractCo
logWriter.close();
// First seq files lost
- seqResources.get(0).delete();
+ seqResources.get(0).remove();
for (int i = 0; i < seqResources.size(); i++) {
Map<String, Pair<Long, Long>> deleteMap = new HashMap<>();
@@ -352,7 +352,7 @@ public class RewriteCrossSpaceCompactionRecoverCompatibleTest extends AbstractCo
logWriter.close();
// First seq files lost
- seqResources.get(0).delete();
+ seqResources.get(0).remove();
for (int i = 0; i < seqResources.size(); i++) {
Map<String, Pair<Long, Long>> deleteMap = new HashMap<>();
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/recover/SizeTieredCompactionRecoverCompatibleTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/recover/SizeTieredCompactionRecoverCompatibleTest.java
index 931ab53e85a..35793690f82 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/recover/SizeTieredCompactionRecoverCompatibleTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/recover/SizeTieredCompactionRecoverCompatibleTest.java
@@ -138,7 +138,7 @@ public class SizeTieredCompactionRecoverCompatibleTest extends AbstractCompactio
Collections.singletonList(targetResource), true, COMPACTION_TEST_SG);
// first source file does not exist
- seqResources.get(0).delete();
+ seqResources.get(0).remove();
for (int i = 0; i < seqResources.size(); i++) {
Map<String, Pair<Long, Long>> deleteMap = new HashMap<>();
@@ -236,7 +236,7 @@ public class SizeTieredCompactionRecoverCompatibleTest extends AbstractCompactio
Collections.singletonList(targetResource), true, "root.compactionTest");
// first source file does not exist
- unseqResources.get(0).delete();
+ unseqResources.get(0).remove();
for (int i = 0; i < unseqResources.size(); i++) {
Map<String, Pair<Long, Long>> deleteMap = new HashMap<>();
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionUpdateFileCountTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionUpdateFileCountTest.java
index d201e1b44ff..b8fcf47f133 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionUpdateFileCountTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionUpdateFileCountTest.java
@@ -35,7 +35,6 @@ import org.junit.Before;
import org.junit.Test;
import java.io.IOException;
-import java.util.concurrent.atomic.AtomicInteger;
public class CompactionUpdateFileCountTest extends AbstractCompactionTest {
@@ -61,13 +60,7 @@ public class CompactionUpdateFileCountTest extends AbstractCompactionTest {
tsFileManager.addAll(seqResources, true);
InnerSpaceCompactionTask task =
new InnerSpaceCompactionTask(
- 0,
- tsFileManager,
- seqResources,
- true,
- new ReadChunkCompactionPerformer(),
- new AtomicInteger(),
- 0);
+ 0, tsFileManager, seqResources, true, new ReadChunkCompactionPerformer(), 0);
Assert.assertTrue(task.start());
Assert.assertEquals(initSeqFileNum - 1, FileMetrics.getInstance().getFileCount(true));
Assert.assertEquals(initUnSeqFileNum, FileMetrics.getInstance().getFileCount(false));
@@ -84,13 +77,7 @@ public class CompactionUpdateFileCountTest extends AbstractCompactionTest {
tsFileManager.addAll(unseqResources, false);
InnerSpaceCompactionTask task =
new InnerSpaceCompactionTask(
- 0,
- tsFileManager,
- unseqResources,
- false,
- new FastCompactionPerformer(false),
- new AtomicInteger(),
- 0);
+ 0, tsFileManager, unseqResources, false, new FastCompactionPerformer(false), 0);
Assert.assertTrue(task.start());
Assert.assertEquals(initSeqFileNum, FileMetrics.getInstance().getFileCount(true));
Assert.assertEquals(initUnSeqFileNum - 1, FileMetrics.getInstance().getFileCount(false));
@@ -113,7 +100,6 @@ public class CompactionUpdateFileCountTest extends AbstractCompactionTest {
seqResources,
unseqResources,
new FastCompactionPerformer(true),
- new AtomicInteger(0),
0,
0);
Assert.assertTrue(task.start());
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/MultiTsFileDeviceIteratorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/MultiTsFileDeviceIteratorTest.java
index 8d6f18375fe..4373ec85bde 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/MultiTsFileDeviceIteratorTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/MultiTsFileDeviceIteratorTest.java
@@ -54,7 +54,6 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.iotdb.commons.conf.IoTDBConstant.PATH_SEPARATOR;
@@ -403,7 +402,6 @@ public class MultiTsFileDeviceIteratorTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new ReadChunkCompactionPerformer(),
- new AtomicInteger(),
0L);
Assert.assertTrue(task.start());
@@ -549,7 +547,6 @@ public class MultiTsFileDeviceIteratorTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new ReadChunkCompactionPerformer(),
- new AtomicInteger(),
0L);
Assert.assertTrue(task.start());
@@ -694,7 +691,6 @@ public class MultiTsFileDeviceIteratorTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(),
0L);
Assert.assertTrue(task.start());
@@ -839,7 +835,6 @@ public class MultiTsFileDeviceIteratorTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new ReadPointCompactionPerformer(),
- new AtomicInteger(),
0L);
Assert.assertTrue(task.start());
@@ -992,7 +987,6 @@ public class MultiTsFileDeviceIteratorTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(),
0L);
Assert.assertTrue(task.start());
@@ -1137,7 +1131,6 @@ public class MultiTsFileDeviceIteratorTest extends AbstractCompactionTest {
tsFileManager.getTsFileList(true),
true,
new FastCompactionPerformer(false),
- new AtomicInteger(),
0L);
Assert.assertTrue(task.start());