Pipe: Configured the air gap timeout to avoid packet loss (#17231) (#17253)

* good-game

* ifx

* fxi

* xif

Author: Caideyipi

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

@@ -95,6 +95,9 @@ public void transfer(final TabletInsertionEvent tabletInsertionEvent) throws Exc
     final AirGapSocket socket = sockets.get(socketIndex);
 
     try {
+      // When receiver encountered packet loss, the transfer will time out
+      // We need to restore the transfer quickly by retry under this circumstance
+      socket.setSoTimeout(PIPE_CONFIG.getPipeAirGapSinkTabletTimeoutMs());
       if (tabletInsertionEvent instanceof PipeInsertNodeTabletInsertionEvent) {
         doTransferWrapper(socket, (PipeInsertNodeTabletInsertionEvent) tabletInsertionEvent);
       } else {
@@ -108,6 +111,8 @@ public void transfer(final TabletInsertionEvent tabletInsertionEvent) throws Exc
               "Network error when transfer tablet insertion event %s, because %s.",
               ((EnrichedEvent) tabletInsertionEvent).coreReportMessage(), e.getMessage()),
           e);
+    } finally {
+      socket.setSoTimeout(PIPE_CONFIG.getPipeSinkTransferTimeoutMs());
     }
   }
 

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

@@ -266,6 +266,7 @@ public class CommonConfig {
   private long pipeSourceMatcherCacheSize = 1024;
 
   private int pipeSinkHandshakeTimeoutMs = 10 * 1000; // 10 seconds
+  private int pipeAirGapSinkTabletTimeoutMs = 60 * 1000; // 1 min
   private int pipeSinkTransferTimeoutMs = 15 * 60 * 1000; // 15 minutes
   private int pipeSinkReadFileBufferSize = 5242880; // 5MB
   private boolean isPipeSinkReadFileBufferMemoryControlEnabled = false;
@@ -1019,6 +1020,26 @@ public void setPipeSinkHandshakeTimeoutMs(long pipeSinkHandshakeTimeoutMs) {
     }
   }
 
+  public int getPipeAirGapSinkTabletTimeoutMs() {
+    return pipeAirGapSinkTabletTimeoutMs;
+  }
+
+  public void setPipeAirGapSinkTabletTimeoutMs(long pipeAirGapSinkTabletTimeoutMs) {
+    final int fPipeAirGapSinkTabletTimeoutMs = this.pipeAirGapSinkTabletTimeoutMs;
+    try {
+      this.pipeAirGapSinkTabletTimeoutMs = Math.toIntExact(pipeAirGapSinkTabletTimeoutMs);
+    } catch (ArithmeticException e) {
+      this.pipeAirGapSinkTabletTimeoutMs = Integer.MAX_VALUE;
+      logger.warn(
+          "Given pipe air gap sink tablet timeout is too large, set to {} ms.", Integer.MAX_VALUE);
+    } finally {
+      if (fPipeAirGapSinkTabletTimeoutMs != this.pipeAirGapSinkTabletTimeoutMs) {
+        logger.info(
+            "pipeAirGapSinkTabletTimeoutMs is set to {}.", this.pipeAirGapSinkTabletTimeoutMs);
+      }
+    }
+  }
+
   public int getPipeSinkTransferTimeoutMs() {
     return pipeSinkTransferTimeoutMs;
   }

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

@@ -179,6 +179,10 @@ public int getPipeSinkHandshakeTimeoutMs() {
     return COMMON_CONFIG.getPipeSinkHandshakeTimeoutMs();
   }
 
+  public int getPipeAirGapSinkTabletTimeoutMs() {
+    return COMMON_CONFIG.getPipeAirGapSinkTabletTimeoutMs();
+  }
+
   public int getPipeSinkTransferTimeoutMs() {
     return COMMON_CONFIG.getPipeSinkTransferTimeoutMs();
   }

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

@@ -360,6 +360,13 @@ public static void loadPipeInternalConfig(CommonConfig config, TrimProperties pr
                     properties.getProperty(
                         "pipe_connector_handshake_timeout_ms",
                         String.valueOf(config.getPipeSinkHandshakeTimeoutMs())))));
+    config.setPipeAirGapSinkTabletTimeoutMs(
+        Long.parseLong(
+            Optional.ofNullable(properties.getProperty("pipe_air_gap_sink_tablet_timeout_ms"))
+                .orElse(
+                    properties.getProperty(
+                        "pipe_air_gap_connector_tablet_timeout_ms",
+                        String.valueOf(config.getPipeAirGapSinkTabletTimeoutMs())))));
     config.setPipeSinkReadFileBufferSize(
         Integer.parseInt(
             Optional.ofNullable(properties.getProperty("pipe_sink_read_file_buffer_size"))