Merge branch 'master' of https://github.com/apache/iotdb into fix-audit-logger

Author: Caideyipi

iotdb-core/ainode/iotdb/ainode/core/ai_node.py

@@ -163,7 +163,11 @@ def stop(self):
             self._rpc_handler.stop()
             if self._rpc_service:
                 self._rpc_service.stop()
-                self._rpc_service.join(1)
-                if self._rpc_service.is_alive():
-                    logger.warning("RPC service thread failed to stop in time.")
+                for retry in range(30):
+                    self._rpc_service.join(2)
+                    if not self._rpc_service.is_alive():
+                        logger.warning(
+                            "RPC service thread failed to stop in time, retrying..."
+                        )
+                        break
             logger.info("IoTDB-AINode has successfully stopped.")

iotdb-core/ainode/iotdb/ainode/core/inference/pool_controller.py

@@ -504,7 +504,7 @@ def get_load(self, model_id, device_id, pool_id) -> int:
             return pool_group.get_load(pool_id)
         return -1
 
-    def shutdown(self):
+    def stop(self):
         self._stop_event.set()
 
         # shutdown pool controller

iotdb-core/ainode/iotdb/ainode/core/manager/inference_manager.py

@@ -366,9 +366,9 @@ def inference(self, req: TInferenceReq):
             single_output=False,
         )
 
-    def shutdown(self):
+    def stop(self):
         self._stop_event.set()
-        self._pool_controller.shutdown()
+        self._pool_controller.stop()
         while not self._result_queue.empty():
             self._result_queue.get_nowait()
         self._result_queue.close()

iotdb-core/ainode/iotdb/ainode/core/rpc/handler.py

@@ -69,7 +69,7 @@ def __init__(self, ainode):
 
     def stop(self) -> None:
         logger.info("Stopping the RPC service handler of IoTDB-AINode...")
-        self._inference_manager.shutdown()
+        self._inference_manager.stop()
 
     def stopAINode(self) -> TSStatus:
         self._ainode.stop()

iotdb-core/ainode/iotdb/ainode/core/rpc/service.py

@@ -125,3 +125,4 @@ def stop(self) -> None:
             logger.info("Stopping the RPC service of IoTDB-AINode...")
             self._stop_event.set()
             self.__pool_server.stop()
+            self._handler.stop()

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

