fix error log caused by ClosedByInterruptException in compaction module

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/InnerSpaceCompactionTask.java

@@ -40,12 +40,12 @@
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.generator.TsFileNameGenerator;
 import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
+import org.apache.iotdb.tsfile.exception.StopReadTsFileByInterruptException;
 import org.apache.iotdb.tsfile.exception.write.TsFileNotCompleteException;
 import org.apache.iotdb.tsfile.utils.TsFileUtils;
 
 import java.io.File;
 import java.io.IOException;
-import java.nio.channels.ClosedByInterruptException;
 import java.nio.file.Files;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -486,8 +486,8 @@ public long getEstimatedMemoryCost() {
     if (innerSpaceEstimator != null && memoryCost == 0L) {
       try {
         memoryCost = innerSpaceEstimator.estimateInnerCompactionMemory(selectedTsFileResourceList);
-      } catch (IOException e) {
-        if (e instanceof ClosedByInterruptException || Thread.interrupted()) {
+      } catch (Exception e) {
+        if (e instanceof StopReadTsFileByInterruptException || Thread.interrupted()) {
           Thread.currentThread().interrupt();
           return -1;
         }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/io/CompactionTsFileInput.java

@@ -0,0 +1,104 @@
+/*
+ * 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.io;
+
+import org.apache.iotdb.tsfile.exception.StopReadTsFileByInterruptException;
+import org.apache.iotdb.tsfile.read.reader.TsFileInput;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+
+public class CompactionTsFileInput implements TsFileInput {
+  private final TsFileInput tsFileInput;
+
+  public CompactionTsFileInput(TsFileInput tsFileInput) {
+    this.tsFileInput = tsFileInput;
+  }
+
+  @Override
+  public long size() throws IOException {
+    try {
+      return tsFileInput.size();
+    } catch (Exception e) {
+      if (Thread.currentThread().isInterrupted()) {
+        throw new StopReadTsFileByInterruptException();
+      }
+      throw e;
+    }
+  }
+
+  @Override
+  public long position() throws IOException {
+    try {
+      return tsFileInput.position();
+    } catch (Exception e) {
+      if (Thread.currentThread().isInterrupted()) {
+        throw new StopReadTsFileByInterruptException();
+      }
+      throw e;
+    }
+  }
+
+  @Override
+  public TsFileInput position(long newPosition) throws IOException {
+    try {
+      return tsFileInput.position(newPosition);
+    } catch (Exception e) {
+      if (Thread.currentThread().isInterrupted()) {
+        throw new StopReadTsFileByInterruptException();
+      }
+      throw e;
+    }
+  }
+
+  @Override
+  public int read(ByteBuffer dst) throws IOException {
+    int readSize = tsFileInput.read(dst);
+    if (Thread.currentThread().isInterrupted()) {
+      throw new StopReadTsFileByInterruptException();
+    }
+    return readSize;
+  }
+
+  @Override
+  public int read(ByteBuffer dst, long position) throws IOException {
+    int readSize = tsFileInput.read(dst, position);
+    if (Thread.currentThread().isInterrupted()) {
+      throw new StopReadTsFileByInterruptException();
+    }
+    return readSize;
+  }
+
+  @Override
+  public InputStream wrapAsInputStream() throws IOException {
+    return tsFileInput.wrapAsInputStream();
+  }
+
+  @Override
+  public void close() throws IOException {
+    tsFileInput.close();
+  }
+
+  @Override
+  public String getFilePath() {
+    return tsFileInput.getFilePath();
+  }
+}

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/io/CompactionTsFileReader.java

@@ -65,6 +65,7 @@ public class CompactionTsFileReader extends TsFileSequenceReader {
    */
   public CompactionTsFileReader(String file, CompactionType compactionType) throws IOException {
     super(file);
+    this.tsFileInput = new CompactionTsFileInput(tsFileInput);
     this.compactionType = compactionType;
   }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/repair/RepairDataFileScanUtil.java

@@ -83,15 +83,15 @@ public void scanTsFile() {
           checkNonAlignedDeviceSeries(reader, device);
         }
       }
-    } catch (IOException ioException) {
+    } catch (CompactionLastTimeCheckFailedException lastTimeCheckFailedException) {
+      this.hasUnsortedData = true;
+    } catch (Exception e) {
       // ignored the exception caused by thread interrupt
       if (Thread.currentThread().isInterrupted()) {
         return;
       }
-      logger.warn("Meet error when read tsfile {}", tsfile.getAbsolutePath(), ioException);
+      logger.warn("Meet error when read tsfile {}", tsfile.getAbsolutePath(), e);
       isBrokenFile = true;
-    } catch (CompactionLastTimeCheckFailedException lastTimeCheckFailedException) {
-      this.hasUnsortedData = true;
     }
   }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduler.java

@@ -184,7 +184,11 @@ private static int addTaskToWaitingQueue(List<? extends AbstractCompactionTask>
     return trySubmitCount;
   }
 
-  private static boolean canAddTaskToWaitingQueue(AbstractCompactionTask task) {
+  private static boolean canAddTaskToWaitingQueue(AbstractCompactionTask task)
+      throws InterruptedException {
+    if (Thread.interrupted()) {
+      throw new InterruptedException();
+    }
     // check file num
     long fileNumLimitForCompaction = SystemInfo.getInstance().getTotalFileLimitForCompaction();
     if (task.getProcessedFileNum() > fileNumLimitForCompaction) {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/estimator/AbstractCompactionEstimator.java

@@ -64,6 +64,8 @@ public abstract class AbstractCompactionEstimator {
 
   protected abstract long calculatingDataMemoryCost(CompactionTaskInfo taskInfo) throws IOException;
 
+  protected abstract TsFileSequenceReader getReader(String filePath) throws IOException;
+
   protected boolean isAllSourceFileExist(List<TsFileResource> resources) {
     for (TsFileResource resource : resources) {
       if (resource.getStatus() == TsFileResourceStatus.DELETED) {
@@ -95,8 +97,7 @@ private FileInfo getFileInfoFromCache(TsFileResource resource) throws IOExceptio
         return fileInfo;
       }
     }
-    try (TsFileSequenceReader reader =
-        new TsFileSequenceReader(resource.getTsFilePath(), true, false)) {
+    try (TsFileSequenceReader reader = getReader(resource.getTsFilePath())) {
       FileInfo fileInfo = CompactionEstimateUtils.calculateFileInfo(reader);
       fileInfoCache.put(resource, fileInfo);
       synchronized (globalFileInfoCacheForFailedCompaction) {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/estimator/AbstractCrossSpaceEstimator.java

@@ -19,7 +19,10 @@
 
 package org.apache.iotdb.db.storageengine.dataregion.compaction.selector.estimator;
 
+import org.apache.iotdb.db.storageengine.dataregion.compaction.io.CompactionTsFileReader;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.schedule.constant.CompactionType;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -31,6 +34,11 @@
  */
 public abstract class AbstractCrossSpaceEstimator extends AbstractCompactionEstimator {
 
+  @Override
+  protected TsFileSequenceReader getReader(String filePath) throws IOException {
+    return new CompactionTsFileReader(filePath, CompactionType.CROSS_COMPACTION);
+  }
+
   public long estimateCrossCompactionMemory(
       List<TsFileResource> seqResources, List<TsFileResource> unseqResources) throws IOException {
     List<TsFileResource> resources = new ArrayList<>(seqResources.size() + unseqResources.size());

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/estimator/AbstractInnerSpaceEstimator.java

@@ -19,7 +19,11 @@
 
 package org.apache.iotdb.db.storageengine.dataregion.compaction.selector.estimator;
 
+import org.apache.iotdb.commons.conf.IoTDBConstant;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.io.CompactionTsFileReader;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.schedule.constant.CompactionType;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
 
 import java.io.IOException;
 import java.util.List;
@@ -30,6 +34,15 @@
  */
 public abstract class AbstractInnerSpaceEstimator extends AbstractCompactionEstimator {
 
+  @Override
+  protected TsFileSequenceReader getReader(String filePath) throws IOException {
+    if (filePath.contains(IoTDBConstant.UNSEQUENCE_FOLDER_NAME)) {
+      return new CompactionTsFileReader(filePath, CompactionType.INNER_UNSEQ_COMPACTION);
+    } else {
+      return new CompactionTsFileReader(filePath, CompactionType.INNER_SEQ_COMPACTION);
+    }
+  }
+
   public long estimateInnerCompactionMemory(List<TsFileResource> resources) throws IOException {
     if (!CompactionEstimateUtils.addReadLock(resources)) {
       return -1L;

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/impl/RewriteCrossSpaceCompactionSelector.java

@@ -38,12 +38,12 @@
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.generator.TsFileNameGenerator;
 import org.apache.iotdb.db.storageengine.rescon.memory.SystemInfo;
+import org.apache.iotdb.tsfile.exception.StopReadTsFileByInterruptException;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
-import java.nio.channels.ClosedByInterruptException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -128,8 +128,8 @@ public CrossCompactionTaskResource selectOneTaskResources(CrossSpaceCompactionCa
           candidate.getUnseqFiles().size());
 
       return executeTaskResourceSelection(candidate);
-    } catch (IOException e) {
-      if (e instanceof ClosedByInterruptException || Thread.interrupted()) {
+    } catch (Exception e) {
+      if (e instanceof StopReadTsFileByInterruptException || Thread.interrupted()) {
         Thread.currentThread().interrupt();
         return new CrossCompactionTaskResource();
       }

iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/exception/StopReadTsFileByInterruptException.java

@@ -0,0 +1,24 @@
+/*
+ * 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.tsfile.exception;
+
+import java.io.IOException;
+
+public class StopReadTsFileByInterruptException extends IOException {}