diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index f56589e52c0d..b5e4e558c440 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java @@ -20,6 +20,10 @@ package org.apache.iotdb.db.i18n; public final class DataNodePipeMessages { + public static final String + LOG_AIR_GAP_UDP_RECEIVE_BUFFER_IS_ARG_BYTES_BELOW_THE_REQUESTED_ARG_BYTES_INCREASE_THE_OPERATING_SYSTEM_RECEIVE_BUFFER_LIMIT_AND_RESTART_THE_DATANODE_TO_REDUCE_PACKET_LOSS_89B20AFC = + "Air gap UDP receive buffer is {} bytes, below the requested {} bytes. Increase the operating system receive-buffer limit and restart the DataNode to reduce packet loss."; + public static final String LOG_SUBSCRIPTION_REPLAY_STALLED_FOR_ARG_MS_ARG_DIAGNOSTICS_ARG_INSPECT_PREFETCH_WORKER_STACKS_AND_WAL_I_O_CB012F98 = "Subscription replay stalled for {} ms: {}; diagnostics={}. Inspect prefetch worker stacks and WAL I/O."; diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index 1b8b488f30f1..0e11ad0539b9 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java @@ -20,6 +20,10 @@ package org.apache.iotdb.db.i18n; public final class DataNodePipeMessages { + public static final String + LOG_AIR_GAP_UDP_RECEIVE_BUFFER_IS_ARG_BYTES_BELOW_THE_REQUESTED_ARG_BYTES_INCREASE_THE_OPERATING_SYSTEM_RECEIVE_BUFFER_LIMIT_AND_RESTART_THE_DATANODE_TO_REDUCE_PACKET_LOSS_89B20AFC = + "Air gap UDP 接收缓冲区实际为 {} 字节，小于请求的 {} 字节。请提高操作系统接收缓冲区上限并重启 DataNode，以减少丢包。"; + public static final String LOG_SUBSCRIPTION_REPLAY_STALLED_FOR_ARG_MS_ARG_DIAGNOSTICS_ARG_INSPECT_PREFETCH_WORKER_STACKS_AND_WAL_I_O_CB012F98 = "订阅回放已停滞 {} ms：{}；diagnostics={}。请检查预取工作线程栈和 WAL I/O。"; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/airgap/IoTDBAirGapReceiverAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/airgap/IoTDBAirGapReceiverAgent.java index da056fcc2c58..20f6aff7aa7d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/airgap/IoTDBAirGapReceiverAgent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/airgap/IoTDBAirGapReceiverAgent.java @@ -262,10 +262,22 @@ public void start() throws StartupException { try { serverSocket = new ServerSocket(PipeConfig.getInstance().getPipeAirGapReceiverPort()); datagramSocket = new DatagramSocket(PipeConfig.getInstance().getPipeAirGapReceiverPort()); - datagramSocket.setReceiveBufferSize( - PipeConfig.getInstance().getPipeAirGapReceiverUdpSocketBufferSizeInBytes()); + final int requestedBufferSize = + PipeConfig.getInstance().getPipeAirGapReceiverUdpSocketBufferSizeInBytes(); + datagramSocket.setReceiveBufferSize(requestedBufferSize); + final int actualBufferSize = datagramSocket.getReceiveBufferSize(); + if (actualBufferSize < requestedBufferSize) { + LOGGER.warn( + DataNodePipeMessages + .LOG_AIR_GAP_UDP_RECEIVE_BUFFER_IS_ARG_BYTES_BELOW_THE_REQUESTED_ARG_BYTES_INCREASE_THE_OPERATING_SYSTEM_RECEIVE_BUFFER_LIMIT_AND_RESTART_THE_DATANODE_TO_REDUCE_PACKET_LOSS_89B20AFC, + actualBufferSize, + requestedBufferSize); + } datagramSocket.setSoTimeout(UDP_SESSION_CLEANUP_INTERVAL_MS); } catch (final IOException e) { + if (Objects.nonNull(datagramSocket)) { + datagramSocket.close(); + } if (Objects.nonNull(serverSocket)) { try { serverSocket.close(); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSinkTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSinkTest.java index e5b016942a12..f042c13aa43d 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSinkTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSinkTest.java @@ -61,6 +61,50 @@ import java.util.zip.CRC32; public class IoTDBDataRegionAirGapSinkTest { + @Test + public void testStrictUdpDefaultPacketSizeAndExplicitOverrides() throws Exception { + for (final String packetSizeKey : + Arrays.asList( + null, + PipeSinkConstant.CONNECTOR_AIR_GAP_UDP_PACKET_SIZE_KEY, + PipeSinkConstant.SINK_AIR_GAP_UDP_PACKET_SIZE_KEY)) { + try (final DatagramSocket receiverSocket = new DatagramSocket(0); + final UdpTestingIoTDBDataRegionAirGapSink sink = + new UdpTestingIoTDBDataRegionAirGapSink()) { + final PipeParameters parameters = buildParameters(false); + parameters.getAttribute().put(PipeSinkConstant.SINK_AIR_GAP_TRANSPORT_KEY, "udp"); + parameters + .getAttribute() + .put(PipeSinkConstant.SINK_AIR_GAP_STRICT_UNIDIRECTIONAL_KEY, "true"); + if (packetSizeKey != null) { + parameters.getAttribute().put(packetSizeKey, "4096"); + } + sink.validate(new PipeParameterValidator(parameters)); + sink.customize( + parameters, + new PipeTaskRuntimeConfiguration(new PipeTaskSinkRuntimeEnvironment("pipe", 1L, 1))); + + final byte[] request = new byte[2500]; + request[0] = 1; + request[2] = (byte) PipeRequestType.TRANSFER_TABLET_RAW.getType(); + Assert.assertTrue(sink.sendUdp(receiverSocket.getLocalPort(), request)); + receiverSocket.setSoTimeout(1000); + final int packetCount = packetSizeKey == null ? 2 : 1; + for (int i = 0; i < packetCount; i++) { + final byte[] buffer = new byte[4097]; + final DatagramPacket packet = new DatagramPacket(buffer, buffer.length); + receiverSocket.receive(packet); + Assert.assertTrue(packet.getLength() <= (packetSizeKey == null ? 1400 : 4096)); + Assert.assertEquals( + packetSizeKey == null + ? PipeRequestType.TRANSFER_SLICE.getType() + : PipeRequestType.TRANSFER_TABLET_RAW.getType(), + decodeAirGapDatagram(buffer, packet.getLength()).type); + } + } + } + } + @Test public void testStrictUdpSendIntervalIsValidatedAndAppliedToEverySlice() throws Exception { for (final String key : @@ -266,9 +310,20 @@ public void testUdpDefaultAndExplicitFalseRequireAcknowledgement() throws Except sink.customize( parameters, new PipeTaskRuntimeConfiguration(new PipeTaskSinkRuntimeEnvironment("pipe", 1L, 1))); + final byte[] request = new byte[2500]; + request[0] = 1; + request[2] = (byte) PipeRequestType.TRANSFER_TABLET_RAW.getType(); Assert.assertThrows( SocketTimeoutException.class, - () -> sink.sendUdp(receiverSocket.getLocalPort(), new byte[] {1, 0, 11})); + () -> sink.sendUdp(receiverSocket.getLocalPort(), request)); + receiverSocket.setSoTimeout(1000); + final byte[] buffer = new byte[3000]; + final DatagramPacket packet = new DatagramPacket(buffer, buffer.length); + receiverSocket.receive(packet); + Assert.assertEquals(request.length + 3 * Long.BYTES, packet.getLength()); + Assert.assertEquals( + PipeRequestType.TRANSFER_TABLET_RAW.getType(), + decodeAirGapDatagram(buffer, packet.getLength()).type); } } } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSinkConstant.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSinkConstant.java index d761d4008300..98dffc5265cd 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSinkConstant.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSinkConstant.java @@ -209,6 +209,8 @@ private static String getDefaultConnectorOrSinkName(final PipeParameters paramet public static final String SINK_AIR_GAP_UDP_PACKET_SIZE_KEY = "sink.air-gap.udp.packet-size-bytes"; public static final int CONNECTOR_AIR_GAP_UDP_PACKET_SIZE_DEFAULT_VALUE = 60 * 1024; + // Leave room for IP/UDP headers and common encapsulation on a 1500-byte MTU link. + public static final int CONNECTOR_AIR_GAP_STRICT_UDP_PACKET_SIZE_DEFAULT_VALUE = 1400; public static final int CONNECTOR_AIR_GAP_UDP_PACKET_SIZE_MIN_VALUE = 1024; public static final int CONNECTOR_AIR_GAP_UDP_PACKET_SIZE_MAX_VALUE = 65_507; public static final String CONNECTOR_AIR_GAP_UDP_SEND_INTERVAL_MS_KEY = diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBAirGapSink.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBAirGapSink.java index 1778c864e687..ed9545a2ded0 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBAirGapSink.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBAirGapSink.java @@ -71,6 +71,7 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_AIR_GAP_E_LANGUAGE_ENABLE_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_AIR_GAP_HANDSHAKE_TIMEOUT_MS_DEFAULT_VALUE; import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_AIR_GAP_HANDSHAKE_TIMEOUT_MS_KEY; +import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_AIR_GAP_STRICT_UDP_PACKET_SIZE_DEFAULT_VALUE; import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_AIR_GAP_STRICT_UNIDIRECTIONAL_DEFAULT_VALUE; import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_AIR_GAP_STRICT_UNIDIRECTIONAL_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_AIR_GAP_TRANSPORT_DEFAULT_VALUE; @@ -327,7 +328,9 @@ public void customize( udpPacketSizeInBytes = parameters.getIntOrDefault( Arrays.asList(CONNECTOR_AIR_GAP_UDP_PACKET_SIZE_KEY, SINK_AIR_GAP_UDP_PACKET_SIZE_KEY), - CONNECTOR_AIR_GAP_UDP_PACKET_SIZE_DEFAULT_VALUE); + strictUnidirectional + ? CONNECTOR_AIR_GAP_STRICT_UDP_PACKET_SIZE_DEFAULT_VALUE + : CONNECTOR_AIR_GAP_UDP_PACKET_SIZE_DEFAULT_VALUE); strictUdpSendIntervalNanos = TimeUnit.MILLISECONDS.toNanos( parameters.getIntOrDefault(