Pipe: Fixed the check for no permission 2 (apache#16804)

* edge

* sonar

* emperor-cloth

* emperor-patch

Author: Caideyipi

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/sink/protocol/IoTDBConfigRegionAirGapSink.java

@@ -191,7 +191,8 @@ private void doTransfer(
           new TSStatus(TSStatusCode.PIPE_RECEIVER_USER_CONFLICT_EXCEPTION.getStatusCode())
               .setMessage(errorMessage),
           errorMessage,
-          pipeConfigRegionWritePlanEvent.toString());
+          pipeConfigRegionWritePlanEvent.toString(),
+          true);
     }
   }
 
@@ -252,7 +253,8 @@ private void doTransfer(
           new TSStatus(TSStatusCode.PIPE_RECEIVER_USER_CONFLICT_EXCEPTION.getStatusCode())
               .setMessage(errorMessage),
           errorMessage,
-          pipeConfigRegionSnapshotEvent.toString());
+          pipeConfigRegionSnapshotEvent.toString(),
+          true);
     } else {
       LOGGER.info("Successfully transferred config region snapshot {}.", snapshot);
     }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/sink/protocol/IoTDBConfigRegionSink.java

@@ -188,7 +188,8 @@ private void doTransfer(final PipeConfigRegionWritePlanEvent pipeConfigRegionWri
           String.format(
               "Transfer config region write plan %s error, result status %s.",
               pipeConfigRegionWritePlanEvent.getConfigPhysicalPlan().getType(), status),
-          pipeConfigRegionWritePlanEvent.getConfigPhysicalPlan().toString());
+          pipeConfigRegionWritePlanEvent.getConfigPhysicalPlan().toString(),
+          true);
     }
 
     if (LOGGER.isDebugEnabled()) {
@@ -279,7 +280,8 @@ private void doTransfer(final PipeConfigRegionSnapshotEvent snapshotEvent)
           String.format(
               "Seal config region snapshot file %s error, result status %s.",
               snapshotFile, resp.getStatus()),
-          snapshotFile.toString());
+          snapshotFile.toString(),
+          true);
     }
 
     LOGGER.info("Successfully transferred config region snapshot {}.", snapshotFile);

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBSchemaRegionAirGapSink.java

@@ -122,7 +122,8 @@ private void doTransfer(
           new TSStatus(TSStatusCode.PIPE_RECEIVER_USER_CONFLICT_EXCEPTION.getStatusCode())
               .setMessage(errorMessage),
           errorMessage,
-          pipeSchemaRegionWritePlanEvent.toString());
+          pipeSchemaRegionWritePlanEvent.toString(),
+          true);
     }
   }
 
