This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new a6db44ff877 [Pipe] Split OPC UA sink requests by advertised server
limits (#18486) (#18510)
a6db44ff877 is described below
commit a6db44ff877c878d79639d3cdf1b9af54f7070f8
Author: Caideyipi <[email protected]>
AuthorDate: Mon Aug 24 14:53:10 2026 +0800
[Pipe] Split OPC UA sink requests by advertised server limits (#18486)
(#18510)
(cherry picked from commit 91a44fd8c009a5dfa336953e4fd5aea04ee81586)
---
.../protocol/opcua/client/IoTDBOpcUaClient.java | 116 ++++++++++++++++++---
.../opcua/client/IoTDBOpcUaClientTest.java | 92 ++++++++++++++++
2 files changed, 191 insertions(+), 17 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClient.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClient.java
index 65c651018ef..6ecdb95a8b3 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClient.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClient.java
@@ -63,6 +63,7 @@ import javax.annotation.Nullable;
import java.nio.file.Paths;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
@@ -73,10 +74,14 @@ import java.util.concurrent.ExecutionException;
import static
org.apache.iotdb.db.pipe.sink.protocol.opcua.server.OpcUaNameSpace.convertToOpcDataType;
import static
org.apache.iotdb.db.pipe.sink.protocol.opcua.server.OpcUaNameSpace.timestampToUtc;
import static org.eclipse.milo.opcua.stack.core.StatusCodes.Bad_Timeout;
+import static
org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn.Neither;
public class IoTDBOpcUaClient {
private static final Logger LOGGER =
LoggerFactory.getLogger(OpcUaNameSpace.class);
+ private static final int DEFAULT_MAX_NODES_PER_WRITE = 10_000;
+ private static final int DEFAULT_MAX_NODES_PER_NODE_MANAGEMENT = 250;
+
// Customized nodes
private static final int NAME_SPACE_INDEX = 2;
@@ -90,6 +95,9 @@ public class IoTDBOpcUaClient {
private final boolean historizing;
private ClientRunner runner;
+ private int maxNodesPerWrite = DEFAULT_MAX_NODES_PER_WRITE;
+ private int maxNodesPerNodeManagement =
DEFAULT_MAX_NODES_PER_NODE_MANAGEMENT;
+
public IoTDBOpcUaClient(
final String nodeUrl,
final SecurityPolicy securityPolicy,
@@ -118,6 +126,62 @@ public class IoTDBOpcUaClient {
}
break;
}
+ updateOperationLimits();
+ }
+
+ private void updateOperationLimits() {
+ try {
+ final List<DataValue> operationLimits =
+ client
+ .readValues(
+ 0.0,
+ Neither,
+ Arrays.asList(
+
Identifiers.Server_ServerCapabilities_OperationLimits_MaxNodesPerWrite,
+ Identifiers
+
.Server_ServerCapabilities_OperationLimits_MaxNodesPerNodeManagement))
+ .get();
+ maxNodesPerWrite = getOperationLimit(operationLimits, 0,
DEFAULT_MAX_NODES_PER_WRITE);
+ maxNodesPerNodeManagement =
+ getOperationLimit(operationLimits, 1,
DEFAULT_MAX_NODES_PER_NODE_MANAGEMENT);
+ LOGGER.info(
+ "OPC UA server operation limits: maxNodesPerWrite={},
maxNodesPerNodeManagement={}",
+ maxNodesPerWrite,
+ maxNodesPerNodeManagement);
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ LOGGER.warn(
+ "Interrupted while reading OPC UA server operation limits, use
defaults: "
+ + "maxNodesPerWrite={}, maxNodesPerNodeManagement={}",
+ DEFAULT_MAX_NODES_PER_WRITE,
+ DEFAULT_MAX_NODES_PER_NODE_MANAGEMENT);
+ } catch (final Exception e) {
+ LOGGER.warn(
+ "Failed to read OPC UA server operation limits, use defaults: "
+ + "maxNodesPerWrite={}, maxNodesPerNodeManagement={}",
+ DEFAULT_MAX_NODES_PER_WRITE,
+ DEFAULT_MAX_NODES_PER_NODE_MANAGEMENT,
+ e);
+ }
+ }
+
+ private static int getOperationLimit(
+ final List<DataValue> operationLimits, final int index, final int
defaultValue) {
+ if (Objects.isNull(operationLimits) || operationLimits.size() <= index) {
+ return defaultValue;
+ }
+
+ final DataValue dataValue = operationLimits.get(index);
+ if (Objects.isNull(dataValue)
+ || Objects.isNull(dataValue.getStatusCode())
+ || !dataValue.getStatusCode().isGood()
+ || Objects.isNull(dataValue.getValue())
+ || !(dataValue.getValue().getValue() instanceof Number)) {
+ return defaultValue;
+ }
+
+ final long limit = ((Number) dataValue.getValue().getValue()).longValue();
+ return limit == 0 ? Integer.MAX_VALUE : (int) Math.min(limit,
Integer.MAX_VALUE);
}
// Only support tree model & client-server
@@ -185,7 +249,8 @@ public class IoTDBOpcUaClient {
if (Objects.nonNull(sink.getValueName()) &&
!sink.getValueName().equals(name)) {
PipeLogger.log(
LOGGER::warn,
- "When the 'with-quality' mode is enabled, the measurement must be
either \"value-name\" or \"quality-name\"");
+ "When the 'with-quality' mode is enabled, the measurement must be
either \"value-name\""
+ + " or \"quality-name\"");
continue;
}
@@ -263,17 +328,23 @@ public class IoTDBOpcUaClient {
}
}
- final AddNodesResponse addStatus = client.addNodes(nodesToAdd).get();
- for (final AddNodesResult result : addStatus.getResults()) {
- if (!result.getStatusCode().equals(StatusCode.GOOD)
- && result.getStatusCode().getValue() !=
StatusCodes.Bad_NodeIdExists) {
- throw new PipeException(
- "Failed to create nodes after transfer data value, creation
status: "
- + addStatus
- + writeRequests
- .get(0)
- .getErrorString(new
StatusCode(StatusCodes.Bad_NodeIdUnknown)));
+ for (int startIndex = 0; startIndex < nodesToAdd.size(); ) {
+ final int endIndex =
+ getBatchEndIndex(startIndex, nodesToAdd.size(),
maxNodesPerNodeManagement);
+ final AddNodesResponse addStatus =
+ client.addNodes(nodesToAdd.subList(startIndex, endIndex)).get();
+ for (final AddNodesResult result : addStatus.getResults()) {
+ if (!result.getStatusCode().equals(StatusCode.GOOD)
+ && result.getStatusCode().getValue() !=
StatusCodes.Bad_NodeIdExists) {
+ throw new PipeException(
+ "Failed to create nodes after transfer data value, creation
status: "
+ + addStatus
+ + writeRequests
+ .get(0)
+ .getErrorString(new
StatusCode(StatusCodes.Bad_NodeIdUnknown)));
+ }
}
+ startIndex = endIndex;
}
}
@@ -290,13 +361,24 @@ public class IoTDBOpcUaClient {
private List<StatusCode> writeValuesOnce(final List<OpcUaWriteRequest>
writeRequests)
throws Exception {
- final List<NodeId> nodeIds = new ArrayList<>(writeRequests.size());
- final List<DataValue> dataValues = new ArrayList<>(writeRequests.size());
- for (final OpcUaWriteRequest writeRequest : writeRequests) {
- nodeIds.add(writeRequest.nodeId);
- dataValues.add(writeRequest.dataValue);
+ final List<StatusCode> writeStatuses = new
ArrayList<>(writeRequests.size());
+ for (int startIndex = 0; startIndex < writeRequests.size(); ) {
+ final int endIndex = getBatchEndIndex(startIndex, writeRequests.size(),
maxNodesPerWrite);
+ final List<NodeId> nodeIds = new ArrayList<>(endIndex - startIndex);
+ final List<DataValue> dataValues = new ArrayList<>(endIndex -
startIndex);
+ for (int i = startIndex; i < endIndex; ++i) {
+ nodeIds.add(writeRequests.get(i).nodeId);
+ dataValues.add(writeRequests.get(i).dataValue);
+ }
+ writeStatuses.addAll(client.writeValues(nodeIds, dataValues).get());
+ startIndex = endIndex;
}
- return client.writeValues(nodeIds, dataValues).get();
+ return writeStatuses;
+ }
+
+ private static int getBatchEndIndex(
+ final int startIndex, final int totalSize, final int batchSize) {
+ return (int) Math.min((long) totalSize, (long) startIndex + batchSize);
}
private static final class OpcUaWriteRequest {
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClientTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClientTest.java
index 0eaa1cf0234..85879ad80ad 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClientTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClientTest.java
@@ -35,8 +35,12 @@ import org.eclipse.milo.opcua.sdk.client.api.UaClient;
import org.eclipse.milo.opcua.sdk.client.api.identity.AnonymousProvider;
import org.eclipse.milo.opcua.stack.core.StatusCodes;
import org.eclipse.milo.opcua.stack.core.security.SecurityPolicy;
+import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue;
+import org.eclipse.milo.opcua.stack.core.types.builtin.ExpandedNodeId;
import org.eclipse.milo.opcua.stack.core.types.builtin.NodeId;
import org.eclipse.milo.opcua.stack.core.types.builtin.StatusCode;
+import org.eclipse.milo.opcua.stack.core.types.builtin.Variant;
+import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn;
import org.eclipse.milo.opcua.stack.core.types.structured.AddNodesItem;
import org.eclipse.milo.opcua.stack.core.types.structured.AddNodesResponse;
import org.eclipse.milo.opcua.stack.core.types.structured.AddNodesResult;
@@ -53,6 +57,8 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
+import static
org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.Unsigned.uint;
+
public class IoTDBOpcUaClientTest {
@Test
@@ -131,6 +137,78 @@ public class IoTDBOpcUaClientTest {
.writeValues(Mockito.argThat(nodeIds("root/db/d1/s1")),
Mockito.argThat(listWithSize(1)));
}
+ @Test
+ public void testTransferSplitsWritesAtServerLimit() throws Exception {
+ final OpcUaClient miloClient = Mockito.mock(OpcUaClient.class);
+ Mockito.when(miloClient.writeValues(Mockito.anyList(), Mockito.anyList()))
+ .thenAnswer(
+ invocation -> {
+ final int size = ((List<?>) invocation.getArguments()[0]).size();
+ return
CompletableFuture.completedFuture(Collections.nCopies(size, StatusCode.GOOD));
+ });
+ final IoTDBOpcUaClient client = createClient(miloClient, 1, 250);
+
+ client.transfer(createTablet(), createSink());
+
+ final InOrder inOrder = Mockito.inOrder(miloClient);
+ inOrder
+ .verify(miloClient)
+ .writeValues(Mockito.argThat(nodeIds("root/db/d1/s1")),
Mockito.argThat(listWithSize(1)));
+ inOrder
+ .verify(miloClient)
+ .writeValues(Mockito.argThat(nodeIds("root/db/d1/s2")),
Mockito.argThat(listWithSize(1)));
+ }
+
+ @Test
+ public void testTransferSplitsMissingNodeCreationAtServerLimit() throws
Exception {
+ final OpcUaClient miloClient = Mockito.mock(OpcUaClient.class);
+ Mockito.when(miloClient.writeValues(Mockito.anyList(), Mockito.anyList()))
+ .thenReturn(
+ CompletableFuture.completedFuture(
+ Arrays.asList(
+ new StatusCode(StatusCodes.Bad_NodeIdUnknown),
+ new StatusCode(StatusCodes.Bad_NodeIdUnknown))))
+ .thenReturn(
+ CompletableFuture.completedFuture(Arrays.asList(StatusCode.GOOD,
StatusCode.GOOD)));
+
+ final AddNodesResponse addNodesResponse =
Mockito.mock(AddNodesResponse.class);
+ final AddNodesResult addNodesResult = Mockito.mock(AddNodesResult.class);
+ Mockito.when(addNodesResult.getStatusCode()).thenReturn(StatusCode.GOOD);
+ Mockito.when(addNodesResponse.getResults()).thenReturn(new
AddNodesResult[] {addNodesResult});
+ Mockito.when(miloClient.addNodes(Mockito.anyList()))
+ .thenReturn(CompletableFuture.completedFuture(addNodesResponse));
+
+ final IoTDBOpcUaClient client = Mockito.spy(createClient(miloClient,
10_000, 1));
+ final AddNodesItem firstNode = Mockito.mock(AddNodesItem.class);
+ final AddNodesItem secondNode = Mockito.mock(AddNodesItem.class);
+ final ExpandedNodeId firstNodeId = new NodeId(2,
"root/db/d1/s1").expanded();
+ final ExpandedNodeId secondNodeId = new NodeId(2,
"root/db/d1/s2").expanded();
+ Mockito.when(firstNode.getRequestedNewNodeId()).thenReturn(firstNodeId);
+ Mockito.when(secondNode.getRequestedNewNodeId()).thenReturn(secondNodeId);
+ Mockito.doAnswer(
+ invocation ->
+ Collections.singletonList(
+ "s1".equals(invocation.getArguments()[1]) ? firstNode :
secondNode))
+ .when(client)
+ .getNodesToAdd(
+ Mockito.any(String[].class),
+ Mockito.anyString(),
+ Mockito.any(NodeId.class),
+ Mockito.any());
+
+ client.transfer(createTablet(), createSink());
+
+ Mockito.verify(miloClient,
Mockito.times(2)).addNodes(Mockito.argThat(listWithSize(1)));
+ final InOrder inOrder = Mockito.inOrder(miloClient);
+ inOrder
+ .verify(miloClient)
+ .writeValues(Mockito.argThat(listWithSize(2)),
Mockito.argThat(listWithSize(2)));
+ inOrder.verify(miloClient,
Mockito.times(2)).addNodes(Mockito.argThat(listWithSize(1)));
+ inOrder
+ .verify(miloClient)
+ .writeValues(Mockito.argThat(listWithSize(2)),
Mockito.argThat(listWithSize(2)));
+ }
+
@Test
public void testTransferFailsOnNonRecoverableStatus() throws Exception {
final OpcUaClient miloClient = Mockito.mock(OpcUaClient.class);
@@ -152,6 +230,12 @@ public class IoTDBOpcUaClientTest {
}
private static IoTDBOpcUaClient createClient(final OpcUaClient miloClient)
throws Exception {
+ return createClient(miloClient, 10_000, 250);
+ }
+
+ private static IoTDBOpcUaClient createClient(
+ final OpcUaClient miloClient, final int maxNodesPerWrite, final int
maxNodesPerNodeManagement)
+ throws Exception {
final IoTDBOpcUaClient client =
new IoTDBOpcUaClient(
"opc.tcp://127.0.0.1:12686", SecurityPolicy.None, new
AnonymousProvider(), false);
@@ -160,6 +244,14 @@ public class IoTDBOpcUaClientTest {
client.setRunner(runner);
final CompletableFuture<UaClient> connectFuture =
CompletableFuture.completedFuture(miloClient);
Mockito.when(miloClient.connect()).thenReturn(connectFuture);
+ Mockito.when(
+ miloClient.readValues(
+ Mockito.anyDouble(), Mockito.eq(TimestampsToReturn.Neither),
Mockito.anyList()))
+ .thenReturn(
+ CompletableFuture.completedFuture(
+ Arrays.asList(
+ new DataValue(new Variant(uint(maxNodesPerWrite))),
+ new DataValue(new
Variant(uint(maxNodesPerNodeManagement))))));
client.run(miloClient);
return client;
}
