This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch codex/udp-loss-tuning-apache in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 69af183656d9e05ffa80998964a6ec28b8d73d34 Author: Caideyipi <[email protected]> AuthorDate: Mon Nov 2 16:55:19 2026 +0800 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. --- .../apache/iotdb/db/i18n/DataNodePipeMessages.java | 4 ++ .../apache/iotdb/db/i18n/DataNodePipeMessages.java | 4 ++ .../protocol/airgap/IoTDBAirGapReceiverAgent.java | 16 +++++- .../airgap/IoTDBDataRegionAirGapSinkTest.java | 57 +++++++++++++++++++++- .../pipe/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 f56589e52c0..b5e4e558c44 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 1b8b488f30f..0e11ad0539b 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 da056fcc2c5..20f6aff7aa7 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 class IoTDBAirGapReceiverAgent implements IService { 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 e5b016942a1..f042c13aa43 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.concurrent.atomic.AtomicReference; 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 class IoTDBDataRegionAirGapSinkTest { 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 d761d400830..98dffc5265c 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 @@ public class PipeSinkConstant { 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 1778c864e68..ed9545a2ded 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.CON 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 abstract class IoTDBAirGapSink extends IoTDBSink { 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(
