This is an automated email from the ASF dual-hosted git repository.
justinchen pushed a commit to branch opc-ua-fix
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/opc-ua-fix by this push:
new 8b769d98f38 fix
8b769d98f38 is described below
commit 8b769d98f38cf9361cd9a595e03419dcd4fd1318
Author: Caideyipi <[email protected]>
AuthorDate: Tue Mar 17 19:05:50 2026 +0800
fix
---
.../task/subtask/sink/PipeSinkSubtaskManager.java | 45 +++++++++++++---------
.../sink/protocol/opcua/server/OpcUaNameSpace.java | 25 ++++++------
.../pipe/config/constant/PipeSinkConstant.java | 8 ++++
3 files changed, 48 insertions(+), 30 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManager.java
index 3a461fe0395..d848c487a68 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManager.java
@@ -64,10 +64,10 @@ public class PipeSinkSubtaskManager {
public synchronized String register(
final Supplier<? extends PipeSinkSubtaskExecutor> executorSupplier,
- final PipeParameters pipeConnectorParameters,
+ final PipeParameters pipeSinkParameters,
final PipeTaskSinkRuntimeEnvironment environment) {
final String connectorKey =
- pipeConnectorParameters
+ pipeSinkParameters
.getStringOrDefault(
Arrays.asList(PipeSinkConstant.CONNECTOR_KEY,
PipeSinkConstant.SINK_KEY),
BuiltinPipePlugin.IOTDB_THRIFT_CONNECTOR.getPipePluginName())
@@ -81,24 +81,32 @@ public class PipeSinkSubtaskManager {
environment.getRegionId(),
connectorKey);
- final boolean isDataRegionConnector =
+ final boolean isDataSinkConnector =
StorageEngine.getInstance()
.getAllDataRegionIds()
.contains(new DataRegionId(environment.getRegionId()))
|| PipeRuntimeMeta.isSourceExternal(environment.getRegionId());
- final int connectorNum;
+ final int sinkNum;
boolean realTimeFirst = false;
- String attributeSortedString =
generateAttributeSortedString(pipeConnectorParameters);
- if (isDataRegionConnector) {
- connectorNum =
- pipeConnectorParameters.getIntOrDefault(
+ String attributeSortedString =
generateAttributeSortedString(pipeSinkParameters);
+ if (isDataSinkConnector) {
+ sinkNum =
+ pipeSinkParameters.getIntOrDefault(
Arrays.asList(
PipeSinkConstant.CONNECTOR_IOTDB_PARALLEL_TASKS_KEY,
PipeSinkConstant.SINK_IOTDB_PARALLEL_TASKS_KEY),
- PipeSinkConstant.CONNECTOR_IOTDB_PARALLEL_TASKS_DEFAULT_VALUE);
+ PipeSinkConstant.SINGLE_THREAD_DEFAULT_SINK.contains(
+ pipeSinkParameters
+ .getStringOrDefault(
+ Arrays.asList(
+ PipeSinkConstant.CONNECTOR_KEY,
PipeSinkConstant.SINK_KEY),
+
BuiltinPipePlugin.IOTDB_THRIFT_SINK.getPipePluginName())
+ .toLowerCase())
+ ? 1
+ :
PipeSinkConstant.CONNECTOR_IOTDB_PARALLEL_TASKS_DEFAULT_VALUE);
realTimeFirst =
- pipeConnectorParameters.getBooleanOrDefault(
+ pipeSinkParameters.getBooleanOrDefault(
Arrays.asList(
PipeSinkConstant.CONNECTOR_REALTIME_FIRST_KEY,
PipeSinkConstant.SINK_REALTIME_FIRST_KEY),
@@ -107,15 +115,14 @@ public class PipeSinkSubtaskManager {
} else {
// Do not allow parallel tasks for schema region connectors
// to avoid the potential disorder of the schema region data transfer
- connectorNum = 1;
+ sinkNum = 1;
attributeSortedString = "schema_" + attributeSortedString;
}
environment.setAttributeSortedString(attributeSortedString);
if
(!attributeSortedString2SubtaskLifeCycleMap.containsKey(attributeSortedString))
{
final PipeSinkSubtaskExecutor executor = executorSupplier.get();
- final List<PipeSinkSubtaskLifeCycle> pipeSinkSubtaskLifeCycleList =
- new ArrayList<>(connectorNum);
+ final List<PipeSinkSubtaskLifeCycle> pipeSinkSubtaskLifeCycleList = new
ArrayList<>(sinkNum);
AtomicInteger counter = new AtomicInteger(0);
// Shared pending queue for all subtasks
@@ -128,20 +135,20 @@ public class PipeSinkSubtaskManager {
((PipeRealtimePriorityBlockingQueue)
pendingQueue).setOfferTsFileCounter(counter);
}
- for (int connectorIndex = 0; connectorIndex < connectorNum;
connectorIndex++) {
+ for (int connectorIndex = 0; connectorIndex < sinkNum; connectorIndex++)
{
final PipeConnector pipeConnector =
- isDataRegionConnector
- ?
PipeDataNodeAgent.plugin().dataRegion().reflectSink(pipeConnectorParameters)
- :
PipeDataNodeAgent.plugin().schemaRegion().reflectSink(pipeConnectorParameters);
+ isDataSinkConnector
+ ?
PipeDataNodeAgent.plugin().dataRegion().reflectSink(pipeSinkParameters)
+ :
PipeDataNodeAgent.plugin().schemaRegion().reflectSink(pipeSinkParameters);
// 1. Construct, validate and customize PipeConnector, and then
handshake (create
// connection) with the target
try {
if (pipeConnector instanceof IoTDBDataRegionAsyncSink) {
((IoTDBDataRegionAsyncSink)
pipeConnector).setTransferTsFileCounter(counter);
}
- pipeConnector.validate(new
PipeParameterValidator(pipeConnectorParameters));
+ pipeConnector.validate(new
PipeParameterValidator(pipeSinkParameters));
pipeConnector.customize(
- pipeConnectorParameters, new
PipeTaskRuntimeConfiguration(environment));
+ pipeSinkParameters, new
PipeTaskRuntimeConfiguration(environment));
pipeConnector.handshake();
} catch (final Exception e) {
try {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpace.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpace.java
index 56a90680185..2c2e860af7b 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpace.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpace.java
@@ -278,6 +278,14 @@ public class OpcUaNameSpace extends
ManagedNamespaceWithLifecycle {
Objects.isNull(sink.getValueName()) ? name :
segments[segments.length - 1];
final NodeId nodeId = newNodeId(currentFolder + nodeName);
final UaVariableNode measurementNode;
+ final long utcTimestamp =
timestampToUtc(timestamps.get(timestamps.size() > 1 ? i : 0));
+ final DataValue dataValue =
+ new DataValue(
+ new Variant(values.get(i)),
+ currentQuality,
+ new DateTime(utcTimestamp),
+ new DateTime());
+
if (!getNodeManager().containsNode(nodeId)) {
measurementNode =
new UaVariableNode.UaVariableNodeBuilder(getNodeContext())
@@ -288,6 +296,7 @@ public class OpcUaNameSpace extends
ManagedNamespaceWithLifecycle {
.setDisplayName(LocalizedText.english(nodeName))
.setDataType(convertToOpcDataType(type))
.setTypeDefinition(Identifiers.BaseDataVariableType)
+ .setValue(dataValue)
.build();
getNodeManager().addNode(measurementNode);
if (Objects.nonNull(folderNode)) {
@@ -311,17 +320,11 @@ public class OpcUaNameSpace extends
ManagedNamespaceWithLifecycle {
String.format("The Node %s does not exist.",
nodeId)));
}
- final long utcTimestamp =
timestampToUtc(timestamps.get(timestamps.size() > 1 ? i : 0));
if (Objects.isNull(sink.getValueName())) {
if (Objects.isNull(measurementNode.getValue())
- ||
Objects.requireNonNull(measurementNode.getValue().getSourceTime()).getUtcTime()
- < utcTimestamp) {
- measurementNode.setValue(
- new DataValue(
- new Variant(values.get(i)),
- currentQuality,
- new DateTime(utcTimestamp),
- new DateTime()));
+ || Objects.isNull(measurementNode.getValue().getSourceTime())
+ || measurementNode.getValue().getSourceTime().getUtcTime() <
utcTimestamp) {
+ measurementNode.setValue(dataValue);
}
} else {
valueNode = measurementNode;
@@ -331,8 +334,8 @@ public class OpcUaNameSpace extends
ManagedNamespaceWithLifecycle {
}
if (Objects.nonNull(valueNode)) {
if (Objects.isNull(valueNode.getValue())
- ||
Objects.requireNonNull(valueNode.getValue().getSourceTime()).getUtcTime()
- < timestamp) {
+ || Objects.isNull(valueNode.getValue().getSourceTime())
+ || valueNode.getValue().getSourceTime().getUtcTime() < timestamp) {
valueNode.setValue(
new DataValue(
new Variant(value), currentQuality, new DateTime(timestamp),
new DateTime()));
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 ec9afce7c6f..7e1e38e1490 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
@@ -20,6 +20,7 @@
package org.apache.iotdb.commons.pipe.config.constant;
import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
import com.github.luben.zstd.Zstd;
@@ -55,6 +56,13 @@ public class PipeSinkConstant {
public static final String SINK_IOTDB_PARALLEL_TASKS_KEY =
"sink.parallel.tasks";
public static final int CONNECTOR_IOTDB_PARALLEL_TASKS_DEFAULT_VALUE =
PipeConfig.getInstance().getPipeSubtaskExecutorMaxThreadNum();
+ public static final Set<String> SINGLE_THREAD_DEFAULT_SINK =
+ new HashSet<>(
+ Arrays.asList(
+ BuiltinPipePlugin.OPC_UA_SINK.getPipePluginName(),
+ BuiltinPipePlugin.OPC_UA_CONNECTOR.getPipePluginName(),
+ BuiltinPipePlugin.OPC_DA_SINK.getPipePluginName(),
+ BuiltinPipePlugin.OPC_DA_CONNECTOR.getPipePluginName()));
public static final String CONNECTOR_REALTIME_FIRST_KEY =
"connector.realtime-first";
public static final String SINK_REALTIME_FIRST_KEY = "sink.realtime-first";
