Pipe: Added log control for pipe receiver auth exception (apache#16659)

Author: Caideyipi

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java

@@ -831,7 +831,7 @@ private TSStatus executeStatementAndClassifyExceptions(
             LOGGER::warn,
             "Receiver id = %s: Failure status encountered while executing statement %s: %s",
             receiverId.get(),
-            statement,
+            statement.getPipeLoggingString(),
             result);
         return statement.accept(STATEMENT_STATUS_VISITOR, result);
       }
@@ -840,7 +840,7 @@ private TSStatus executeStatementAndClassifyExceptions(
           LOGGER::warn,
           "Receiver id = %s: Exception encountered while executing statement %s: ",
           receiverId.get(),
-          statement,
+          statement.getPipeLoggingString(),
           e);
       return statement.accept(STATEMENT_EXCEPTION_VISITOR, e);
     } finally {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/Statement.java

@@ -64,4 +64,8 @@ public org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Statement toRelat
       final MPPQueryContext context) {
     throw new UnsupportedOperationException("Method not implemented yet");
   }
+
+  public String getPipeLoggingString() {
+    return toString();
+  }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertMultiTabletsStatement.java

@@ -200,4 +200,12 @@ public void removeAttributeColumns() {
   protected void subRemoveAttributeColumns(List<Integer> columnsToKeep) {
     insertTabletStatementList.forEach(InsertBaseStatement::removeAttributeColumns);
   }
+
+  @Override
+  public String toString() {
+    return "InsertMultiTabletsStatement{"
+        + "insertTabletStatementList="
+        + insertTabletStatementList
+        + '}';
+  }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertRowStatement.java

@@ -548,4 +548,28 @@ protected void subRemoveAttributeColumns(List<Integer> columnsToKeep) {
       values = columnsToKeep.stream().map(i -> values[i]).toArray();
     }
   }
+
+  @Override
+  public String getPipeLoggingString() {
+    return "InsertRowNode{"
+        + "deviceID="
+        + deviceID
+        + ", time="
+        + time
+        + ", valueLength="
+        + (Objects.nonNull(values) ? values.length : 0)
+        + '}';
+  }
+
+  @Override
+  public String toString() {
+    return "InsertRowNode{"
+        + "deviceID="
+        + deviceID
+        + ", time="
+        + time
+        + ", values="
+        + Arrays.toString(values)
+        + '}';
+  }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertRowsStatement.java

@@ -237,4 +237,9 @@ public void removeAttributeColumns() {
   protected void subRemoveAttributeColumns(List<Integer> columnsToKeep) {
     insertRowStatementList.forEach(InsertBaseStatement::removeAttributeColumns);
   }
+
+  @Override
+  public String toString() {
+    return "InsertRowsStatement{" + "insertRowStatementList=" + insertRowStatementList + '}';
+  }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertTabletStatement.java

@@ -640,4 +640,21 @@ protected void subRemoveAttributeColumns(List<Integer> columnsToKeep) {
       nullBitMaps = columnsToKeep.stream().map(i -> nullBitMaps[i]).toArray(BitMap[]::new);
     }
   }
+
+  @Override
+  public String toString() {
+    return "InsertTabletStatement{"
+        + "deviceIDs="
+        + Arrays.toString(deviceIDs)
+        + ", measurements="
+        + Arrays.toString(measurements)
+        + ", rowCount="
+        + rowCount
+        + ", timeRange=["
+        + (Objects.nonNull(times) && times.length > 0
+            ? times[0] + ", " + times[times.length - 1]
+            : "")
+        + "]"
+        + '}';
+  }
 }

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/pipe/PipeRuntimeSinkCriticalException.java

@@ -93,7 +93,7 @@ public static PipeRuntimeSinkCriticalException deserializeFrom(
 
   @Override
   public String toString() {
-    return "PipeRuntimeConnectorCriticalException{"
+    return "PipeRuntimeSinkCriticalException{"
         + "message='"
         + getMessage()
         + "', timeStamp="

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/pipe/PipeRuntimeSinkRetryTimesConfigurableException.java

@@ -33,4 +33,10 @@ public PipeRuntimeSinkRetryTimesConfigurableException(
   public int getRetryTimes() {
     return retryTimes;
   }
+
+  // We do not record the timestamp here for logger reduction detection
+  @Override
+  public String toString() {
+    return "PipeRuntimeSinkRetryTimesConfigurableException{" + "message='" + getMessage() + "}";
+  }
 }