Fix pipe permission retry and table parser progress (#17844) (#17848)

(cherry picked from commit eaa5bcb)

Author: Caideyipi

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/writeback/WriteBackSink.java

@@ -20,6 +20,7 @@
 package org.apache.iotdb.db.pipe.sink.protocol.writeback;
 
 import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkNonReportTimeConfigurableException;
 import org.apache.iotdb.db.auth.AuthorityChecker;
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
 import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent;
@@ -131,7 +132,8 @@ private void doTransfer(
     status = statement.isEmpty() ? RpcUtils.SUCCESS_STATUS : executeStatement(statement);
 
     if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
-      throw new PipeException(
+      throwWriteBackExceptionIfNecessary(
+          status,
           String.format(
               "Transfer PipeInsertNodeTabletInsertionEvent %s error, result status %s",
               pipeInsertNodeTabletInsertionEvent, status));
@@ -162,13 +164,23 @@ private void doTransfer(final PipeRawTabletInsertionEvent pipeRawTabletInsertion
         statement.isEmpty() ? RpcUtils.SUCCESS_STATUS : executeStatement(statement);
 
     if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
-      throw new PipeException(
+      throwWriteBackExceptionIfNecessary(
+          status,
           String.format(
               "Transfer PipeRawTabletInsertionEvent %s error, result status %s",
               pipeRawTabletInsertionEvent, status));
     }
   }
 
+  private static void throwWriteBackExceptionIfNecessary(
+      final TSStatus status, final String exceptionMessage) {
+    if (status.getCode() == TSStatusCode.NO_PERMISSION.getStatusCode()) {
+      throw new PipeRuntimeSinkNonReportTimeConfigurableException(exceptionMessage, Long.MAX_VALUE);
+    }
+
+    throw new PipeException(exceptionMessage);
+  }
+
   private TSStatus executeStatement(final InsertBaseStatement statement) {
     return Coordinator.getInstance()
         .executeForTreeModel(