This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch feat/air-gap-strict-unidirectional in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit f03440874c1b404808b4f512795d64c092bea9f1 Author: Caideyipi <[email protected]> AuthorDate: Fri Oct 30 15:28:38 2026 +0800 Support strictly unidirectional UDP air-gap transfer --- .../protocol/airgap/IoTDBAirGapReceiver.java | 4 +- .../protocol/airgap/IoTDBAirGapReceiverTest.java | 56 +++++++++ .../airgap/IoTDBDataRegionAirGapSinkTest.java | 137 +++++++++++++++++++++ .../apache/iotdb/commons/i18n/PipeMessages.java | 6 + .../apache/iotdb/commons/i18n/PipeMessages.java | 6 + .../pipe/config/constant/PipeSinkConstant.java | 6 + .../sink/payload/airgap/AirGapUdpProtocol.java | 29 +++++ .../pipe/sink/protocol/IoTDBAirGapSink.java | 64 +++++++++- 8 files changed, 303 insertions(+), 5 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/airgap/IoTDBAirGapReceiver.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/airgap/IoTDBAirGapReceiver.java index e3413b009b2..1a9cb1616d0 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/airgap/IoTDBAirGapReceiver.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/airgap/IoTDBAirGapReceiver.java @@ -25,6 +25,7 @@ import org.apache.iotdb.commons.pipe.config.PipeConfig; import org.apache.iotdb.commons.pipe.sink.payload.airgap.AirGapELanguageConstant; import org.apache.iotdb.commons.pipe.sink.payload.airgap.AirGapOneByteResponse; import org.apache.iotdb.commons.pipe.sink.payload.airgap.AirGapPseudoTPipeTransferRequest; +import org.apache.iotdb.commons.pipe.sink.payload.airgap.AirGapUdpProtocol; import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent; import org.apache.iotdb.db.pipe.receiver.protocol.thrift.IoTDBDataNodeReceiverAgent; @@ -296,7 +297,8 @@ public class IoTDBAirGapReceiver extends WrappedRunnable { private void respond(final byte[] response) throws IOException { if (currentDatagramSocket != null && currentRemoteSocketAddress != null) { - if (currentDatagramRequestId == null) { + if (currentDatagramRequestId == null + || (currentDatagramRequestId & AirGapUdpProtocol.NO_ACK_REQUEST_ID_MASK) != 0) { return; } final byte[] responseBytes = diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/airgap/IoTDBAirGapReceiverTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/airgap/IoTDBAirGapReceiverTest.java index 57cacb38799..86481ce3517 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/airgap/IoTDBAirGapReceiverTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/airgap/IoTDBAirGapReceiverTest.java @@ -27,6 +27,7 @@ import org.apache.iotdb.commons.pipe.receiver.runtime.PipeReceiverRuntimeRegistr import org.apache.iotdb.commons.pipe.sink.payload.airgap.AirGapELanguageConstant; import org.apache.iotdb.commons.pipe.sink.payload.airgap.AirGapOneByteResponse; import org.apache.iotdb.commons.pipe.sink.payload.airgap.AirGapPseudoTPipeTransferRequest; +import org.apache.iotdb.commons.pipe.sink.payload.airgap.AirGapUdpProtocol; import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.IoTDBSinkRequestVersion; import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent; import org.apache.iotdb.db.pipe.receiver.protocol.thrift.IoTDBDataNodeReceiverAgent; @@ -202,6 +203,61 @@ public class IoTDBAirGapReceiverTest { } } + @Test + public void testStrictUnidirectionalUdpNeverRespondsToSuccessOrFailure() throws Exception { + for (final TSStatusCode status : + new TSStatusCode[] { + TSStatusCode.SUCCESS_STATUS, TSStatusCode.PIPE_RECEIVER_USER_CONFLICT_EXCEPTION + }) { + try (final DatagramSocket receiverSocket = new NoResponseDatagramSocket()) { + final byte[] request = + toUdpDatagram(AirGapUdpProtocol.NO_ACK_REQUEST_ID_MASK | 73L, toRawRequestBytes()); + final DatagramPacket packet = + new DatagramPacket(request, request.length, InetAddress.getLoopbackAddress(), 12345); + final IoTDBAirGapReceiver receiver = new IoTDBAirGapReceiver(new RecordingSocket(), 5L); + final StubIoTDBDataNodeReceiverAgent stubAgent = new StubIoTDBDataNodeReceiverAgent(); + final AtomicInteger receivedRequests = new AtomicInteger(); + stubAgent.setStubReceiver( + "udp-client", + new StubReceiver(new TSStatus(status.getStatusCode())) { + @Override + public TPipeTransferResp receive(final TPipeTransferReq req) { + receivedRequests.incrementAndGet(); + return super.receive(req); + } + }); + setField(receiver, "agent", stubAgent); + receiver.receiveUdp(receiverSocket, packet, "udp-client", request, () -> {}); + Assert.assertEquals(1, receivedRequests.get()); + } + } + } + + @Test + public void testStrictUnidirectionalUdpDoesNotRespondToCorruptRequest() throws Exception { + try (final DatagramSocket receiverSocket = new NoResponseDatagramSocket()) { + final byte[] request = + toUdpDatagram(AirGapUdpProtocol.NO_ACK_REQUEST_ID_MASK | 74L, toRawRequestBytes()); + request[request.length - 1] ^= 1; + final DatagramPacket packet = + new DatagramPacket(request, request.length, InetAddress.getLoopbackAddress(), 12345); + final AtomicBoolean registered = new AtomicBoolean(); + new IoTDBAirGapReceiver(new RecordingSocket(), 6L) + .receiveUdp(receiverSocket, packet, "udp-client", request, () -> registered.set(true)); + Assert.assertFalse(registered.get()); + } + } + + private static class NoResponseDatagramSocket extends DatagramSocket { + + private NoResponseDatagramSocket() throws SocketException {} + + @Override + public void send(final DatagramPacket packet) { + Assert.fail("Strictly unidirectional requests must never send an ACK or NACK"); + } + } + @Test public void testTemporaryUnavailableRetryUsesFreshRequestBody() throws Exception { assertTemporaryUnavailableRetryUsesFreshRequestBody(null); 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 82000564cfe..cdc0a300067 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 @@ -24,6 +24,7 @@ import org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant; import org.apache.iotdb.commons.pipe.config.plugin.configuraion.PipeTaskRuntimeConfiguration; import org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskSinkRuntimeEnvironment; import org.apache.iotdb.commons.pipe.sink.payload.airgap.AirGapOneByteResponse; +import org.apache.iotdb.commons.pipe.sink.payload.airgap.AirGapUdpProtocol; import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeRequestType; import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent; import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent; @@ -32,6 +33,7 @@ import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTsFil import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator; import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters; import org.apache.iotdb.pipe.api.exception.PipeConnectionException; +import org.apache.iotdb.pipe.api.exception.PipeParameterNotValidException; import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq; import org.apache.tsfile.enums.TSDataType; @@ -60,6 +62,137 @@ import java.util.zip.CRC32; public class IoTDBDataRegionAirGapSinkTest { + @Test + public void testStrictUnidirectionalRequiresUdpAndBooleanValue() throws Exception { + try (final IoTDBDataRegionAirGapSink sink = new IoTDBDataRegionAirGapSink()) { + for (final String key : + Arrays.asList( + PipeSinkConstant.CONNECTOR_AIR_GAP_STRICT_UNIDIRECTIONAL_KEY, + PipeSinkConstant.SINK_AIR_GAP_STRICT_UNIDIRECTIONAL_KEY)) { + final PipeParameters parameters = buildParameters(false); + parameters.getAttribute().put(key, "true"); + Assert.assertThrows( + PipeParameterNotValidException.class, + () -> sink.validate(new PipeParameterValidator(parameters))); + parameters.getAttribute().put(PipeSinkConstant.CONNECTOR_AIR_GAP_TRANSPORT_KEY, "udp"); + sink.validate(new PipeParameterValidator(parameters)); + parameters.getAttribute().put(key, "invalid"); + Assert.assertThrows( + PipeParameterNotValidException.class, + () -> sink.validate(new PipeParameterValidator(parameters))); + } + } + } + + @Test + public void testStrictUnidirectionalUdpSendsAllSlicesWithoutAcknowledgements() throws Exception { + for (final String key : + Arrays.asList( + PipeSinkConstant.CONNECTOR_AIR_GAP_STRICT_UNIDIRECTIONAL_KEY, + PipeSinkConstant.SINK_AIR_GAP_STRICT_UNIDIRECTIONAL_KEY)) { + try (final DatagramSocket receiverSocket = new DatagramSocket(0); + final UdpTestingIoTDBDataRegionAirGapSink sink = + new UdpTestingIoTDBDataRegionAirGapSink()) { + final PipeParameters parameters = buildParameters(false); + parameters.getAttribute().put(PipeSinkConstant.CONNECTOR_AIR_GAP_TRANSPORT_KEY, "udp"); + parameters + .getAttribute() + .put(PipeSinkConstant.CONNECTOR_AIR_GAP_UDP_PACKET_SIZE_KEY, "1024"); + parameters.getAttribute().put(key, "true"); + 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(); + // Nothing receives or acknowledges until sendUdp has returned. + Assert.assertTrue(sink.sendUdp(receiverSocket.getLocalPort(), request)); + + receiverSocket.setSoTimeout(1000); + for (int i = 0; i < 3; i++) { + final byte[] buffer = new byte[1025]; + final DatagramPacket packet = new DatagramPacket(buffer, buffer.length); + receiverSocket.receive(packet); + Assert.assertTrue(packet.getLength() <= 1024); + final long requestId = ByteBuffer.wrap(buffer).getLong(); + Assert.assertEquals(AirGapUdpProtocol.NO_ACK_REQUEST_ID_MASK, requestId & Long.MIN_VALUE); + Assert.assertEquals(i + 1, requestId & Long.MAX_VALUE); + Assert.assertEquals( + PipeRequestType.TRANSFER_SLICE.getType(), + decodeAirGapDatagram(buffer, packet.getLength()).type); + } + } + } + } + + @Test + public void testStrictUnidirectionalHandshakeIsRefreshedBeforeEachEvent() throws Exception { + try (final DatagramSocket receiverSocket = new DatagramSocket(0); + final UdpTestingIoTDBDataRegionAirGapSink sink = + new UdpTestingIoTDBDataRegionAirGapSink()) { + final PipeParameters parameters = buildParameters(false); + parameters.getAttribute().put(PipeSinkConstant.CONNECTOR_AIR_GAP_TRANSPORT_KEY, "udp"); + parameters + .getAttribute() + .put(PipeSinkConstant.SINK_AIR_GAP_STRICT_UNIDIRECTIONAL_KEY, "true"); + parameters + .getAttribute() + .put( + PipeSinkConstant.CONNECTOR_IOTDB_NODE_URLS_KEY, + "127.0.0.1:" + receiverSocket.getLocalPort()); + sink.validate(new PipeParameterValidator(parameters)); + sink.customize( + parameters, + new PipeTaskRuntimeConfiguration(new PipeTaskSinkRuntimeEnvironment("pipe", 1L, 1))); + sink.handshake(); + final byte[] request = new byte[] {1, 0, 11}; + sink.sendEvent(request); + sink.sendEvent(request); + + receiverSocket.setSoTimeout(1000); + for (final short type : + new short[] { + PipeRequestType.HANDSHAKE_DATANODE_V2.getType(), + PipeRequestType.HANDSHAKE_DATANODE_V2.getType(), + 11, + PipeRequestType.HANDSHAKE_DATANODE_V2.getType(), + 11 + }) { + final byte[] buffer = new byte[65_507]; + final DatagramPacket packet = new DatagramPacket(buffer, buffer.length); + receiverSocket.receive(packet); + Assert.assertEquals(type, decodeAirGapDatagram(buffer, packet.getLength()).type); + Assert.assertTrue(ByteBuffer.wrap(buffer).getLong() < 0); + } + } + } + + @Test + public void testUdpDefaultAndExplicitFalseRequireAcknowledgement() throws Exception { + for (final String strictValue : Arrays.asList(null, "false")) { + try (final DatagramSocket receiverSocket = new DatagramSocket(0); + final UdpTestingIoTDBDataRegionAirGapSink sink = + new UdpTestingIoTDBDataRegionAirGapSink()) { + final PipeParameters parameters = buildParameters(false); + parameters.getAttribute().put(PipeSinkConstant.CONNECTOR_AIR_GAP_TRANSPORT_KEY, "udp"); + if (strictValue != null) { + parameters + .getAttribute() + .put(PipeSinkConstant.SINK_AIR_GAP_STRICT_UNIDIRECTIONAL_KEY, strictValue); + } + sink.validate(new PipeParameterValidator(parameters)); + sink.customize( + parameters, + new PipeTaskRuntimeConfiguration(new PipeTaskSinkRuntimeEnvironment("pipe", 1L, 1))); + Assert.assertThrows( + SocketTimeoutException.class, + () -> sink.sendUdp(receiverSocket.getLocalPort(), new byte[] {1, 0, 11})); + } + } + } + @Test public void testCompletionBarrierRetriesUnacknowledgedBatch() throws Exception { try (final RecordingIoTDBDataRegionAirGapSink sink = new RecordingIoTDBDataRegionAirGapSink()) { @@ -455,6 +588,10 @@ public class IoTDBDataRegionAirGapSinkTest { private static class UdpTestingIoTDBDataRegionAirGapSink extends IoTDBDataRegionAirGapSink { + private void sendEvent(final byte[] request) throws IOException { + Assert.assertTrue(sendBytes(sockets.get(nextSocketIndex()), request)); + } + private boolean sendUdp(final int port, final byte[] request) throws Exception { final AirGapSocket socket = new AirGapSocket("127.0.0.1", port); try (final Socket ignored = socket) { diff --git a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java index 437e4dd68de..7cdf30ca6e6 100644 --- a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java @@ -20,6 +20,12 @@ package org.apache.iotdb.commons.i18n; public final class PipeMessages { + public static final String EXCEPTION_STRICTLY_UNIDIRECTIONAL_AIR_GAP_TRANSPORT_REQUIRES_UDP_78921AF9 = + "Strictly unidirectional air-gap transport requires UDP."; + public static final String LOG_STRICTLY_UNIDIRECTIONAL_UDP_TRANSPORT_IS_ENABLED_REQUESTS_ARE_SENT_WITHOUT_ACKNOWLEDGEMENTS_DELIVERY_AND_RECEIVER_ERRORS_CANNOT_BE_CONFIRMED_0A2F4382 = + "Strictly unidirectional UDP transport is enabled. Requests are sent without acknowledgements; delivery and receiver errors cannot be confirmed."; + public static final String LOG_AIR_GAP_HANDSHAKE_REQUEST_SENT_ON_SOCKET_ARG_WITHOUT_WAITING_FOR_ACKNOWLEDGEMENT_E1ECD582 = + "Air gap handshake request sent on socket {} without waiting for acknowledgement."; // ===================== Common / Shared ===================== diff --git a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java index f67f0a351ac..7b57a22fa37 100644 --- a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java @@ -20,6 +20,12 @@ package org.apache.iotdb.commons.i18n; public final class PipeMessages { + public static final String EXCEPTION_STRICTLY_UNIDIRECTIONAL_AIR_GAP_TRANSPORT_REQUIRES_UDP_78921AF9 = + "严格单向的 air-gap 传输必须使用 UDP。"; + public static final String LOG_STRICTLY_UNIDIRECTIONAL_UDP_TRANSPORT_IS_ENABLED_REQUESTS_ARE_SENT_WITHOUT_ACKNOWLEDGEMENTS_DELIVERY_AND_RECEIVER_ERRORS_CANNOT_BE_CONFIRMED_0A2F4382 = + "已启用严格单向 UDP 传输。请求发送后不等待确认；无法确认送达情况和接收端错误。"; + public static final String LOG_AIR_GAP_HANDSHAKE_REQUEST_SENT_ON_SOCKET_ARG_WITHOUT_WAITING_FOR_ACKNOWLEDGEMENT_E1ECD582 = + "已通过 socket {} 发送 air-gap 握手请求，不等待确认。"; // ===================== Common / Shared ===================== 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 e7a36577a06..c04f6b87c79 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 @@ -198,6 +198,12 @@ public class PipeSinkConstant { Arrays.asList( CONNECTOR_AIR_GAP_TRANSPORT_TCP_VALUE, CONNECTOR_AIR_GAP_TRANSPORT_UDP_VALUE))); + public static final String CONNECTOR_AIR_GAP_STRICT_UNIDIRECTIONAL_KEY = + "connector.air-gap.strict-unidirectional"; + public static final String SINK_AIR_GAP_STRICT_UNIDIRECTIONAL_KEY = + "sink.air-gap.strict-unidirectional"; + public static final boolean CONNECTOR_AIR_GAP_STRICT_UNIDIRECTIONAL_DEFAULT_VALUE = false; + public static final String CONNECTOR_AIR_GAP_UDP_PACKET_SIZE_KEY = "connector.air-gap.udp.packet-size-bytes"; public static final String SINK_AIR_GAP_UDP_PACKET_SIZE_KEY = diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/payload/airgap/AirGapUdpProtocol.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/payload/airgap/AirGapUdpProtocol.java new file mode 100644 index 00000000000..783c6af94ff --- /dev/null +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/payload/airgap/AirGapUdpProtocol.java @@ -0,0 +1,29 @@ +/* + * 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.sink.payload.airgap; + +public final class AirGapUdpProtocol { + + // Reserve the request ID's sign bit for requests that must not produce any reverse traffic. + // The remaining 63 bits identify the request; acknowledged requests keep their existing envelope. + public static final long NO_ACK_REQUEST_ID_MASK = Long.MIN_VALUE; + + private AirGapUdpProtocol() {} +} 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 bdffcfe4734..cd3c151d36f 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 @@ -26,6 +26,7 @@ import org.apache.iotdb.commons.pipe.config.PipeConfig; import org.apache.iotdb.commons.pipe.resource.log.PipeLogger; import org.apache.iotdb.commons.pipe.sink.payload.airgap.AirGapELanguageConstant; import org.apache.iotdb.commons.pipe.sink.payload.airgap.AirGapOneByteResponse; +import org.apache.iotdb.commons.pipe.sink.payload.airgap.AirGapUdpProtocol; import org.apache.iotdb.commons.pipe.sink.payload.thrift.common.PipeTransferSliceReqBuilder; import org.apache.iotdb.commons.utils.ErrorHandlingCommonUtils; import org.apache.iotdb.pipe.api.annotation.TableModel; @@ -69,6 +70,8 @@ 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_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; import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_AIR_GAP_TRANSPORT_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_AIR_GAP_TRANSPORT_SET; @@ -82,6 +85,7 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CON import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_LOAD_BALANCE_ROUND_ROBIN_STRATEGY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_AIR_GAP_E_LANGUAGE_ENABLE_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_AIR_GAP_HANDSHAKE_TIMEOUT_MS_KEY; +import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_AIR_GAP_STRICT_UNIDIRECTIONAL_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_AIR_GAP_TRANSPORT_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_AIR_GAP_UDP_PACKET_SIZE_KEY; import static org.apache.iotdb.commons.utils.BasicStructureSerDeUtil.LONG_LEN; @@ -182,6 +186,7 @@ public abstract class IoTDBAirGapSink extends IoTDBSink { private boolean eLanguageEnable; private boolean useUdpTransport; + private boolean strictUnidirectional; private int udpPacketSizeInBytes; // The air gap connector does not use clientManager thus we put handshake type here @@ -209,6 +214,21 @@ public abstract class IoTDBAirGapSink extends IoTDBSink { CONNECTOR_AIR_GAP_TRANSPORT_SET, airGapTransport), airGapTransport); + validator + .validateAttributeValueRange( + CONNECTOR_AIR_GAP_STRICT_UNIDIRECTIONAL_KEY, true, "true", "false") + .validateAttributeValueRange(SINK_AIR_GAP_STRICT_UNIDIRECTIONAL_KEY, true, "true", "false") + .validate( + arg -> + !parameters.getBooleanOrDefault( + Arrays.asList( + CONNECTOR_AIR_GAP_STRICT_UNIDIRECTIONAL_KEY, + SINK_AIR_GAP_STRICT_UNIDIRECTIONAL_KEY), + CONNECTOR_AIR_GAP_STRICT_UNIDIRECTIONAL_DEFAULT_VALUE) + || CONNECTOR_AIR_GAP_TRANSPORT_UDP_VALUE.equals(airGapTransport), + PipeMessages.EXCEPTION_STRICTLY_UNIDIRECTIONAL_AIR_GAP_TRANSPORT_REQUIRES_UDP_78921AF9, + airGapTransport); + final int packetSize = parameters.getIntOrDefault( Arrays.asList(CONNECTOR_AIR_GAP_UDP_PACKET_SIZE_KEY, SINK_AIR_GAP_UDP_PACKET_SIZE_KEY), @@ -273,6 +293,17 @@ public abstract class IoTDBAirGapSink extends IoTDBSink { CONNECTOR_AIR_GAP_TRANSPORT_DEFAULT_VALUE) .trim() .toLowerCase()); + strictUnidirectional = + parameters.getBooleanOrDefault( + Arrays.asList( + CONNECTOR_AIR_GAP_STRICT_UNIDIRECTIONAL_KEY, + SINK_AIR_GAP_STRICT_UNIDIRECTIONAL_KEY), + CONNECTOR_AIR_GAP_STRICT_UNIDIRECTIONAL_DEFAULT_VALUE); + if (strictUnidirectional) { + LOGGER.warn( + PipeMessages + .LOG_STRICTLY_UNIDIRECTIONAL_UDP_TRANSPORT_IS_ENABLED_REQUESTS_ARE_SENT_WITHOUT_ACKNOWLEDGEMENTS_DELIVERY_AND_RECEIVER_ERRORS_CANNOT_BE_CONFIRMED_0A2F4382); + } udpPacketSizeInBytes = parameters.getIntOrDefault( Arrays.asList(CONNECTOR_AIR_GAP_UDP_PACKET_SIZE_KEY, SINK_AIR_GAP_UDP_PACKET_SIZE_KEY), @@ -370,7 +401,14 @@ public abstract class IoTDBAirGapSink extends IoTDBSink { supportModsIfIsDataNodeReceiver = true; } socket.setSoTimeout(PIPE_CONFIG.getPipeSinkTransferTimeoutMs()); - LOGGER.info(PipeMessages.HANDSHAKE_SUCCESS_SOCKET, socket); + if (strictUnidirectional) { + LOGGER.debug( + PipeMessages + .LOG_AIR_GAP_HANDSHAKE_REQUEST_SENT_ON_SOCKET_ARG_WITHOUT_WAITING_FOR_ACKNOWLEDGEMENT_E1ECD582, + socket); + } else { + LOGGER.info(PipeMessages.HANDSHAKE_SUCCESS_SOCKET, socket); + } } protected abstract byte[] generateHandShakeV1Payload() throws IOException; @@ -444,8 +482,19 @@ public abstract class IoTDBAirGapSink extends IoTDBSink { protected abstract byte[] getTransferMultiFilePieceBytes( final String fileName, final long position, final byte[] payLoad) throws IOException; - protected int nextSocketIndex() { - return loadBalancer.nextSocketIndex(); + protected int nextSocketIndex() throws IOException { + final int socketIndex = loadBalancer.nextSocketIndex(); + if (strictUnidirectional) { + // There is no feedback when a receiver restarts or its idle session expires. Re-establish + // the session before each event, never between the pieces/slices of the same file/request. + try { + sendHandshakeReq(sockets.get(socketIndex)); + } catch (final IOException e) { + isSocketAlive.set(socketIndex, false); + throw e; + } + } + return socketIndex; } protected boolean send( @@ -488,7 +537,9 @@ public abstract class IoTDBAirGapSink extends IoTDBSink { private boolean sendOneDatagram(final AirGapSocket socket, final TPipeTransferReq request) throws IOException { - final long requestId = udpRequestIdGenerator.incrementAndGet(); + final long requestId = + (udpRequestIdGenerator.incrementAndGet() & Long.MAX_VALUE) + | (strictUnidirectional ? AirGapUdpProtocol.NO_ACK_REQUEST_ID_MASK : 0); final ByteBuffer body = request.body.duplicate(); final int requestSize = Byte.BYTES + Short.BYTES + body.remaining(); final int eLanguageSize = @@ -536,6 +587,11 @@ public abstract class IoTDBAirGapSink extends IoTDBSink { socket.getDatagramAddress(), socket.getEndPoint().getPort())); + if (strictUnidirectional) { + // This only confirms a local send. A one-way link cannot report delivery or remote errors. + return true; + } + final int originalTimeoutMs = datagramSocket.getSoTimeout(); final long deadlineNanos = originalTimeoutMs == 0
