Pipe: Fixed the bug that lower version tablet may cause NPE when sent to 2.x version & The temporary exception may be wrongly reported (#16843)

* older_version_compatibility

* protect

* dependency

Author: Caideyipi

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java

@@ -45,6 +45,7 @@
 import org.apache.iotdb.pipe.api.exception.PipeException;
 
 import com.google.common.util.concurrent.ListeningExecutorService;
+import org.apache.tsfile.external.commons.lang3.exception.ExceptionUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -219,6 +220,12 @@ protected boolean executeOnce() throws Exception {
           e);
       return false;
     } catch (final Exception e) {
+      if (ExceptionUtils.getRootCause(e) instanceof PipeRuntimeOutOfMemoryCriticalException) {
+        LOGGER.info(
+            "Temporarily out of memory in pipe event processing, will wait for the memory to release.",
+            e);
+        return false;
+      }
       if (!isClosed.get()) {
         throw new PipeException(
             String.format(

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java

@@ -468,7 +468,7 @@ public void throwIfNoPrivilege() {
           }
         }
       }
-    } catch (final AccessDeniedException e) {
+    } catch (final AccessDeniedException | PipeRuntimeOutOfMemoryCriticalException e) {
       throw e;
     } catch (final Exception e) {
       if (e instanceof InterruptedException) {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/sorter/PipeTabletEventSorter.java

@@ -27,6 +27,7 @@
 import org.apache.tsfile.write.schema.IMeasurementSchema;
 
 import java.time.LocalDate;
+import java.util.Objects;
 
 public class PipeTabletEventSorter {
 
@@ -84,6 +85,10 @@ protected Object reorderValueListAndBitMap(
       final TSDataType dataType,
       final BitMap originalBitMap,
       final BitMap deDuplicatedBitMap) {
+    // Older version's sender may contain null values, we need to cover this case
+    if (Objects.isNull(valueList)) {
+      return null;
+    }
     switch (dataType) {
       case BOOLEAN:
         final boolean[] boolValues = (boolean[]) valueList;