From 69af183656d9e05ffa80998964a6ec28b8d73d34 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 2 Nov 2026 16:55:19 +0800 Subject: [PATCH] Use smaller default datagrams for strictly unidirectional air-gap UDP Default strictly unidirectional UDP packets to 1400 bytes while preserving explicit overrides and the acknowledged transport default. Warn when the operating system caps the receiver socket buffer, and cover defaults and overrides with socket tests. --- .../iotdb/db/i18n/DataNodePipeMessages.java | 4 ++ .../iotdb/db/i18n/DataNodePipeMessages.java | 4 ++ .../airgap/IoTDBAirGapReceiverAgent.java | 16 +++++- .../airgap/IoTDBDataRegionAirGapSinkTest.java | 57 ++++++++++++++++++- .../config/constant/PipeSinkConstant.java | 2 + .../pipe/sink/protocol/IoTDBAirGapSink.java | 5 +- 6 files changed, 84 insertions(+), 4 deletions(-) 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(