[IOTDB-2478] Try to fix IoTDBRemovePartitionIT and CompactionSchedulerTest (apache#4967)

Author: THUMarkLau

integration/src/test/java/org/apache/iotdb/db/integration/IoTDBRemovePartitionIT.java

@@ -19,18 +19,19 @@
 
 package org.apache.iotdb.db.integration;
 
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
 import org.apache.iotdb.db.engine.StorageEngine;
 import org.apache.iotdb.db.exception.metadata.IllegalPathException;
 import org.apache.iotdb.db.metadata.path.PartialPath;
 import org.apache.iotdb.db.utils.EnvironmentUtils;
 import org.apache.iotdb.itbase.category.LocalStandaloneTest;
 import org.apache.iotdb.jdbc.Config;
 
-import ch.qos.logback.classic.Level;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
+import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.sql.Connection;
@@ -47,14 +48,12 @@
 @Category({LocalStandaloneTest.class})
 public class IoTDBRemovePartitionIT {
 
+  private static final Logger logger = LoggerFactory.getLogger(IoTDBRemovePartitionIT.class);
+
   private static int partitionInterval = 100;
 
   @Before
   public void setUp() throws Exception {
-    ch.qos.logback.classic.Logger rootLogger =
-        (ch.qos.logback.classic.Logger)
-            LoggerFactory.getLogger(ch.qos.logback.classic.Logger.ROOT_LOGGER_NAME);
-    rootLogger.setLevel(Level.toLevel("trace"));
     EnvironmentUtils.envSetUp();
     StorageEngine.setEnablePartition(true);
     StorageEngine.setTimePartitionInterval(partitionInterval);
@@ -70,6 +69,7 @@ public void tearDown() throws Exception {
 
   @Test
   public void testRemoveNoPartition() throws IllegalPathException {
+    logger.warn("running testRemoveNoPartition");
     StorageEngine.getInstance()
         .removePartitions(
             new PartialPath("root.test1"), (storageGroupName, timePartitionId) -> false);
@@ -94,6 +94,7 @@ public void testRemoveNoPartition() throws IllegalPathException {
 
   @Test
   public void testRemovePartialPartition() throws IllegalPathException {
+    logger.warn("running testRemovePartialPartition");
     StorageEngine.getInstance()
         .removePartitions(
             new PartialPath("root.test1"),
@@ -133,6 +134,7 @@ public void testRemovePartialPartition() throws IllegalPathException {
 
   @Test
   public void testRemoveAllPartition() throws IllegalPathException {
+    logger.warn("running testRemoveAllPartition");
     StorageEngine.getInstance()
         .removePartitions(
             new PartialPath("root.test1"), (storageGroupName, timePartitionId) -> true);
@@ -151,6 +153,7 @@ public void testRemoveAllPartition() throws IllegalPathException {
 
   @Test
   public void testSQLRemovePartition() {
+    logger.warn("running testSQLRemovePartition");
     try (Connection connection =
             DriverManager.getConnection(
                 Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
@@ -172,6 +175,7 @@ public void testSQLRemovePartition() {
 
   @Test
   public void testRemoveOnePartitionAndInsertData() {
+    logger.warn("running testRemoveOnePartitionAndInsertData");
     try (Connection connection =
             DriverManager.getConnection(
                 Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
@@ -196,6 +200,7 @@ public void testRemoveOnePartitionAndInsertData() {
 
   @Test
   public void testRemovePartitionAndInsertUnSeqDataAndMerge() {
+    logger.warn("running testRemovePartitionAndInsertUnSeqDataAndMerge");
     try (Connection connection =
             DriverManager.getConnection(
                 Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
@@ -225,6 +230,7 @@ public void testRemovePartitionAndInsertUnSeqDataAndMerge() {
 
   @Test
   public void testRemovePartitionAndInsertUnSeqDataAndUnSeqDataMerge() {
+    logger.warn("running testRemovePartitionAndInsertUnSeqDataAndUnSeqDataMerge");
     try (Connection connection =
             DriverManager.getConnection(
                 Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
@@ -254,6 +260,7 @@ public void testRemovePartitionAndInsertUnSeqDataAndUnSeqDataMerge() {
 
   @Test
   public void testFlushAndRemoveOnePartitionAndInsertData() {
+    logger.warn("running testFlushAndRemoveOnePartitionAndInsertData");
     try (Connection connection =
             DriverManager.getConnection(
                 Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
@@ -312,6 +319,9 @@ private static void insertData() throws ClassNotFoundException {
       }
     }
     Class.forName(Config.JDBC_DRIVER_NAME);
+    IoTDBDescriptor.getInstance().getConfig().setEnableSeqSpaceCompaction(false);
+    IoTDBDescriptor.getInstance().getConfig().setEnableUnseqSpaceCompaction(false);
+    IoTDBDescriptor.getInstance().getConfig().setEnableCrossSpaceCompaction(false);
     try (Connection connection =
             DriverManager.getConnection(
                 Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");

server/src/main/java/org/apache/iotdb/db/engine/compaction/CompactionTaskManager.java

@@ -346,8 +346,13 @@ public synchronized void submitTask(
    * corresponding storage group.
    */
   public void abortCompaction(String fullStorageGroupName) {
+    logger.warn("Aborting compaction task for {}", fullStorageGroupName);
     Set<Future<Void>> subTasks =
         storageGroupTasks.getOrDefault(fullStorageGroupName, Collections.emptySet());
+    logger.warn("Compaction task queue is {}", candidateCompactionTaskQueue);
+    candidateCompactionTaskQueue.clear();
+    logger.warn("the task map is {}", storageGroupTasks);
+    logger.warn("the size of compaction task of {} is {}", fullStorageGroupName, subTasks.size());
     Iterator<Future<Void>> subIterator = subTasks.iterator();
     while (subIterator.hasNext()) {
       Future<Void> next = subIterator.next();

server/src/main/java/org/apache/iotdb/db/engine/storagegroup/VirtualStorageGroupProcessor.java

@@ -3054,8 +3054,10 @@ public void removePartitions(TimePartitionFilter filter) {
     // this requires blocking all other activities
     writeLock("removePartitions");
     try {
+      tsFileManager.setAllowCompaction(false);
       // abort ongoing comapctions and merges
-      CompactionTaskManager.getInstance().abortCompaction(logicalStorageGroupName);
+      CompactionTaskManager.getInstance()
+          .abortCompaction(logicalStorageGroupName + "-" + virtualStorageGroupId);
       // close all working files that should be removed
       removePartitions(filter, workSequenceTsFileProcessors.entrySet(), true);
       removePartitions(filter, workUnsequenceTsFileProcessors.entrySet(), false);

server/src/main/java/org/apache/iotdb/db/utils/datastructure/FixedPriorityBlockingQueue.java

@@ -137,7 +137,9 @@ public void clear() {
     final ReentrantLock lock = this.lock;
     lock.lock();
     try {
-      queue.clear();
+      while (queue.size() != 0) {
+        this.pollLast();
+      }
     } finally {
       lock.unlock();
     }
@@ -180,4 +182,9 @@ public int getMaxSize() {
   public interface PollLastHook<T> {
     void apply(T x);
   }
+
+  @Override
+  public String toString() {
+    return queue.toString();
+  }
 }