@@ -187,7 +188,8 @@ private void doTransfer(
           new TSStatus(TSStatusCode.PIPE_RECEIVER_USER_CONFLICT_EXCEPTION.getStatusCode())
               .setMessage(errorMessage),
           errorMessage,
-          pipeSchemaRegionSnapshotEvent.toString());
+          pipeSchemaRegionSnapshotEvent.toString(),
+          true);
     } else {
       LOGGER.info(
           "Successfully transferred schema region snapshot {}, {} and {}.",

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/pipeconsensus/handler/PipeConsensusDeleteEventHandler.java

@@ -79,7 +79,7 @@ public void onComplete(TPipeConsensusTransferResp response) {
       // Only handle the failed statuses to avoid string format performance overhead
       if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()
           && status.getCode() != TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()) {
-        connector.statusHandler().handle(status, status.getMessage(), event.toString());
+        connector.statusHandler().handle(status, status.getMessage(), event.toString(), true);
       }
       event.decreaseReferenceCount(PipeConsensusDeleteEventHandler.class.getName(), true);
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java

@@ -253,7 +253,8 @@ private void doTransfer(final PipeDeleteDataNodeEvent pipeDeleteDataNodeEvent)
           String.format(
               "Transfer deletion %s error, result status %s.",
               pipeDeleteDataNodeEvent.getDeleteDataNode().getType(), status),
-          pipeDeleteDataNodeEvent.getDeletionResource().toString());
+          pipeDeleteDataNodeEvent.getDeletionResource().toString(),
+          true);
     }
 
     if (LOGGER.isDebugEnabled()) {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBSchemaRegionSink.java

@@ -127,7 +127,8 @@ private void doTransfer(final PipeSchemaRegionWritePlanEvent pipeSchemaRegionWri
           String.format(
               "Transfer data node write plan %s error, result status %s.",
               pipeSchemaRegionWritePlanEvent.getPlanNode().getType(), status),
-          pipeSchemaRegionWritePlanEvent.getPlanNode().toString());
+          pipeSchemaRegionWritePlanEvent.getPlanNode().toString(),
+          true);
     }
 
     if (LOGGER.isDebugEnabled()) {
@@ -222,7 +223,8 @@ private void doTransfer(final PipeSchemaRegionSnapshotEvent snapshotEvent)
           String.format(
               "Seal file %s, %s and %s error, result status %s.",
               mTreeSnapshotFile, tagLogSnapshotFile, attributeSnapshotFile, resp.getStatus()),
-          snapshotEvent.toString());
+          snapshotEvent.toString(),
+          true);
     }
 
     LOGGER.info(

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

@@ -27,6 +27,7 @@
 import org.apache.iotdb.commons.pipe.config.PipeConfig;
 import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
 import org.apache.iotdb.commons.utils.RetryUtils;
+import org.apache.iotdb.commons.utils.TestOnly;
 import org.apache.iotdb.pipe.api.event.Event;
 import org.apache.iotdb.pipe.api.exception.PipeException;
 import org.apache.iotdb.rpc.TSStatusCode;
@@ -44,7 +45,7 @@
 
 public class PipeReceiverStatusHandler {
 
-  private static final Logger LOGGER = LoggerFactory.getLogger(PipeReceiverStatusHandler.class);
+  private static 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";
@@ -86,6 +87,11 @@ public PipeReceiverStatusHandler(
     this.skipIfNoPrivileges = skipIfNoPrivileges;
   }
 
+  public void handle(
+      final TSStatus status, final String exceptionMessage, final String recordMessage) {
+    handle(status, exceptionMessage, recordMessage, false);
+  }
+
   /**
    * Handle {@link TSStatus} returned by receiver. Do nothing if ignore the {@link Event}, and throw
    * exception if retry the {@link Event}. Upper class must ensure that the method is invoked only
@@ -99,7 +105,10 @@ public PipeReceiverStatusHandler(
    *     put any time-related info here
    */
   public void handle(
-      final TSStatus status, final String exceptionMessage, final String recordMessage) {
+      final TSStatus status,
+      final String exceptionMessage,
+      final String recordMessage,
+      final boolean log4NoPrivileges) {
 
     if (RetryUtils.needRetryForWrite(status.getCode())) {
       LOGGER.info("IoTConsensusV2: will retry with increasing interval. status: {}", status);
@@ -184,17 +193,28 @@ public void handle(
 
       case 803: // NO_PERMISSION
         if (skipIfNoPrivileges) {
+          if (log4NoPrivileges && LOGGER.isWarnEnabled()) {
+            LOGGER.warn(
+                "{}: Skip if no privileges. will be ignored. event: {}. status: {}",
+                getNoPermission(true),
+                shouldRecordIgnoredDataWhenOtherExceptionsOccur ? recordMessage : "not recorded",
+                status);
+          }
           return;
         }
         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) {
+            if (log4NoPrivileges && LOGGER.isWarnEnabled()) {
+              LOGGER.warn(
+                  "{}: Skip if no privileges. will be ignored. event: {}. status: {}",
+                  getNoPermission(true),
+                  shouldRecordIgnoredDataWhenOtherExceptionsOccur ? recordMessage : "not recorded",
+                  status);
+            }
             return;
           }
           handleOtherExceptions(status, exceptionMessage, recordMessage, true);
@@ -314,4 +334,9 @@ public static TSStatus getPriorStatus(final List<TSStatus> givenStatusList) {
     resultStatus.setSubStatus(givenStatusList);
     return resultStatus;
   }
+
+  @TestOnly
+  public static void setLogger(final Logger logger) {
+    LOGGER = logger;
+  }
 }