Pipe: Adjusted the waiting time of temporary unavailable exceptions (apache#16798)

* fff

* fix

* fix

* fix

* fix

* f

Author: Caideyipi

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeSubtask.java

@@ -19,11 +19,13 @@
 
 package org.apache.iotdb.confignode.manager.pipe.agent.task;
 
+import org.apache.iotdb.commons.exception.pipe.PipeNonReportException;
 import org.apache.iotdb.commons.exception.pipe.PipeRuntimeException;
 import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin;
 import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
 import org.apache.iotdb.commons.pipe.agent.task.progress.PipeEventCommitManager;
 import org.apache.iotdb.commons.pipe.agent.task.subtask.PipeAbstractSinkSubtask;
+import org.apache.iotdb.commons.pipe.config.PipeConfig;
 import org.apache.iotdb.commons.pipe.config.constant.PipeProcessorConstant;
 import org.apache.iotdb.commons.pipe.config.plugin.configuraion.PipeTaskRuntimeConfiguration;
 import org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskProcessorRuntimeEnvironment;
@@ -188,7 +190,9 @@ protected boolean executeOnce() throws Exception {
         PipeConfigRegionSinkMetrics.getInstance().markConfigEvent(taskID);
       }
       decreaseReferenceCountAndReleaseLastEvent(event, true);
-
+      sleepInterval = PipeConfig.getInstance().getPipeSinkSubtaskSleepIntervalInitMs();
+    } catch (final PipeNonReportException e) {
+      sleep4NonReportException();
     } catch (final PipeException e) {
       setLastExceptionEvent(event);
       if (!isClosed.get()) {

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

@@ -23,6 +23,7 @@
 import org.apache.iotdb.commons.exception.pipe.PipeRuntimeException;
 import org.apache.iotdb.commons.pipe.agent.task.connection.UnboundedBlockingPendingQueue;
 import org.apache.iotdb.commons.pipe.agent.task.subtask.PipeAbstractSinkSubtask;
+import org.apache.iotdb.commons.pipe.config.PipeConfig;
 import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
 import org.apache.iotdb.commons.pipe.sink.protocol.IoTDBSink;
 import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
@@ -132,8 +133,9 @@ protected boolean executeOnce() {
       }
 
       decreaseReferenceCountAndReleaseLastEvent(event, true);
+      sleepInterval = PipeConfig.getInstance().getPipeSinkSubtaskSleepIntervalInitMs();
     } catch (final PipeNonReportException e) {
-      // Ignore, go directly next round
+      sleep4NonReportException();
     } catch (final PipeException e) {
       if (!isClosed.get()) {
         setLastExceptionEvent(event);

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java

@@ -254,6 +254,9 @@ public class CommonConfig {
   private long pipeSubtaskExecutorBasicCheckPointIntervalByTimeDuration = 10 * 1000L;
   private long pipeSubtaskExecutorPendingQueueMaxBlockingTimeMs = 50;
 
+  private volatile long pipeSinkSubtaskSleepIntervalInitMs = 250L;
+  private volatile long pipeSinkSubtaskSleepIntervalMaxMs = 1000L;
+
   private long pipeSubtaskExecutorCronHeartbeatEventIntervalSeconds = 20;
 
   private long pipeMaxWaitFinishTime = 10 * 1000;
@@ -1415,6 +1418,32 @@ public void setPipeRetryLocallyForParallelOrUserConflict(
         "pipeRetryLocallyForParallelOrUserConflict is set to {}.", pipeSubtaskExecutorMaxThreadNum);
   }
 
+  public long getPipeSinkSubtaskSleepIntervalInitMs() {
+    return pipeSinkSubtaskSleepIntervalInitMs;
+  }
+
+  public void setPipeSinkSubtaskSleepIntervalInitMs(long pipeSinkSubtaskSleepIntervalInitMs) {
+    if (this.pipeSinkSubtaskSleepIntervalInitMs == pipeSinkSubtaskSleepIntervalInitMs) {
+      return;
+    }
+    this.pipeSinkSubtaskSleepIntervalInitMs = pipeSinkSubtaskSleepIntervalInitMs;
+    logger.info(
+        "pipeSinkSubtaskSleepIntervalInitMs is set to {}.", pipeSinkSubtaskSleepIntervalInitMs);
+  }
+
+  public long getPipeSinkSubtaskSleepIntervalMaxMs() {
+    return pipeSinkSubtaskSleepIntervalMaxMs;
+  }
+
+  public void setPipeSinkSubtaskSleepIntervalMaxMs(long pipeSinkSubtaskSleepIntervalMaxMs) {
+    if (this.pipeSinkSubtaskSleepIntervalMaxMs == pipeSinkSubtaskSleepIntervalMaxMs) {
+      return;
+    }
+    this.pipeSinkSubtaskSleepIntervalMaxMs = pipeSinkSubtaskSleepIntervalMaxMs;
+    logger.info(
+        "pipeSinkSubtaskSleepIntervalMaxMs is set to {}.", pipeSinkSubtaskSleepIntervalMaxMs);
+  }
+
   public long getPipeSubtaskExecutorPendingQueueMaxBlockingTimeMs() {
     return pipeSubtaskExecutorPendingQueueMaxBlockingTimeMs;
   }

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java

@@ -53,6 +53,8 @@ public abstract class PipeAbstractSinkSubtask extends PipeReportableSubtask {
   @SuppressWarnings("java:S3077")
   protected volatile Event lastExceptionEvent;
 
+  protected long sleepInterval = PipeConfig.getInstance().getPipeSinkSubtaskSleepIntervalInitMs();
+
   protected PipeAbstractSinkSubtask(
       final String taskID, final long creationTime, final PipeConnector outputPipeSink) {
     super(taskID, creationTime);
@@ -248,4 +250,15 @@ protected synchronized void clearReferenceCountAndReleaseLastExceptionEvent() {
       lastExceptionEvent = null;
     }
   }
+
+  public void sleep4NonReportException() {
+    if (sleepInterval < PipeConfig.getInstance().getPipeSinkSubtaskSleepIntervalMaxMs()) {
+      sleepInterval <<= 1;
+    }
+    try {
+      Thread.sleep(sleepInterval);
+    } catch (final InterruptedException e) {
+      Thread.currentThread().interrupt();
+    }
+  }
 }

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java

@@ -143,6 +143,14 @@ public long getPipeMaxWaitFinishTime() {
     return COMMON_CONFIG.getPipeMaxWaitFinishTime();
   }
 
+  public long getPipeSinkSubtaskSleepIntervalInitMs() {
+    return COMMON_CONFIG.getPipeSinkSubtaskSleepIntervalInitMs();
+  }
+
+  public long getPipeSinkSubtaskSleepIntervalMaxMs() {
+    return COMMON_CONFIG.getPipeSinkSubtaskSleepIntervalMaxMs();
+  }
+
   /////////////////////////////// Source ///////////////////////////////
 
   public int getPipeSourceAssignerDisruptorRingBufferSize() {
@@ -484,6 +492,8 @@ public void printAllConfigs() {
         "PipeSubtaskExecutorCronHeartbeatEventIntervalSeconds: {}",
         getPipeSubtaskExecutorCronHeartbeatEventIntervalSeconds());
     LOGGER.info("PipeMaxWaitFinishTime: {}", getPipeMaxWaitFinishTime());
+    LOGGER.info("PipeSinkSubtaskSleepIntervalInitMs: {}", getPipeSinkSubtaskSleepIntervalInitMs());
+    LOGGER.info("PipeSinkSubtaskSleepIntervalMaxMs: {}", getPipeSinkSubtaskSleepIntervalMaxMs());
 
     LOGGER.info(
         "PipeSourceAssignerDisruptorRingBufferSize: {}",

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java

@@ -296,6 +296,17 @@ public static void loadPipeInternalConfig(CommonConfig config, TrimProperties pr
                 "pipe_retry_locally_for_user_conflict",
                 String.valueOf(config.isPipeRetryLocallyForParallelOrUserConflict()))));
 
+    config.setPipeSinkSubtaskSleepIntervalInitMs(
+        Long.parseLong(
+            properties.getProperty(
+                "pipe_sink_subtask_sleep_interval_init_ms",
+                String.valueOf(config.getPipeSinkSubtaskSleepIntervalInitMs()))));
+    config.setPipeSinkSubtaskSleepIntervalMaxMs(
+        Long.parseLong(
+            properties.getProperty(
+                "pipe_sink_subtask_sleep_interval_max_ms",
+                String.valueOf(config.getPipeSinkSubtaskSleepIntervalMaxMs()))));
+
     config.setPipeSourceAssignerDisruptorRingBufferSize(
         Integer.parseInt(
             Optional.ofNullable(

iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/task/PipeSleepIntervalTest.java

@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.commons.pipe.task;
+
+import org.apache.iotdb.commons.conf.CommonConfig;
+import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.commons.exception.pipe.PipeRuntimeException;
+import org.apache.iotdb.commons.pipe.agent.task.subtask.PipeAbstractSinkSubtask;
+import org.apache.iotdb.commons.pipe.config.PipeConfig;
+import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+public class PipeSleepIntervalTest {
+  private long oldPipeSinkSubtaskSleepIntervalInitMs;
+  private long oldPipeSinkSubtaskSleepIntervalMaxMs;
+
+  @Before
+  public void setUp() throws Exception {
+    final CommonConfig config = CommonDescriptor.getInstance().getConfig();
+    oldPipeSinkSubtaskSleepIntervalInitMs = config.getPipeSinkSubtaskSleepIntervalInitMs();
+    oldPipeSinkSubtaskSleepIntervalMaxMs = config.getPipeSinkSubtaskSleepIntervalMaxMs();
+    config.setPipeSinkSubtaskSleepIntervalInitMs(25L);
+    config.setPipeSinkSubtaskSleepIntervalMaxMs(50L);
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    final CommonConfig config = CommonDescriptor.getInstance().getConfig();
+    config.setPipeSinkSubtaskSleepIntervalInitMs(oldPipeSinkSubtaskSleepIntervalInitMs);
+    config.setPipeSinkSubtaskSleepIntervalMaxMs(oldPipeSinkSubtaskSleepIntervalMaxMs);
+  }
+
+  @Test
+  public void test() {
+    try (final PipeAbstractSinkSubtask subtask =
+        new PipeAbstractSinkSubtask(null, 0, null) {
+          @Override
+          protected String getRootCause(Throwable throwable) {
+            return null;
+          }
+
+          @Override
+          protected void report(EnrichedEvent event, PipeRuntimeException exception) {}
+
+          @Override
+          protected boolean executeOnce() {
+            return false;
+          }
+        }) {
+      long startTime = System.currentTimeMillis();
+      subtask.sleep4NonReportException();
+      Assert.assertTrue(
+          System.currentTimeMillis() - startTime
+              >= PipeConfig.getInstance().getPipeSinkSubtaskSleepIntervalInitMs());
+      startTime = System.currentTimeMillis() - startTime;
+      subtask.sleep4NonReportException();
+      Assert.assertTrue(
+          System.currentTimeMillis() - startTime
+              >= PipeConfig.getInstance().getPipeSinkSubtaskSleepIntervalInitMs());
+    }
+  }
+}