@@ -164,7 +164,8 @@ protected void handleException(Logger logger, Exception e) {
     } else if (e instanceof InterruptedException
         || Thread.interrupted()
         || e instanceof StopReadTsFileByInterruptException
-        || !tsFileManager.isAllowCompaction()) {
+        || !tsFileManager.isAllowCompaction()
+        || CompactionTaskManager.getInstance().isStopAllCompactionWorker()) {
       logger.warn(
           "{}-{} [Compaction] {} task interrupted",
           storageGroupName,

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

@@ -116,14 +116,14 @@ public InnerSpaceCompactionTask(
   }
 
   protected static class InnerCompactionTaskFilesView {
-    protected List<TsFileResource> sortedAllSourceFilesInTask;
-    protected List<TsFileResource> sourceFilesInCompactionPerformer;
-    protected List<TsFileResource> skippedSourceFiles;
+    protected List<TsFileResource> sortedAllSourceFilesInTask = Collections.emptyList();
+    protected List<TsFileResource> sourceFilesInCompactionPerformer = Collections.emptyList();
+    protected List<TsFileResource> skippedSourceFiles = Collections.emptyList();
     protected boolean sequence;
-    protected List<TsFileResource> sourceFilesInLog;
-    protected List<TsFileResource> targetFilesInLog;
-    protected List<TsFileResource> targetFilesInPerformer;
-    protected List<TsFileResource> renamedTargetFiles;
+    protected List<TsFileResource> sourceFilesInLog = Collections.emptyList();
+    protected List<TsFileResource> targetFilesInLog = Collections.emptyList();
+    protected List<TsFileResource> targetFilesInPerformer = Collections.emptyList();
+    protected List<TsFileResource> renamedTargetFiles = Collections.emptyList();
 
     protected long selectedFileSize;
     protected int sumOfCompactionCount;

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionTestFileWriter.java

@@ -19,6 +19,8 @@
 
 package org.apache.iotdb.db.storageengine.dataregion.compaction.utils;
 
+import org.apache.iotdb.db.storageengine.dataregion.compaction.io.CompactionTsFileWriter;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.schedule.constant.CompactionType;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
 
 import org.apache.tsfile.enums.TSDataType;
@@ -49,7 +51,9 @@ public class CompactionTestFileWriter implements Closeable {
 
   public CompactionTestFileWriter(TsFileResource emptyFile) throws IOException {
     this.resource = emptyFile;
-    fileWriter = new TsFileIOWriter(emptyFile.getTsFile());
+    fileWriter =
+        new CompactionTsFileWriter(
+            emptyFile.getTsFile(), 1024 * 1024 * 1024, CompactionType.INNER_SEQ_COMPACTION);
   }
 
   public IDeviceID startChunkGroup(String deviceNameWithoutParentPath) throws IOException {

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/PipeReceiverStatusHandler.java

@@ -45,6 +45,9 @@
 public class PipeReceiverStatusHandler {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(PipeReceiverStatusHandler.class);
+  private static final String NO_PERMISSION = "No permission";
+  private static final String UNCLASSIFIED_EXCEPTION = "Unclassified exception";
+  private static final String NO_PERMISSION_STR = "No permissions for this operation";
 
   private static final int CONFLICT_RETRY_MAX_TIMES = 100;
 
@@ -183,83 +186,74 @@ public void handle(
         if (skipIfNoPrivileges) {
           return;
         }
-
-        synchronized (this) {
-          recordExceptionStatusIfNecessary(recordMessage);
-
-          if (exceptionEventHasBeenRetried.get()
-              && System.currentTimeMillis() - exceptionFirstEncounteredTime.get()
-                  > retryMaxMillisWhenOtherExceptionsOccur) {
-            LOGGER.warn(
-                "No permission: retry timeout. will be ignored. event: {}. status: {}",
-                shouldRecordIgnoredDataWhenOtherExceptionsOccur ? recordMessage : "not recorded",
-                status);
-            resetExceptionStatus();
+        handleOtherExceptions(status, exceptionMessage, recordMessage, true);
+        break;
+      case 305:
+        handleOtherExceptions(status, exceptionMessage, recordMessage, false);
+        break;
+      default:
+        // Some auth error may be wrapped in other codes
+        if (exceptionMessage.contains(NO_PERMISSION_STR)) {
+          if (skipIfNoPrivileges) {
             return;
           }
-
-          // Reduce the log if retry forever
-          if (retryMaxMillisWhenOtherExceptionsOccur == Long.MAX_VALUE) {
-            PipeLogger.log(LOGGER::warn, "No permission: will retry forever. status: %s", status);
-          } else {
-            LOGGER.warn(
-                "No permission: will retry for at least {} seconds. status: {}",
-                (retryMaxMillisWhenOtherExceptionsOccur
-                        + exceptionFirstEncounteredTime.get()
-                        - System.currentTimeMillis())
-                    / 1000.0,
-                status);
-          }
-
-          exceptionEventHasBeenRetried.set(true);
-          throw new PipeRuntimeSinkRetryTimesConfigurableException(
-              exceptionMessage,
-              (int)
-                  Math.max(
-                      PipeSubtask.MAX_RETRY_TIMES,
-                      Math.min(
-                          CONFLICT_RETRY_MAX_TIMES, retryMaxMillisWhenOtherExceptionsOccur * 1.1)));
+          handleOtherExceptions(status, exceptionMessage, recordMessage, true);
+          break;
         }
+        // Other exceptions
+        handleOtherExceptions(status, exceptionMessage, recordMessage, false);
+        break;
+    }
+  }
 
-      default: // Other exceptions
-        synchronized (this) {
-          recordExceptionStatusIfNecessary(recordMessage);
+  private synchronized void handleOtherExceptions(
+      final TSStatus status,
+      final String exceptionMessage,
+      final String recordMessage,
+      final boolean noPermission) {
+    recordExceptionStatusIfNecessary(recordMessage);
+
+    if (exceptionEventHasBeenRetried.get()
+        && System.currentTimeMillis() - exceptionFirstEncounteredTime.get()
+            > retryMaxMillisWhenOtherExceptionsOccur) {
+      LOGGER.warn(
+          "{}: retry timeout. will be ignored. event: {}. status: {}",
+          getNoPermission(noPermission),
+          shouldRecordIgnoredDataWhenOtherExceptionsOccur ? recordMessage : "not recorded",
+          status);
+      resetExceptionStatus();
+      return;
+    }
 
-          if (exceptionEventHasBeenRetried.get()
-              && System.currentTimeMillis() - exceptionFirstEncounteredTime.get()
-                  > retryMaxMillisWhenOtherExceptionsOccur) {
-            LOGGER.warn(
-                "Unclassified exception: retry timeout. will be ignored. event: {}. status: {}",
-                shouldRecordIgnoredDataWhenOtherExceptionsOccur ? recordMessage : "not recorded",
-                status);
-            resetExceptionStatus();
-            return;
-          }
+    // Reduce the log if retry forever
+    if (retryMaxMillisWhenOtherExceptionsOccur == Long.MAX_VALUE) {
+      PipeLogger.log(
+          LOGGER::warn,
+          "%s: will retry forever. status: %s",
+          getNoPermission(noPermission),
+          status);
+    } else {
+      LOGGER.warn(
+          "{}: will retry for at least {} seconds. status: {}",
+          getNoPermission(noPermission),
+          (retryMaxMillisWhenOtherExceptionsOccur
+                  + exceptionFirstEncounteredTime.get()
+                  - System.currentTimeMillis())
+              / 1000.0,
+          status);
+    }
 
-          // Reduce the log if retry forever
-          if (retryMaxMillisWhenOtherExceptionsOccur == Long.MAX_VALUE) {
-            PipeLogger.log(
-                LOGGER::warn, "Unclassified exception: will retry forever. status: %s", status);
-          } else {
-            LOGGER.warn(
-                "Unclassified exception: will retry for at least {} seconds. status: {}",
-                (retryMaxMillisWhenOtherExceptionsOccur
-                        + exceptionFirstEncounteredTime.get()
-                        - System.currentTimeMillis())
-                    / 1000.0,
-                status);
-          }
+    exceptionEventHasBeenRetried.set(true);
+    throw new PipeRuntimeSinkRetryTimesConfigurableException(
+        exceptionMessage,
+        (int)
+            Math.max(
+                PipeSubtask.MAX_RETRY_TIMES,
+                Math.min(CONFLICT_RETRY_MAX_TIMES, retryMaxMillisWhenOtherExceptionsOccur * 1.1)));
+  }
 
-          exceptionEventHasBeenRetried.set(true);
-          throw new PipeRuntimeSinkRetryTimesConfigurableException(
-              exceptionMessage,
-              (int)
-                  Math.max(
-                      PipeSubtask.MAX_RETRY_TIMES,
-                      Math.min(
-                          CONFLICT_RETRY_MAX_TIMES, retryMaxMillisWhenOtherExceptionsOccur * 1.1)));
-        }
-    }
+  private static String getNoPermission(final boolean noPermission) {
+    return noPermission ? NO_PERMISSION : UNCLASSIFIED_EXCEPTION;
   }
 
   private void recordExceptionStatusIfNecessary(final String message) {

scripts/sbin/start-ainode.sh

@@ -27,10 +27,14 @@ export IOTDB_AINODE_HOME
 echo "IOTDB_AINODE_HOME: $IOTDB_AINODE_HOME"
 
 # fetch parameters with names
-while getopts "i:rn" opt; do
+daemon_mode=false
+while getopts "i:rnd" opt; do
   case $opt in
     n)
     ;;
+    d)
+      daemon_mode=true
+    ;;
     \?) echo "Invalid option -$OPTARG" >&2
     exit 1
     ;;
@@ -41,6 +45,11 @@ ain_ainode_executable="$IOTDB_AINODE_HOME/lib/ainode"
 
 echo Script got ainode executable: "$ain_ainode_executable"
 
-echo Starting AINode...
-
-$ain_ainode_executable start
+if [ "$daemon_mode" = true ]; then
+  echo Starting AINode in daemon mode...
+  nohup $ain_ainode_executable start > /dev/null 2>&1 &
+  echo AINode started in background
+else
+  echo Starting AINode...
+  $ain_ainode_executable start
+fi