From 28b1104b8e72b96e1fd2d8ceaaa71c243b68198b Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Thu, 13 Aug 2026 14:09:56 +0800 Subject: [PATCH 1/2] Move OPC UA sink to pipe library (cherry picked from commit a9007d652e8ceb2a9ea9813e03b18090bcd8d919) --- LICENSE-binary | 18 ++ distribution/pom.xml | 9 + distribution/src/assembly/library-pipe.xml | 51 ++++ example/pom.xml | 1 - integration-test/pom.xml | 7 + .../pipe/it/single/AbstractPipeSingleIT.java | 70 ++++- .../pipe/it/single/IoTDBPipeOPCUAIT.java | 9 +- iotdb-core/datanode/pom.xml | 24 -- .../PipeDataRegionSinkConstructor.java | 3 - .../pipe/sink/protocol/opcda/OpcDaSink.java | 9 +- .../util/PipeTabletEventTransferUtils.java | 97 ++++++ .../sink/PipeSinkSubtaskManagerTest.java | 4 +- .../iotdb/db/pipe/sink/PipeSinkTest.java | 93 ------ .../plugin/builtin/BuiltinPipePlugin.java | 5 - .../plugin/builtin/sink/opcua/OpcUaSink.java | 34 --- .../config/constant/PipeSinkConstant.java | 7 +- .../opc-ua-sink-example}/pom.xml | 5 +- .../org/apache/iotdb/opcua/ClientExample.java | 0 .../iotdb/opcua/ClientExampleRunner.java | 0 .../org/apache/iotdb/opcua/ClientTest.java | 0 .../opcua/IoTDBKeyStoreLoaderClient.java | 0 library-pipe/opc-ua-sink/README.md | 73 +++++ library-pipe/opc-ua-sink/pom.xml | 159 ++++++++++ .../pipe/plugin/sink}/opcua/OpcUaSink.java | 91 ++---- .../sink}/opcua/client/ClientRunner.java | 2 +- .../client/IoTDBKeyStoreLoaderClient.java | 2 +- .../sink}/opcua/client/IoTDBOpcUaClient.java | 10 +- .../opcua/server/OpcUaKeyStoreLoader.java | 2 +- .../sink}/opcua/server/OpcUaNameSpace.java | 4 +- .../opcua/server/OpcUaServerBuilder.java | 2 +- .../pipe/plugin/sink/opcua/OpcUaSinkTest.java | 285 ++++++++++++++++++ .../sink}/opcua/client/ClientRunnerTest.java | 2 +- .../opcua/client/IoTDBOpcUaClientTest.java | 4 +- .../opcua/server/OpcUaServerBuilderTest.java | 2 +- .../server/OpcUaTcpNoneCompatibilityTest.java | 2 +- library-pipe/pom.xml | 68 +++++ licenses/Bouncy Castle Licence | 17 ++ pom.xml | 27 +- 38 files changed, 907 insertions(+), 291 deletions(-) create mode 100644 distribution/src/assembly/library-pipe.xml create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/PipeTabletEventTransferUtils.java delete mode 100644 iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/sink/opcua/OpcUaSink.java rename {example/pipe-opc-ua-sink => library-pipe/opc-ua-sink-example}/pom.xml (95%) rename {example/pipe-opc-ua-sink => library-pipe/opc-ua-sink-example}/src/main/java/org/apache/iotdb/opcua/ClientExample.java (100%) rename {example/pipe-opc-ua-sink => library-pipe/opc-ua-sink-example}/src/main/java/org/apache/iotdb/opcua/ClientExampleRunner.java (100%) rename {example/pipe-opc-ua-sink => library-pipe/opc-ua-sink-example}/src/main/java/org/apache/iotdb/opcua/ClientTest.java (100%) rename {example/pipe-opc-ua-sink => library-pipe/opc-ua-sink-example}/src/main/java/org/apache/iotdb/opcua/IoTDBKeyStoreLoaderClient.java (100%) create mode 100644 library-pipe/opc-ua-sink/README.md create mode 100644 library-pipe/opc-ua-sink/pom.xml rename {iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol => library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink}/opcua/OpcUaSink.java (89%) rename {iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol => library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink}/opcua/client/ClientRunner.java (99%) rename {iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol => library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink}/opcua/client/IoTDBKeyStoreLoaderClient.java (98%) rename {iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol => library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink}/opcua/client/IoTDBOpcUaClient.java (97%) rename {iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol => library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink}/opcua/server/OpcUaKeyStoreLoader.java (98%) rename {iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol => library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink}/opcua/server/OpcUaNameSpace.java (99%) rename {iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol => library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink}/opcua/server/OpcUaServerBuilder.java (99%) create mode 100644 library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/OpcUaSinkTest.java rename {iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol => library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink}/opcua/client/ClientRunnerTest.java (98%) rename {iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol => library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink}/opcua/client/IoTDBOpcUaClientTest.java (98%) rename {iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol => library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink}/opcua/server/OpcUaServerBuilderTest.java (99%) rename {iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol => library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink}/opcua/server/OpcUaTcpNoneCompatibilityTest.java (98%) create mode 100644 library-pipe/pom.xml create mode 100644 licenses/Bouncy Castle Licence diff --git a/LICENSE-binary b/LICENSE-binary index 4bc2041d9fc2..6dc4df2cea69 100644 --- a/LICENSE-binary +++ b/LICENSE-binary @@ -218,6 +218,8 @@ Apache License 2.0 commons-cli:commons-cli:1.5.0 com.google.code.gson:gson:2.13.1 com.google.guava.guava:32.1.2-jre +com.google.guava:failureaccess:1.0.3 +com.google.guava:guava:33.6.0-jre com.fasterxml.jackson.core:jackson-annotations:2.16.2 com.fasterxml.jackson.core:jackson-core:2.16.2 com.fasterxml.jackson.core:jackson-databind:2.16.2 @@ -260,6 +262,8 @@ io.netty:netty-transport-native-unix-common:4.1.136.Final org.osgi:org.osgi.core:7.0.0 org.osgi:osgi.cmpn:7.0.0 org.ops4j.pax.jdbc:pax-jdbc-common:1.5.6 +org.jspecify:jspecify:1.0.0 +org.jctools:jctools-core:4.0.5 org.xerial.snappy:snappy-java:1.1.10.5 io.airlift.airline:0.9 @@ -292,8 +296,22 @@ EPL 2.0 ------------ ch.qos.logback:logback-classic:1.5.34 ch.qos.logback:logback-core:1.5.34 +com.digitalpetri.fsm:strict-machine:1.0.0 +com.digitalpetri.netty:netty-channel-fsm:1.0.2 jakarta.annotation:jakarta.annotation-api:3.0.0 jakarta.servlet:jakarta.servlet-api:6.0.0 +org.eclipse.milo:milo-guava-dependencies:1.1.6 +org.eclipse.milo:milo-sdk-client:1.1.6 +org.eclipse.milo:milo-sdk-core:1.1.6 +org.eclipse.milo:milo-sdk-server:1.1.6 +org.eclipse.milo:milo-stack-core:1.1.6 +org.eclipse.milo:milo-transport:1.1.6 + +Bouncy Castle Licence +------------ +org.bouncycastle:bcpkix-jdk18on:1.84 +org.bouncycastle:bcprov-jdk18on:1.84 +org.bouncycastle:bcutil-jdk18on:1.84 EDL 1.0 ------------ diff --git a/distribution/pom.xml b/distribution/pom.xml index 3e618451f58f..b38c127a8b8f 100644 --- a/distribution/pom.xml +++ b/distribution/pom.xml @@ -33,6 +33,13 @@ ${os.classifier} + + org.apache.iotdb + opc-ua-sink + ${project.version} + jar-with-dependencies + provided + org.apache.iotdb iotdb-server @@ -89,6 +96,7 @@ src/assembly/confignode.xml src/assembly/cli.xml src/assembly/library-udf.xml + src/assembly/library-pipe.xml apache-iotdb-${project.version} @@ -122,6 +130,7 @@ apache-iotdb-${project.version}-cli-bin.zip apache-iotdb-${project.version}-confignode-bin.zip apache-iotdb-${project.version}-library-udf-bin.zip + apache-iotdb-${project.version}-library-pipe-bin.zip apache-iotdb-${project.version}-external-service-impl-bin.zip diff --git a/distribution/src/assembly/library-pipe.xml b/distribution/src/assembly/library-pipe.xml new file mode 100644 index 000000000000..b10a33d8ea3d --- /dev/null +++ b/distribution/src/assembly/library-pipe.xml @@ -0,0 +1,51 @@ + + + + library-pipe-bin + + dir + zip + + apache-iotdb-${project.version}-library-pipe-bin + + + ${project.basedir}/../licenses + licenses + + + + + ${project.basedir}/../LICENSE-binary + licenses + LICENSE + + + ${project.basedir}/../NOTICE-binary + licenses + NOTICE + + + ${maven.multiModuleProjectDirectory}/library-pipe/opc-ua-sink/target/opc-ua-sink-${project.version}-jar-with-dependencies.jar + ext/pipe + + + diff --git a/example/pom.xml b/example/pom.xml index f7767ce6cb58..78054302f71c 100644 --- a/example/pom.xml +++ b/example/pom.xml @@ -34,7 +34,6 @@ mqtt mqtt-customize pipe-count-point-processor - pipe-opc-ua-sink rest-java-example schema session diff --git a/integration-test/pom.xml b/integration-test/pom.xml index 4dfea2524780..abe519cd2bea 100644 --- a/integration-test/pom.xml +++ b/integration-test/pom.xml @@ -256,6 +256,12 @@ provided + + org.apache.iotdb + opc-ua-sink + ${project.version} + test + @@ -330,6 +336,7 @@ false ${integrationTest.testEnv} + ${maven.multiModuleProjectDirectory}/library-pipe/opc-ua-sink/target/opc-ua-sink-${project.version}-jar-with-dependencies.jar ${integrationTest.randomSelectWriteNode} ${integrationTest.readAndVerifyWithMultiNode} ${integrationTest.dataRegionPerDataNode} diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/AbstractPipeSingleIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/AbstractPipeSingleIT.java index 61d4f0157e4f..7cff83623bef 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/AbstractPipeSingleIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/AbstractPipeSingleIT.java @@ -19,18 +19,33 @@ package org.apache.iotdb.pipe.it.single; +import org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant; +import org.apache.iotdb.db.it.utils.TestUtils; import org.apache.iotdb.it.env.MultiEnvFactory; import org.apache.iotdb.itbase.env.BaseEnv; import org.junit.After; import org.junit.Before; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.List; +import java.util.stream.Stream; + abstract class AbstractPipeSingleIT { + private static final String OPC_UA_SINK_CLASS_NAME = + "org.apache.iotdb.pipe.plugin.sink.opcua.OpcUaSink"; + private static final String OPC_UA_SINK_JAR_PATH_PROPERTY = "OpcUaSinkPluginJar"; + private static final String OPC_UA_SINK_JAR_PREFIX = "opc-ua-sink-"; + private static final String OPC_UA_SINK_JAR_SUFFIX = "-jar-with-dependencies.jar"; + protected BaseEnv env; @Before - public void setUp() { + public void setUp() throws Exception { MultiEnvFactory.createEnv(1); env = MultiEnvFactory.getEnv(0); env.getConfig() @@ -43,6 +58,59 @@ public void setUp() { env.initClusterEnvironment(); } + protected final void registerOpcUaSinkPlugin() throws IOException { + final String pluginUri = locateOpcUaSinkPluginJar().toUri().toString(); + registerOpcUaSinkPlugin(PipeSinkConstant.OPC_UA_SINK_NAME, pluginUri); + } + + private void registerOpcUaSinkPlugin(final String pluginName, final String pluginUri) { + TestUtils.executeNonQuery( + env, + String.format( + "CREATE PIPEPLUGIN IF NOT EXISTS `%s` AS '%s' USING URI '%s'", + pluginName, OPC_UA_SINK_CLASS_NAME, pluginUri)); + } + + private static Path locateOpcUaSinkPluginJar() throws IOException { + final String configuredPluginJar = System.getProperty(OPC_UA_SINK_JAR_PATH_PROPERTY); + if (configuredPluginJar != null && !configuredPluginJar.isEmpty()) { + final Path pluginJar = Paths.get(configuredPluginJar).toAbsolutePath().normalize(); + if (Files.isRegularFile(pluginJar)) { + return pluginJar; + } + throw new IOException("Cannot locate the OPC UA sink plugin jar at " + pluginJar + "."); + } + + Path current = Paths.get(System.getProperty("user.dir")).toAbsolutePath().normalize(); + while (current != null) { + final Path targetDir = + current.resolve("library-pipe").resolve("opc-ua-sink").resolve("target"); + if (Files.isDirectory(targetDir)) { + try (final Stream stream = Files.list(targetDir)) { + final List pluginJars = + stream + .filter(Files::isRegularFile) + .filter(path -> path.getFileName().toString().startsWith(OPC_UA_SINK_JAR_PREFIX)) + .filter(path -> path.getFileName().toString().endsWith(OPC_UA_SINK_JAR_SUFFIX)) + .toList(); + if (pluginJars.size() == 1) { + return pluginJars.get(0); + } + if (pluginJars.size() > 1) { + throw new IOException( + "Multiple OPC UA sink plugin jars found in " + + targetDir + + ". Set -D" + + OPC_UA_SINK_JAR_PATH_PROPERTY + + " to the expected jar."); + } + } + } + current = current.getParent(); + } + throw new IOException("Cannot locate the OPC UA sink plugin jar."); + } + @After public final void tearDown() { env.cleanClusterEnvironment(); diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBPipeOPCUAIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBPipeOPCUAIT.java index 4054f7fae5f2..e556de47e65c 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBPipeOPCUAIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBPipeOPCUAIT.java @@ -23,14 +23,14 @@ import org.apache.iotdb.confignode.rpc.thrift.TAlterPipeReq; import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq; import org.apache.iotdb.db.it.utils.TestUtils; -import org.apache.iotdb.db.pipe.sink.protocol.opcua.client.ClientRunner; -import org.apache.iotdb.db.pipe.sink.protocol.opcua.client.IoTDBOpcUaClient; import org.apache.iotdb.it.env.MultiEnvFactory; import org.apache.iotdb.it.env.cluster.EnvUtils; import org.apache.iotdb.it.framework.IoTDBTestRunner; import org.apache.iotdb.itbase.category.MultiClusterIT1; import org.apache.iotdb.pipe.api.exception.PipeException; import org.apache.iotdb.pipe.it.dual.tablemodel.TableModelUtils; +import org.apache.iotdb.pipe.plugin.sink.opcua.client.ClientRunner; +import org.apache.iotdb.pipe.plugin.sink.opcua.client.IoTDBOpcUaClient; import org.apache.iotdb.rpc.TSStatusCode; import org.apache.tsfile.common.conf.TSFileConfig; @@ -61,14 +61,14 @@ import java.util.UUID; import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_SECURITY_DIR_DEFAULT_VALUE; -import static org.apache.iotdb.db.pipe.sink.protocol.opcua.server.OpcUaNameSpace.timestampToUtc; +import static org.apache.iotdb.pipe.plugin.sink.opcua.server.OpcUaNameSpace.timestampToUtc; @RunWith(IoTDBTestRunner.class) @Category({MultiClusterIT1.class}) public class IoTDBPipeOPCUAIT extends AbstractPipeSingleIT { @Before - public void setUp() { + public void setUp() throws Exception { MultiEnvFactory.createEnv(1); env = MultiEnvFactory.getEnv(0); env.getConfig() @@ -81,6 +81,7 @@ public void setUp() { .setIsPipeEnableMemoryCheck(false) .setPipeAutoSplitFullEnabled(false); env.initClusterEnvironment(1, 1); + registerOpcUaSinkPlugin(); } @Test diff --git a/iotdb-core/datanode/pom.xml b/iotdb-core/datanode/pom.xml index b615061fc745..c3bc904f5c76 100644 --- a/iotdb-core/datanode/pom.xml +++ b/iotdb-core/datanode/pom.xml @@ -185,26 +185,6 @@ net.java.dev.jna jna-platform - - org.eclipse.milo - milo-stack-core - - - org.eclipse.milo - milo-sdk-core - - - org.eclipse.milo - milo-transport - - - org.eclipse.milo - milo-sdk-client - - - org.bouncycastle - bcprov-jdk18on - org.apache.commons commons-pool2 @@ -217,10 +197,6 @@ io.airlift airline - - org.eclipse.milo - milo-sdk-server - commons-cli commons-cli diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/dataregion/PipeDataRegionSinkConstructor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/dataregion/PipeDataRegionSinkConstructor.java index c8e87890afd6..7044336f6397 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/dataregion/PipeDataRegionSinkConstructor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/dataregion/PipeDataRegionSinkConstructor.java @@ -27,7 +27,6 @@ import org.apache.iotdb.db.pipe.sink.protocol.iotconsensusv2.IoTConsensusV2AsyncSink; import org.apache.iotdb.db.pipe.sink.protocol.legacy.IoTDBLegacyPipeSink; import org.apache.iotdb.db.pipe.sink.protocol.opcda.OpcDaSink; -import org.apache.iotdb.db.pipe.sink.protocol.opcua.OpcUaSink; import org.apache.iotdb.db.pipe.sink.protocol.thrift.async.IoTDBDataRegionAsyncSink; import org.apache.iotdb.db.pipe.sink.protocol.thrift.sync.IoTDBDataRegionSyncSink; import org.apache.iotdb.db.pipe.sink.protocol.websocket.WebSocketSink; @@ -68,7 +67,6 @@ protected void initConstructors() { IoTDBDataRegionAirGapSink::new); pluginConstructors.put( BuiltinPipePlugin.WEBSOCKET_CONNECTOR.getPipePluginName(), WebSocketSink::new); - pluginConstructors.put(BuiltinPipePlugin.OPC_UA_CONNECTOR.getPipePluginName(), OpcUaSink::new); pluginConstructors.put(BuiltinPipePlugin.OPC_DA_CONNECTOR.getPipePluginName(), OpcDaSink::new); pluginConstructors.put( BuiltinPipePlugin.DO_NOTHING_CONNECTOR.getPipePluginName(), DoNothingSink::new); @@ -90,7 +88,6 @@ protected void initConstructors() { BuiltinPipePlugin.IOTDB_AIR_GAP_SINK.getPipePluginName(), IoTDBDataRegionAirGapSink::new); pluginConstructors.put( BuiltinPipePlugin.WEBSOCKET_SINK.getPipePluginName(), WebSocketSink::new); - pluginConstructors.put(BuiltinPipePlugin.OPC_UA_SINK.getPipePluginName(), OpcUaSink::new); pluginConstructors.put(BuiltinPipePlugin.OPC_DA_SINK.getPipePluginName(), OpcDaSink::new); pluginConstructors.put( BuiltinPipePlugin.DO_NOTHING_SINK.getPipePluginName(), DoNothingSink::new); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcda/OpcDaSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcda/OpcDaSink.java index d13e743bfc9a..56025e3fdf92 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcda/OpcDaSink.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcda/OpcDaSink.java @@ -20,7 +20,7 @@ package org.apache.iotdb.db.pipe.sink.protocol.opcda; import org.apache.iotdb.db.i18n.DataNodePipeMessages; -import org.apache.iotdb.db.pipe.sink.protocol.opcua.OpcUaSink; +import org.apache.iotdb.db.pipe.sink.util.PipeTabletEventTransferUtils; import org.apache.iotdb.pipe.api.PipeConnector; import org.apache.iotdb.pipe.api.annotation.TreeModel; import org.apache.iotdb.pipe.api.customizer.configuration.PipeConnectorRuntimeConfiguration; @@ -110,8 +110,11 @@ public void heartbeat() throws Exception { @Override public void transfer(final TabletInsertionEvent tabletInsertionEvent) throws Exception { - OpcUaSink.transferByTablet( - tabletInsertionEvent, LOGGER, (tablet, isTableModel) -> handle.transfer(tablet)); + PipeTabletEventTransferUtils.transferByTablet( + tabletInsertionEvent, + OpcDaSink.class.getName(), + LOGGER, + (tablet, isTableModel) -> handle.transfer(tablet)); } @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/PipeTabletEventTransferUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/PipeTabletEventTransferUtils.java new file mode 100644 index 000000000000..719c839e56bf --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/PipeTabletEventTransferUtils.java @@ -0,0 +1,97 @@ +/* + * 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.db.pipe.sink.util; + +import org.apache.iotdb.db.i18n.DataNodePipeMessages; +import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent; +import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent; +import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent; + +import org.apache.tsfile.write.record.Tablet; +import org.slf4j.Logger; + +public final class PipeTabletEventTransferUtils { + + private PipeTabletEventTransferUtils() {} + + public static void transferByTablet( + final TabletInsertionEvent tabletInsertionEvent, + final String holderMessage, + final Logger logger, + final ThrowingBiConsumer transferTablet) + throws Exception { + // PipeProcessor can change the type of TabletInsertionEvent. + if (!(tabletInsertionEvent instanceof PipeInsertNodeTabletInsertionEvent) + && !(tabletInsertionEvent instanceof PipeRawTabletInsertionEvent)) { + logger.warn( + DataNodePipeMessages + .THIS_CONNECTOR_ONLY_SUPPORT_PIPEINSERTNODETABLETINSERTIONEVENT_AND_PIPERAWTABLET, + tabletInsertionEvent); + return; + } + + if (tabletInsertionEvent instanceof PipeInsertNodeTabletInsertionEvent) { + transferTabletWrapper( + (PipeInsertNodeTabletInsertionEvent) tabletInsertionEvent, holderMessage, transferTablet); + } else { + transferTabletWrapper( + (PipeRawTabletInsertionEvent) tabletInsertionEvent, holderMessage, transferTablet); + } + } + + private static void transferTabletWrapper( + final PipeInsertNodeTabletInsertionEvent pipeInsertNodeTabletInsertionEvent, + final String holderMessage, + final ThrowingBiConsumer transferTablet) + throws Exception { + if (!pipeInsertNodeTabletInsertionEvent.increaseReferenceCount(holderMessage)) { + return; + } + try { + for (final Tablet tablet : pipeInsertNodeTabletInsertionEvent.convertToTablets()) { + transferTablet.accept(tablet, pipeInsertNodeTabletInsertionEvent.isTableModelEvent()); + } + } finally { + pipeInsertNodeTabletInsertionEvent.decreaseReferenceCount(holderMessage, false); + } + } + + private static void transferTabletWrapper( + final PipeRawTabletInsertionEvent pipeRawTabletInsertionEvent, + final String holderMessage, + final ThrowingBiConsumer transferTablet) + throws Exception { + if (!pipeRawTabletInsertionEvent.increaseReferenceCount(holderMessage)) { + return; + } + try { + transferTablet.accept( + pipeRawTabletInsertionEvent.convertToTablet(), + pipeRawTabletInsertionEvent.isTableModelEvent()); + } finally { + pipeRawTabletInsertionEvent.decreaseReferenceCount(holderMessage, false); + } + } + + @FunctionalInterface + public interface ThrowingBiConsumer { + void accept(final T t, final U u) throws E; + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManagerTest.java index dad190442513..6f40cd506671 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManagerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManagerTest.java @@ -19,7 +19,6 @@ package org.apache.iotdb.db.pipe.agent.task.subtask.sink; -import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin; import org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant; import org.apache.iotdb.commons.pipe.config.constant.SystemConstant; import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters; @@ -76,8 +75,7 @@ public void testCalculateSinkSubtaskNumUsesSingleThreadDefaultSinkAndSchemaRegio final Map singleThreadAttributes = new HashMap<>(); singleThreadAttributes.put( PipeSinkConstant.CONNECTOR_SERIALIZE_BY_REGION_KEY, Boolean.FALSE.toString()); - singleThreadAttributes.put( - PipeSinkConstant.CONNECTOR_KEY, BuiltinPipePlugin.OPC_UA_SINK.getPipePluginName()); + singleThreadAttributes.put(PipeSinkConstant.CONNECTOR_KEY, PipeSinkConstant.OPC_UA_SINK_NAME); Assert.assertEquals( 1, PipeSinkSubtaskManager.calculateSinkSubtaskNum( diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java index 11c433ac1b5e..f285287e16b6 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java @@ -27,7 +27,6 @@ import org.apache.iotdb.db.pipe.event.common.statement.PipeStatementInsertionEvent; import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent; import org.apache.iotdb.db.pipe.sink.protocol.legacy.IoTDBLegacyPipeSink; -import org.apache.iotdb.db.pipe.sink.protocol.opcua.OpcUaSink; import org.apache.iotdb.db.pipe.sink.protocol.thrift.async.IoTDBDataRegionAsyncSink; import org.apache.iotdb.db.pipe.sink.protocol.thrift.sync.IoTDBDataRegionSyncSink; import org.apache.iotdb.db.pipe.sink.protocol.websocket.WebSocketConnectorServer; @@ -47,7 +46,6 @@ import org.mockito.Mockito; import java.lang.reflect.Field; -import java.security.SecureRandom; import java.util.Arrays; import java.util.Collections; import java.util.HashMap; @@ -386,97 +384,6 @@ public void testWebSocketSinkDropDoesNotRequeueDroppedPipeEvents() { } } - @Test - public void testOpcUaSink() { - final List schemaList = - Arrays.asList( - new MeasurementSchema("s1", TSDataType.INT64), - new MeasurementSchema("s2", TSDataType.INT64)); - - final Tablet tablet = new Tablet("root.db.d1.vector6", schemaList, 100); - - long timestamp = System.currentTimeMillis(); - for (long row = 0; row < 100; row++) { - final int rowSize = tablet.getRowSize(); - tablet.addTimestamp(rowSize, timestamp); - for (int i = 0; i < 2; i++) { - tablet.addValue( - schemaList.get(i).getMeasurementName(), rowSize, new SecureRandom().nextLong()); - } - timestamp++; - } - - final List opcSchemaList = - Arrays.asList( - new MeasurementSchema("value1", TSDataType.INT64), - new MeasurementSchema("quality1", TSDataType.BOOLEAN)); - final Tablet qualityTablet = new Tablet("root.db.d1.vector6.s3", opcSchemaList, 100); - - timestamp = System.currentTimeMillis(); - for (long row = 0; row < 100; row++) { - final int rowSize = qualityTablet.getRowSize(); - qualityTablet.addTimestamp(rowSize, timestamp); - qualityTablet.addValue( - opcSchemaList.get(0).getMeasurementName(), rowSize, new SecureRandom().nextLong()); - qualityTablet.addValue(opcSchemaList.get(1).getMeasurementName(), rowSize, true); - timestamp++; - } - - try (final OpcUaSink qualityOPC = new OpcUaSink(); - final OpcUaSink normalOPC = new OpcUaSink()) { - final PipeTaskRuntimeConfiguration configuration = - new PipeTaskRuntimeConfiguration(new PipeTaskSinkRuntimeEnvironment("temp", 0, 1)); - qualityOPC.customize( - new PipeParameters( - new HashMap() { - { - put( - PipeSinkConstant.CONNECTOR_KEY, - BuiltinPipePlugin.OPC_UA_SINK.getPipePluginName()); - put(PipeSinkConstant.CONNECTOR_OPC_UA_WITH_QUALITY_KEY, "true"); - put(PipeSinkConstant.CONNECTOR_OPC_UA_VALUE_NAME_KEY, "value1"); - put(PipeSinkConstant.CONNECTOR_OPC_UA_QUALITY_NAME_KEY, "quality1"); - } - }), - configuration); - normalOPC.customize( - new PipeParameters( - new HashMap() { - { - put( - PipeSinkConstant.CONNECTOR_KEY, - BuiltinPipePlugin.OPC_UA_SINK.getPipePluginName()); - } - }), - configuration); - final PipeRawTabletInsertionEvent event = - new PipeRawTabletInsertionEvent( - false, "root.db", "db", "root.db", tablet, false, "pipe", 0L, null, null, false); - event.increaseReferenceCount(""); - normalOPC.transfer(event); - // Shall not throw - qualityOPC.transfer(event); - event.decreaseReferenceCount("", false); - - qualityOPC.transfer( - new PipeRawTabletInsertionEvent( - false, - "root.db", - "db", - "root.db", - qualityTablet, - false, - "pipe", - 0L, - null, - null, - false)); - - } catch (Exception e) { - Assert.fail(); - } - } - @Test public void testWriteBackSinkTargetDatabaseValidation() throws Exception { assertWriteBackSinkTargetDatabaseValid("target"); diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/BuiltinPipePlugin.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/BuiltinPipePlugin.java index 76b45d1e7989..37b4868243ab 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/BuiltinPipePlugin.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/BuiltinPipePlugin.java @@ -39,7 +39,6 @@ import org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.iotdb.thrift.IoTDBThriftSslSink; import org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.iotdb.thrift.IoTDBThriftSyncSink; import org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.opcda.OpcDaSink; -import org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.opcua.OpcUaSink; import org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.websocket.WebSocketSink; import org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.writeback.WriteBackSink; import org.apache.iotdb.commons.pipe.agent.plugin.builtin.source.donothing.DoNothingSource; @@ -92,7 +91,6 @@ public enum BuiltinPipePlugin { PIPE_CONSENSUS_ASYNC_CONNECTOR("pipe-consensus-async-connector", IoTConsensusV2AsyncSink.class), WEBSOCKET_CONNECTOR("websocket-connector", WebSocketSink.class), - OPC_UA_CONNECTOR("opc-ua-connector", OpcUaSink.class), OPC_DA_CONNECTOR("opc-da-connector", OpcDaSink.class), WRITE_BACK_CONNECTOR("write-back-connector", WriteBackSink.class), @@ -104,7 +102,6 @@ public enum BuiltinPipePlugin { IOTDB_LEGACY_PIPE_SINK("iotdb-legacy-pipe-sink", IoTDBLegacyPipeSink.class), IOTDB_AIR_GAP_SINK("iotdb-air-gap-sink", IoTDBAirGapSink.class), WEBSOCKET_SINK("websocket-sink", WebSocketSink.class), - OPC_UA_SINK("opc-ua-sink", OpcUaSink.class), OPC_DA_SINK("opc-da-sink", OpcDaSink.class), WRITE_BACK_SINK("write-back-sink", WriteBackSink.class), SUBSCRIPTION_SINK("subscription-sink", DoNothingSink.class), @@ -175,7 +172,6 @@ public String getClassName() { IOTDB_LEGACY_PIPE_CONNECTOR.getPipePluginName().toUpperCase(), IOTDB_AIR_GAP_CONNECTOR.getPipePluginName().toUpperCase(), WEBSOCKET_CONNECTOR.getPipePluginName().toUpperCase(), - OPC_UA_CONNECTOR.getPipePluginName().toUpperCase(), OPC_DA_CONNECTOR.getPipePluginName().toUpperCase(), WRITE_BACK_CONNECTOR.getPipePluginName().toUpperCase(), IOT_CONSENSUS_V2_ASYNC_CONNECTOR.getPipePluginName().toUpperCase(), @@ -185,7 +181,6 @@ public String getClassName() { IOTDB_THRIFT_ASYNC_SINK.getPipePluginName().toUpperCase(), IOTDB_LEGACY_PIPE_SINK.getPipePluginName().toUpperCase(), WEBSOCKET_SINK.getPipePluginName().toUpperCase(), - OPC_UA_SINK.getPipePluginName().toUpperCase(), OPC_DA_SINK.getPipePluginName().toUpperCase(), SUBSCRIPTION_SINK.getPipePluginName().toUpperCase(), IOT_CONSENSUS_V2_ASYNC_SINK.getPipePluginName().toUpperCase(), diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/sink/opcua/OpcUaSink.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/sink/opcua/OpcUaSink.java deleted file mode 100644 index 462fb099f712..000000000000 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/sink/opcua/OpcUaSink.java +++ /dev/null @@ -1,34 +0,0 @@ -/* - * 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.agent.plugin.builtin.sink.opcua; - -import org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.PlaceholderSink; -import org.apache.iotdb.pipe.api.annotation.TableModel; -import org.apache.iotdb.pipe.api.annotation.TreeModel; - -/** - * This class is a placeholder and should not be initialized. It represents the OPC UA connector. - * There is a real implementation in the server module but cannot be imported here. The pipe agent - * in the server module will replace this class with the real implementation when initializing the - * OPC UA connector. - */ -@TreeModel -@TableModel -public class OpcUaSink extends PlaceholderSink {} 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 2db1bd0c06b4..dba46ce9fb51 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 @@ -41,6 +41,9 @@ public class PipeSinkConstant { public static final String CONNECTOR_KEY = "connector"; public static final String SINK_KEY = "sink"; + public static final String OPC_UA_CONNECTOR_NAME = "opc-ua-connector"; + public static final String OPC_UA_SINK_NAME = "opc-ua-sink"; + public static final String CONNECTOR_IOTDB_IP_KEY = "connector.ip"; public static final String SINK_IOTDB_IP_KEY = "sink.ip"; public static final String CONNECTOR_IOTDB_HOST_KEY = "connector.host"; @@ -71,8 +74,8 @@ public class PipeSinkConstant { public static final Set SINGLE_THREAD_DEFAULT_SINK = new HashSet<>( Arrays.asList( - BuiltinPipePlugin.OPC_UA_SINK.getPipePluginName(), - BuiltinPipePlugin.OPC_UA_CONNECTOR.getPipePluginName(), + OPC_UA_SINK_NAME, + OPC_UA_CONNECTOR_NAME, BuiltinPipePlugin.OPC_DA_SINK.getPipePluginName(), BuiltinPipePlugin.OPC_DA_CONNECTOR.getPipePluginName())); diff --git a/example/pipe-opc-ua-sink/pom.xml b/library-pipe/opc-ua-sink-example/pom.xml similarity index 95% rename from example/pipe-opc-ua-sink/pom.xml rename to library-pipe/opc-ua-sink-example/pom.xml index 0627a8540963..921de8aba3ff 100644 --- a/example/pipe-opc-ua-sink/pom.xml +++ b/library-pipe/opc-ua-sink-example/pom.xml @@ -22,17 +22,16 @@ org.apache.iotdb - iotdb-examples + library-pipe 2.0.11-SNAPSHOT 4.0.0 pipe-opc-ua-sink-example - IoTDB: Example: Pipe: OPCUA Sink + IoTDB: Library-Pipe: OPC UA Sink Example org.eclipse.milo milo-sdk-client - ${milo.version} org.eclipse.milo diff --git a/example/pipe-opc-ua-sink/src/main/java/org/apache/iotdb/opcua/ClientExample.java b/library-pipe/opc-ua-sink-example/src/main/java/org/apache/iotdb/opcua/ClientExample.java similarity index 100% rename from example/pipe-opc-ua-sink/src/main/java/org/apache/iotdb/opcua/ClientExample.java rename to library-pipe/opc-ua-sink-example/src/main/java/org/apache/iotdb/opcua/ClientExample.java diff --git a/example/pipe-opc-ua-sink/src/main/java/org/apache/iotdb/opcua/ClientExampleRunner.java b/library-pipe/opc-ua-sink-example/src/main/java/org/apache/iotdb/opcua/ClientExampleRunner.java similarity index 100% rename from example/pipe-opc-ua-sink/src/main/java/org/apache/iotdb/opcua/ClientExampleRunner.java rename to library-pipe/opc-ua-sink-example/src/main/java/org/apache/iotdb/opcua/ClientExampleRunner.java diff --git a/example/pipe-opc-ua-sink/src/main/java/org/apache/iotdb/opcua/ClientTest.java b/library-pipe/opc-ua-sink-example/src/main/java/org/apache/iotdb/opcua/ClientTest.java similarity index 100% rename from example/pipe-opc-ua-sink/src/main/java/org/apache/iotdb/opcua/ClientTest.java rename to library-pipe/opc-ua-sink-example/src/main/java/org/apache/iotdb/opcua/ClientTest.java diff --git a/example/pipe-opc-ua-sink/src/main/java/org/apache/iotdb/opcua/IoTDBKeyStoreLoaderClient.java b/library-pipe/opc-ua-sink-example/src/main/java/org/apache/iotdb/opcua/IoTDBKeyStoreLoaderClient.java similarity index 100% rename from example/pipe-opc-ua-sink/src/main/java/org/apache/iotdb/opcua/IoTDBKeyStoreLoaderClient.java rename to library-pipe/opc-ua-sink-example/src/main/java/org/apache/iotdb/opcua/IoTDBKeyStoreLoaderClient.java diff --git a/library-pipe/opc-ua-sink/README.md b/library-pipe/opc-ua-sink/README.md new file mode 100644 index 000000000000..90b750698d2c --- /dev/null +++ b/library-pipe/opc-ua-sink/README.md @@ -0,0 +1,73 @@ + + +# OPC UA Sink Plugin + +`opc-ua-sink` is an external Pipe Sink plugin that publishes IoTDB data through OPC UA or writes it to an external OPC UA server. Eclipse Milo and the plugin's other runtime dependencies are bundled in its fat JAR instead of the default IoTDB server package. + +## Build + +Run from the repository root: + +```bash +mvn clean package -pl library-pipe/opc-ua-sink -am -DskipTests +``` + +The plugin artifact is generated at: + +`library-pipe/opc-ua-sink/target/opc-ua-sink--jar-with-dependencies.jar` + +It is also included under `ext/pipe` in the standalone `library-pipe` distribution ZIP. + +## Register + +Upload the fat JAR to a URI accessible to IoTDB and register the plugin before creating an OPC UA Pipe: + +```sql +CREATE PIPEPLUGIN `opc-ua-sink` +AS 'org.apache.iotdb.pipe.plugin.sink.opcua.OpcUaSink' +USING URI 'file:///path/to/opc-ua-sink--jar-with-dependencies.jar'; +``` + +`opc-ua-connector` is the legacy-compatible alias. Register it separately when existing Pipe definitions use that name: + +```sql +CREATE PIPEPLUGIN `opc-ua-connector` +AS 'org.apache.iotdb.pipe.plugin.sink.opcua.OpcUaSink' +USING URI 'file:///path/to/opc-ua-sink--jar-with-dependencies.jar'; +``` + +## Example + +The following Pipe starts an OPC UA server with no message security: + +```sql +CREATE PIPE opc_ua_pipe +WITH SINK ( + 'sink' = 'opc-ua-sink', + 'sink.opcua.model' = 'client-server', + 'sink.opcua.security-policy' = 'NONE', + 'sink.opcua.tcp.port' = '12686', + 'sink.opcua.https.port' = '8443' +); +``` + +The client example is located in `library-pipe/opc-ua-sink-example`. diff --git a/library-pipe/opc-ua-sink/pom.xml b/library-pipe/opc-ua-sink/pom.xml new file mode 100644 index 000000000000..2daa7352b695 --- /dev/null +++ b/library-pipe/opc-ua-sink/pom.xml @@ -0,0 +1,159 @@ + + + + 4.0.0 + + org.apache.iotdb + library-pipe + 2.0.11-SNAPSHOT + + opc-ua-sink + IoTDB: Library-Pipe: OPC UA Sink + + 17 + 17 + UTF-8 + + + + org.apache.iotdb + iotdb-server + ${project.version} + provided + + + org.apache.iotdb + node-commons + ${project.version} + provided + + + org.apache.iotdb + pipe-api + ${project.version} + provided + + + org.apache.tsfile + tsfile + provided + + + org.apache.tsfile + common + ${tsfile.version} + provided + + + com.google.guava + guava + provided + + + org.slf4j + slf4j-api + provided + + + com.google.code.findbugs + jsr305 + provided + + + org.eclipse.milo + milo-stack-core + + + org.eclipse.milo + milo-sdk-core + + + org.eclipse.milo + milo-transport + + + org.eclipse.milo + milo-sdk-client + + + org.eclipse.milo + milo-sdk-server + + + org.bouncycastle + bcprov-jdk18on + + + junit + junit + test + + + org.mockito + mockito-core + test + + + + + + org.apache.maven.plugins + maven-surefire-plugin + + + ${maven.multiModuleProjectDirectory}/iotdb-core/datanode/src/test/resources + + false + + + + org.apache.maven.plugins + maven-assembly-plugin + + + jar-with-dependencies + + true + + + + make-assembly + package + + single + + + + + + org.apache.maven.plugins + maven-dependency-plugin + + + + org.apache.tsfile:common + + + + + + diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSink.java b/library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/OpcUaSink.java similarity index 89% rename from iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSink.java rename to library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/OpcUaSink.java index 1263603bd491..f270b28cc426 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSink.java +++ b/library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/OpcUaSink.java @@ -17,18 +17,13 @@ * under the License. */ -package org.apache.iotdb.db.pipe.sink.protocol.opcua; +package org.apache.iotdb.pipe.plugin.sink.opcua; import org.apache.iotdb.commons.consensus.DataRegionId; import org.apache.iotdb.commons.utils.PathUtils; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.i18n.DataNodePipeMessages; -import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent; -import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent; -import org.apache.iotdb.db.pipe.sink.protocol.opcua.client.ClientRunner; -import org.apache.iotdb.db.pipe.sink.protocol.opcua.client.IoTDBOpcUaClient; -import org.apache.iotdb.db.pipe.sink.protocol.opcua.server.OpcUaNameSpace; -import org.apache.iotdb.db.pipe.sink.protocol.opcua.server.OpcUaServerBuilder; +import org.apache.iotdb.db.pipe.sink.util.PipeTabletEventTransferUtils; import org.apache.iotdb.db.storageengine.StorageEngine; import org.apache.iotdb.db.storageengine.dataregion.DataRegion; import org.apache.iotdb.pipe.api.PipeConnector; @@ -40,10 +35,13 @@ import org.apache.iotdb.pipe.api.event.Event; import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent; import org.apache.iotdb.pipe.api.exception.PipeException; +import org.apache.iotdb.pipe.plugin.sink.opcua.client.ClientRunner; +import org.apache.iotdb.pipe.plugin.sink.opcua.client.IoTDBOpcUaClient; +import org.apache.iotdb.pipe.plugin.sink.opcua.server.OpcUaNameSpace; +import org.apache.iotdb.pipe.plugin.sink.opcua.server.OpcUaServerBuilder; import org.apache.tsfile.common.conf.TSFileConfig; import org.apache.tsfile.utils.Pair; -import org.apache.tsfile.write.record.Tablet; import org.eclipse.milo.opcua.sdk.client.identity.AnonymousProvider; import org.eclipse.milo.opcua.sdk.client.identity.IdentityProvider; import org.eclipse.milo.opcua.sdk.client.identity.UsernameProvider; @@ -163,9 +161,11 @@ public class OpcUaSink implements PipeConnector { // Inner server private @Nullable OpcUaNameSpace nameSpace; + private boolean serverRegistered; // Outer server private @Nullable IoTDBOpcUaClient client; + private boolean clientRegistered; @Override public void validate(final PipeParameterValidator validator) throws Exception { @@ -372,6 +372,7 @@ private void customizeServer(final PipeParameters parameters) { }) .getRight(); SERVER_KEY_TO_REFERENCE_COUNT_AND_NAME_SPACE_MAP.get(serverKey).getLeft().incrementAndGet(); + serverRegistered = true; } } @@ -452,6 +453,7 @@ private void customizeClient(final PipeParameters parameters) { }) .getRight(); CLIENT_KEY_TO_REFERENCE_COUNT_AND_CLIENT_MAP.get(nodeUrl).getLeft().incrementAndGet(); + clientRegistered = true; } } @@ -504,8 +506,9 @@ public void transfer(final Event event) throws Exception { @Override public void transfer(final TabletInsertionEvent tabletInsertionEvent) throws Exception { - transferByTablet( + PipeTabletEventTransferUtils.transferByTablet( tabletInsertionEvent, + OpcUaSink.class.getName(), LOGGER, (tablet, isTableModel) -> { if (Objects.nonNull(nameSpace)) { @@ -518,75 +521,14 @@ public void transfer(final TabletInsertionEvent tabletInsertionEvent) throws Exc }); } - public static void transferByTablet( - final TabletInsertionEvent tabletInsertionEvent, - final Logger logger, - final ThrowingBiConsumer transferTablet) - throws Exception { - // PipeProcessor can change the type of TabletInsertionEvent - if (!(tabletInsertionEvent instanceof PipeInsertNodeTabletInsertionEvent) - && !(tabletInsertionEvent instanceof PipeRawTabletInsertionEvent)) { - logger.warn( - DataNodePipeMessages - .THIS_CONNECTOR_ONLY_SUPPORT_PIPEINSERTNODETABLETINSERTIONEVENT_AND_PIPERAWTABLET, - tabletInsertionEvent); - return; - } - - if (tabletInsertionEvent instanceof PipeInsertNodeTabletInsertionEvent) { - transferTabletWrapper( - (PipeInsertNodeTabletInsertionEvent) tabletInsertionEvent, transferTablet); - } else { - transferTabletWrapper((PipeRawTabletInsertionEvent) tabletInsertionEvent, transferTablet); - } - } - - private static void transferTabletWrapper( - final PipeInsertNodeTabletInsertionEvent pipeInsertNodeTabletInsertionEvent, - final ThrowingBiConsumer transferTablet) - throws Exception { - // We increase the reference count for this event to determine if the event may be released. - if (!pipeInsertNodeTabletInsertionEvent.increaseReferenceCount(OpcUaSink.class.getName())) { - return; - } - try { - for (final Tablet tablet : pipeInsertNodeTabletInsertionEvent.convertToTablets()) { - transferTablet.accept(tablet, pipeInsertNodeTabletInsertionEvent.isTableModelEvent()); - } - } finally { - pipeInsertNodeTabletInsertionEvent.decreaseReferenceCount(OpcUaSink.class.getName(), false); - } - } - - private static void transferTabletWrapper( - final PipeRawTabletInsertionEvent pipeRawTabletInsertionEvent, - final ThrowingBiConsumer transferTablet) - throws Exception { - // We increase the reference count for this event to determine if the event may be released. - if (!pipeRawTabletInsertionEvent.increaseReferenceCount(OpcUaSink.class.getName())) { - return; - } - try { - transferTablet.accept( - pipeRawTabletInsertionEvent.convertToTablet(), - pipeRawTabletInsertionEvent.isTableModelEvent()); - } finally { - pipeRawTabletInsertionEvent.decreaseReferenceCount(OpcUaSink.class.getName(), false); - } - } - - @FunctionalInterface - public interface ThrowingBiConsumer { - void accept(final T t, final U u) throws E; - } - @Override public void close() throws Exception { - if (serverKey != null) { + if (serverKey != null && serverRegistered) { synchronized (SERVER_KEY_TO_REFERENCE_COUNT_AND_NAME_SPACE_MAP) { final Pair pair = SERVER_KEY_TO_REFERENCE_COUNT_AND_NAME_SPACE_MAP.get(serverKey); if (pair == null) { + serverRegistered = false; return; } @@ -597,14 +539,16 @@ public void close() throws Exception { SERVER_KEY_TO_REFERENCE_COUNT_AND_NAME_SPACE_MAP.remove(serverKey); } } + serverRegistered = false; } } - if (nodeUrl != null) { + if (nodeUrl != null && clientRegistered) { synchronized (CLIENT_KEY_TO_REFERENCE_COUNT_AND_CLIENT_MAP) { final Pair pair = CLIENT_KEY_TO_REFERENCE_COUNT_AND_CLIENT_MAP.get(nodeUrl); if (pair == null) { + clientRegistered = false; return; } @@ -615,6 +559,7 @@ public void close() throws Exception { CLIENT_KEY_TO_REFERENCE_COUNT_AND_CLIENT_MAP.remove(nodeUrl); } } + clientRegistered = false; } } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/ClientRunner.java b/library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/client/ClientRunner.java similarity index 99% rename from iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/ClientRunner.java rename to library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/client/ClientRunner.java index 64757596d16f..4854d02ce305 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/ClientRunner.java +++ b/library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/client/ClientRunner.java @@ -17,7 +17,7 @@ * under the License. */ -package org.apache.iotdb.db.pipe.sink.protocol.opcua.client; +package org.apache.iotdb.pipe.plugin.sink.opcua.client; import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.pipe.api.exception.PipeException; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBKeyStoreLoaderClient.java b/library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/client/IoTDBKeyStoreLoaderClient.java similarity index 98% rename from iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBKeyStoreLoaderClient.java rename to library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/client/IoTDBKeyStoreLoaderClient.java index 455018695dc2..c522f8e7b835 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBKeyStoreLoaderClient.java +++ b/library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/client/IoTDBKeyStoreLoaderClient.java @@ -17,7 +17,7 @@ * under the License. */ -package org.apache.iotdb.db.pipe.sink.protocol.opcua.client; +package org.apache.iotdb.pipe.plugin.sink.opcua.client; import org.apache.iotdb.db.i18n.DataNodePipeMessages; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClient.java b/library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/client/IoTDBOpcUaClient.java similarity index 97% rename from iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClient.java rename to library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/client/IoTDBOpcUaClient.java index 8c6c30dfebab..1c59937d05ae 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClient.java +++ b/library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/client/IoTDBOpcUaClient.java @@ -17,14 +17,14 @@ * under the License. */ -package org.apache.iotdb.db.pipe.sink.protocol.opcua.client; +package org.apache.iotdb.pipe.plugin.sink.opcua.client; import org.apache.iotdb.commons.pipe.resource.log.PipeLogger; import org.apache.iotdb.commons.utils.TestOnly; import org.apache.iotdb.db.i18n.DataNodePipeMessages; -import org.apache.iotdb.db.pipe.sink.protocol.opcua.OpcUaSink; -import org.apache.iotdb.db.pipe.sink.protocol.opcua.server.OpcUaNameSpace; import org.apache.iotdb.pipe.api.exception.PipeException; +import org.apache.iotdb.pipe.plugin.sink.opcua.OpcUaSink; +import org.apache.iotdb.pipe.plugin.sink.opcua.server.OpcUaNameSpace; import org.apache.tsfile.common.constant.TsFileConstant; import org.apache.tsfile.enums.TSDataType; @@ -67,8 +67,8 @@ import java.util.Set; 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.apache.iotdb.pipe.plugin.sink.opcua.server.OpcUaNameSpace.convertToOpcDataType; +import static org.apache.iotdb.pipe.plugin.sink.opcua.server.OpcUaNameSpace.timestampToUtc; import static org.eclipse.milo.opcua.stack.core.StatusCodes.Bad_Timeout; public class IoTDBOpcUaClient { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaKeyStoreLoader.java b/library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/server/OpcUaKeyStoreLoader.java similarity index 98% rename from iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaKeyStoreLoader.java rename to library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/server/OpcUaKeyStoreLoader.java index 2e27ffc22777..838624403d19 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaKeyStoreLoader.java +++ b/library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/server/OpcUaKeyStoreLoader.java @@ -17,7 +17,7 @@ * under the License. */ -package org.apache.iotdb.db.pipe.sink.protocol.opcua.server; +package org.apache.iotdb.pipe.plugin.sink.opcua.server; import org.apache.iotdb.commons.utils.FileUtils; import org.apache.iotdb.db.i18n.DataNodePipeMessages; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpace.java b/library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/server/OpcUaNameSpace.java similarity index 99% rename from iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpace.java rename to library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/server/OpcUaNameSpace.java index 1d6262c2c6a1..551bf7abdb15 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpace.java +++ b/library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/server/OpcUaNameSpace.java @@ -17,7 +17,7 @@ * under the License. */ -package org.apache.iotdb.db.pipe.sink.protocol.opcua.server; +package org.apache.iotdb.pipe.plugin.sink.opcua.server; import org.apache.iotdb.commons.exception.pipe.PipeRuntimeCriticalException; import org.apache.iotdb.commons.exception.pipe.PipeRuntimeNonCriticalException; @@ -25,10 +25,10 @@ import org.apache.iotdb.commons.queryengine.utils.DateTimeUtils; import org.apache.iotdb.commons.queryengine.utils.TimestampPrecisionUtils; import org.apache.iotdb.db.i18n.DataNodePipeMessages; -import org.apache.iotdb.db.pipe.sink.protocol.opcua.OpcUaSink; import org.apache.iotdb.db.pipe.sink.util.sorter.PipeTableModelTabletEventSorter; import org.apache.iotdb.db.pipe.sink.util.sorter.PipeTreeModelTabletEventSorter; import org.apache.iotdb.pipe.api.event.Event; +import org.apache.iotdb.pipe.plugin.sink.opcua.OpcUaSink; import org.apache.tsfile.common.constant.TsFileConstant; import org.apache.tsfile.enums.ColumnCategory; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaServerBuilder.java b/library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/server/OpcUaServerBuilder.java similarity index 99% rename from iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaServerBuilder.java rename to library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/server/OpcUaServerBuilder.java index 5b7d820f8f7b..1ed4038775b0 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaServerBuilder.java +++ b/library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/server/OpcUaServerBuilder.java @@ -17,7 +17,7 @@ * under the License. */ -package org.apache.iotdb.db.pipe.sink.protocol.opcua.server; +package org.apache.iotdb.pipe.plugin.sink.opcua.server; import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.pipe.api.exception.PipeException; diff --git a/library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/OpcUaSinkTest.java b/library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/OpcUaSinkTest.java new file mode 100644 index 000000000000..4d4a8a8c7e1a --- /dev/null +++ b/library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/OpcUaSinkTest.java @@ -0,0 +1,285 @@ +/* + * 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.pipe.plugin.sink.opcua; + +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.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent; +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.PipeException; +import org.apache.iotdb.pipe.api.exception.PipeParameterNotValidException; + +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.write.record.Tablet; +import org.apache.tsfile.write.schema.IMeasurementSchema; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.eclipse.milo.opcua.stack.core.types.builtin.StatusCode; +import org.junit.Assert; +import org.junit.Test; + +import java.io.File; +import java.io.IOException; +import java.net.ServerSocket; +import java.security.SecureRandom; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; + +public class OpcUaSinkTest { + + @Test + public void testValidateClientServerOnlyOptions() { + assertValidationFailure( + "must be client-server", + PipeSinkConstant.CONNECTOR_OPC_UA_WITH_QUALITY_KEY, + "true", + PipeSinkConstant.CONNECTOR_OPC_UA_MODEL_KEY, + PipeSinkConstant.CONNECTOR_OPC_UA_MODEL_PUB_SUB_VALUE); + + assertValidationFailure( + "must be client-server", + PipeSinkConstant.CONNECTOR_OPC_UA_NODE_URL_KEY, + "opc.tcp://127.0.0.1:12686/iotdb", + PipeSinkConstant.CONNECTOR_OPC_UA_MODEL_KEY, + PipeSinkConstant.CONNECTOR_OPC_UA_MODEL_PUB_SUB_VALUE); + } + + @Test + public void testCustomizeQualityParameters() throws Exception { + final int[] qualityPorts = findTwoFreePorts(); + final int[] normalPorts = findTwoFreePorts(); + try (final OpcUaSink qualitySink = new OpcUaSink(); + final OpcUaSink normalSink = new OpcUaSink()) { + qualitySink.customize( + createParameters( + PipeSinkConstant.CONNECTOR_KEY, + PipeSinkConstant.OPC_UA_SINK_NAME, + PipeSinkConstant.CONNECTOR_OPC_UA_WITH_QUALITY_KEY, + "true", + PipeSinkConstant.CONNECTOR_OPC_UA_VALUE_NAME_KEY, + "value1", + PipeSinkConstant.CONNECTOR_OPC_UA_QUALITY_NAME_KEY, + "quality1", + PipeSinkConstant.CONNECTOR_OPC_UA_DEFAULT_QUALITY_KEY, + "BAD", + PipeSinkConstant.CONNECTOR_OPC_UA_SECURITY_POLICY_KEY, + "None", + PipeSinkConstant.CONNECTOR_OPC_UA_TCP_BIND_PORT_KEY, + Integer.toString(qualityPorts[0]), + PipeSinkConstant.CONNECTOR_OPC_UA_HTTPS_BIND_PORT_KEY, + Integer.toString(qualityPorts[1]), + PipeSinkConstant.CONNECTOR_OPC_UA_SECURITY_DIR_KEY, + createSecurityDir()), + createRuntimeConfiguration()); + normalSink.customize( + createOpcUaServerParameters(normalPorts[0], normalPorts[1], "root", "root"), + createRuntimeConfiguration()); + + Assert.assertEquals("value1", qualitySink.getValueName()); + Assert.assertEquals("quality1", qualitySink.getQualityName()); + Assert.assertEquals(StatusCode.BAD, qualitySink.getDefaultQuality()); + Assert.assertNull(normalSink.getValueName()); + Assert.assertNull(normalSink.getQualityName()); + Assert.assertEquals(StatusCode.GOOD, normalSink.getDefaultQuality()); + } + } + + @Test + public void testTransferWithQualityAndNormalTablets() throws Exception { + final List schemaList = + Arrays.asList( + new MeasurementSchema("s1", TSDataType.INT64), + new MeasurementSchema("s2", TSDataType.INT64)); + final Tablet tablet = new Tablet("root.db.d1.vector6", schemaList, 100); + long timestamp = System.currentTimeMillis(); + for (long row = 0; row < 100; row++) { + final int rowSize = tablet.getRowSize(); + tablet.addTimestamp(rowSize, timestamp++); + for (int i = 0; i < 2; i++) { + tablet.addValue( + schemaList.get(i).getMeasurementName(), rowSize, new SecureRandom().nextLong()); + } + } + + final List opcSchemaList = + Arrays.asList( + new MeasurementSchema("value1", TSDataType.INT64), + new MeasurementSchema("quality1", TSDataType.BOOLEAN)); + final Tablet qualityTablet = new Tablet("root.db.d1.vector6.s3", opcSchemaList, 100); + timestamp = System.currentTimeMillis(); + for (long row = 0; row < 100; row++) { + final int rowSize = qualityTablet.getRowSize(); + qualityTablet.addTimestamp(rowSize, timestamp++); + qualityTablet.addValue( + opcSchemaList.get(0).getMeasurementName(), rowSize, new SecureRandom().nextLong()); + qualityTablet.addValue(opcSchemaList.get(1).getMeasurementName(), rowSize, true); + } + + final int[] qualityPorts = findTwoFreePorts(); + final int[] normalPorts = findTwoFreePorts(); + try (final OpcUaSink qualitySink = new OpcUaSink(); + final OpcUaSink normalSink = new OpcUaSink()) { + qualitySink.customize( + createQualityServerParameters(qualityPorts[0], qualityPorts[1]), + createRuntimeConfiguration()); + normalSink.customize( + createOpcUaServerParameters(normalPorts[0], normalPorts[1], "root", "root"), + createRuntimeConfiguration()); + + final PipeRawTabletInsertionEvent event = + new PipeRawTabletInsertionEvent( + false, "root.db", "db", "root.db", tablet, false, "pipe", 0L, null, null, false); + event.increaseReferenceCount(""); + normalSink.transfer(event); + qualitySink.transfer(event); + event.decreaseReferenceCount("", false); + + qualitySink.transfer( + new PipeRawTabletInsertionEvent( + false, + "root.db", + "db", + "root.db", + qualityTablet, + false, + "pipe", + 0L, + null, + null, + false)); + } + } + + @Test + public void testSharedServerLifecycle() throws Exception { + final int[] ports = findTwoFreePorts(); + final PipeTaskRuntimeConfiguration configuration = createRuntimeConfiguration(); + final PipeParameters parameters = + createOpcUaServerParameters(ports[0], ports[1], "root", "root"); + final PipeParameters conflictingParameters = + createOpcUaServerParameters(ports[0], ports[1], "root", "conflict"); + + try (final OpcUaSink firstSink = new OpcUaSink(); + final OpcUaSink secondSink = new OpcUaSink()) { + firstSink.customize(parameters, configuration); + secondSink.customize(parameters, configuration); + + assertCustomizeFailure(conflictingParameters, configuration); + secondSink.close(); + secondSink.close(); + assertCustomizeFailure(conflictingParameters, configuration); + } + } + + private static void assertCustomizeFailure( + final PipeParameters parameters, final PipeTaskRuntimeConfiguration configuration) { + try (final OpcUaSink conflictingSink = new OpcUaSink()) { + final PipeException exception = + Assert.assertThrows( + PipeException.class, () -> conflictingSink.customize(parameters, configuration)); + Assert.assertTrue(exception.getMessage(), exception.getMessage().contains("reject reusing")); + } catch (final Exception e) { + throw new AssertionError(e); + } + } + + private static void assertValidationFailure( + final String expectedMessagePart, final String... keyValues) { + final PipeParameterNotValidException exception = + Assert.assertThrows( + PipeParameterNotValidException.class, + () -> + new OpcUaSink().validate(new PipeParameterValidator(createParameters(keyValues)))); + Assert.assertTrue(exception.getMessage(), exception.getMessage().contains(expectedMessagePart)); + } + + private static PipeParameters createQualityServerParameters( + final int tcpPort, final int httpsPort) { + final Map attributes = + new HashMap<>( + createOpcUaServerParameters(tcpPort, httpsPort, "root", "root").getAttribute()); + attributes.put(PipeSinkConstant.CONNECTOR_OPC_UA_WITH_QUALITY_KEY, "true"); + attributes.put(PipeSinkConstant.CONNECTOR_OPC_UA_VALUE_NAME_KEY, "value1"); + attributes.put(PipeSinkConstant.CONNECTOR_OPC_UA_QUALITY_NAME_KEY, "quality1"); + return new PipeParameters(attributes); + } + + private static PipeParameters createOpcUaServerParameters( + final int tcpPort, final int httpsPort, final String user, final String password) { + return createParameters( + PipeSinkConstant.CONNECTOR_KEY, + PipeSinkConstant.OPC_UA_SINK_NAME, + PipeSinkConstant.CONNECTOR_OPC_UA_TCP_BIND_PORT_KEY, + Integer.toString(tcpPort), + PipeSinkConstant.CONNECTOR_OPC_UA_HTTPS_BIND_PORT_KEY, + Integer.toString(httpsPort), + PipeSinkConstant.CONNECTOR_OPC_UA_SECURITY_POLICY_KEY, + "None", + PipeSinkConstant.CONNECTOR_IOTDB_USER_KEY, + user, + PipeSinkConstant.CONNECTOR_IOTDB_PASSWORD_KEY, + password, + PipeSinkConstant.CONNECTOR_OPC_UA_SECURITY_DIR_KEY, + createSecurityDir(), + PipeSinkConstant.CONNECTOR_OPC_UA_DEBOUNCE_TIME_MS_KEY, + "1"); + } + + private static PipeParameters createParameters(final String... keyValues) { + final Map attributes = new HashMap<>(); + for (int i = 0; i < keyValues.length; i += 2) { + attributes.put(keyValues[i], keyValues[i + 1]); + } + return new PipeParameters(attributes); + } + + private static PipeTaskRuntimeConfiguration createRuntimeConfiguration() { + return new PipeTaskRuntimeConfiguration(new PipeTaskSinkRuntimeEnvironment("temp", 0, 1)); + } + + private static String createSecurityDir() { + return new File( + "target" + + File.separatorChar + + "opc-ua-sink-test" + + File.separatorChar + + UUID.randomUUID()) + .getAbsolutePath(); + } + + private static int[] findTwoFreePorts() throws IOException { + final int firstPort = findFreePort(); + int secondPort; + do { + secondPort = findFreePort(); + } while (secondPort == firstPort); + return new int[] {firstPort, secondPort}; + } + + private static int findFreePort() throws IOException { + try (final ServerSocket socket = new ServerSocket(0)) { + return socket.getLocalPort(); + } + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/ClientRunnerTest.java b/library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/client/ClientRunnerTest.java similarity index 98% rename from iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/ClientRunnerTest.java rename to library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/client/ClientRunnerTest.java index 9ccffbfc4507..5f9c027866e0 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/ClientRunnerTest.java +++ b/library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/client/ClientRunnerTest.java @@ -17,7 +17,7 @@ * under the License. */ -package org.apache.iotdb.db.pipe.sink.protocol.opcua.client; +package org.apache.iotdb.pipe.plugin.sink.opcua.client; import org.apache.iotdb.pipe.api.exception.PipeException; diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClientTest.java b/library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/client/IoTDBOpcUaClientTest.java similarity index 98% rename from iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClientTest.java rename to library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/client/IoTDBOpcUaClientTest.java index 5cb881939d40..dac3acca0f57 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClientTest.java +++ b/library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/client/IoTDBOpcUaClientTest.java @@ -17,10 +17,10 @@ * under the License. */ -package org.apache.iotdb.db.pipe.sink.protocol.opcua.client; +package org.apache.iotdb.pipe.plugin.sink.opcua.client; -import org.apache.iotdb.db.pipe.sink.protocol.opcua.OpcUaSink; import org.apache.iotdb.pipe.api.exception.PipeException; +import org.apache.iotdb.pipe.plugin.sink.opcua.OpcUaSink; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.write.record.Tablet; diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaServerBuilderTest.java b/library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/server/OpcUaServerBuilderTest.java similarity index 99% rename from iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaServerBuilderTest.java rename to library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/server/OpcUaServerBuilderTest.java index 8c21c6632ce8..1942f86515e7 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaServerBuilderTest.java +++ b/library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/server/OpcUaServerBuilderTest.java @@ -17,7 +17,7 @@ * under the License. */ -package org.apache.iotdb.db.pipe.sink.protocol.opcua.server; +package org.apache.iotdb.pipe.plugin.sink.opcua.server; import org.apache.iotdb.pipe.api.exception.PipeException; diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaTcpNoneCompatibilityTest.java b/library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/server/OpcUaTcpNoneCompatibilityTest.java similarity index 98% rename from iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaTcpNoneCompatibilityTest.java rename to library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/server/OpcUaTcpNoneCompatibilityTest.java index fabee999e2ed..a199447774c4 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaTcpNoneCompatibilityTest.java +++ b/library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/server/OpcUaTcpNoneCompatibilityTest.java @@ -17,7 +17,7 @@ * under the License. */ -package org.apache.iotdb.db.pipe.sink.protocol.opcua.server; +package org.apache.iotdb.pipe.plugin.sink.opcua.server; import org.eclipse.milo.opcua.sdk.client.OpcUaClient; import org.eclipse.milo.opcua.sdk.client.identity.AnonymousProvider; diff --git a/library-pipe/pom.xml b/library-pipe/pom.xml new file mode 100644 index 000000000000..3f4749cb9d1c --- /dev/null +++ b/library-pipe/pom.xml @@ -0,0 +1,68 @@ + + + + 4.0.0 + + org.apache.iotdb + iotdb-parent + 2.0.11-SNAPSHOT + + library-pipe + IoTDB: Library-Pipe + pom + + 1.1.6 + + + opc-ua-sink + opc-ua-sink-example + + + + + org.eclipse.milo + milo-stack-core + ${milo.version} + + + org.eclipse.milo + milo-sdk-core + ${milo.version} + + + org.eclipse.milo + milo-transport + ${milo.version} + + + org.eclipse.milo + milo-sdk-client + ${milo.version} + + + org.eclipse.milo + milo-sdk-server + ${milo.version} + + + + diff --git a/licenses/Bouncy Castle Licence b/licenses/Bouncy Castle Licence new file mode 100644 index 000000000000..24506f04ccdd --- /dev/null +++ b/licenses/Bouncy Castle Licence @@ -0,0 +1,17 @@ +Copyright (c) 2000-2023 The Legion of the Bouncy Castle Inc. (https://www.bouncycastle.org) + +Permission is hereby granted, free of charge, to any person obtaining a copy of this software +and associated documentation files (the "Software"), to deal in the Software without restriction, +including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, +and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, +subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all copies or substantial +portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, +INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR +PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE +LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR +OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER +DEALINGS IN THE SOFTWARE. diff --git a/pom.xml b/pom.xml index 331a45246138..37ce4382c85e 100644 --- a/pom.xml +++ b/pom.xml @@ -44,6 +44,7 @@ iotdb-client iotdb-core iotdb-protocol + library-pipe distribution example library-udf @@ -109,7 +110,6 @@ 17 17 1.11.4 - 1.1.6 2.23.4 @@ -371,26 +371,6 @@ units ${airlift-units.version} - - org.eclipse.milo - milo-stack-core - ${milo.version} - - - org.eclipse.milo - milo-sdk-core - ${milo.version} - - - org.eclipse.milo - milo-transport - ${milo.version} - - - org.eclipse.milo - milo-sdk-client - ${milo.version} - io.airlift @@ -403,11 +383,6 @@ - - org.eclipse.milo - milo-sdk-server - ${milo.version} - org.reflections reflections From 19fd6eaeba6801e2fa3695377881e2c6f17552dd Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 14 Aug 2026 10:09:23 +0800 Subject: [PATCH 2/2] [Pipe] Fix OPC UA plugin CI failures --- .../env/cluster/config/MppConfigNodeConfig.java | 6 ++++++ .../it/env/cluster/config/MppDataNodeConfig.java | 6 ++++++ .../remote/config/RemoteConfigNodeConfig.java | 5 +++++ .../env/remote/config/RemoteDataNodeConfig.java | 5 +++++ .../iotdb/itbase/env/ConfigNodeConfig.java | 2 ++ .../apache/iotdb/itbase/env/DataNodeConfig.java | 2 ++ .../iotdb/pipe/it/single/IoTDBPipeOPCUAIT.java | 8 ++++++++ .../apache/iotdb/db/conf/IoTDBDescriptor.java | 8 +++++--- library-pipe/opc-ua-sink-example/pom.xml | 16 ++++++++++++++++ library-pipe/opc-ua-sink/README.md | 9 +++++++++ 10 files changed, 64 insertions(+), 3 deletions(-) diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppConfigNodeConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppConfigNodeConfig.java index 8da0f4e02b24..11a339df021b 100644 --- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppConfigNodeConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppConfigNodeConfig.java @@ -74,6 +74,12 @@ public ConfigNodeConfig setLeaderDistributionPolicy(String policy) { return this; } + @Override + public ConfigNodeConfig setRatisLogAppenderBufferSizeMax(int size) { + properties.setProperty("config_node_ratis_log_appender_buffer_size_max", String.valueOf(size)); + return this; + } + @Override public ConfigNodeConfig setMetadataLeaseFenceMs(long metadataLeaseFenceMs) { properties.setProperty("metadata_lease_fence_ms", String.valueOf(metadataLeaseFenceMs)); diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppDataNodeConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppDataNodeConfig.java index 8399955a5c59..ef753309f435 100644 --- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppDataNodeConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppDataNodeConfig.java @@ -80,6 +80,12 @@ public DataNodeConfig setConnectionTimeoutInMS(int connectionTimeoutInMS) { return this; } + @Override + public DataNodeConfig setThriftMaxFrameSize(int thriftMaxFrameSize) { + properties.setProperty("dn_thrift_max_frame_size", String.valueOf(thriftMaxFrameSize)); + return this; + } + @Override public DataNodeConfig setLoadTsFileAnalyzeSchemaMemorySizeInBytes( long loadTsFileAnalyzeSchemaMemorySizeInBytes) { diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteConfigNodeConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteConfigNodeConfig.java index 36fbdcbfc0e8..e697b98effef 100644 --- a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteConfigNodeConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteConfigNodeConfig.java @@ -44,6 +44,11 @@ public ConfigNodeConfig setLeaderDistributionPolicy(String policy) { return this; } + @Override + public ConfigNodeConfig setRatisLogAppenderBufferSizeMax(int size) { + return this; + } + @Override public ConfigNodeConfig setMetadataLeaseFenceMs(long metadataLeaseFenceMs) { return this; diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteDataNodeConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteDataNodeConfig.java index a76608e48515..81c28f0be61a 100644 --- a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteDataNodeConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteDataNodeConfig.java @@ -48,6 +48,11 @@ public DataNodeConfig setConnectionTimeoutInMS(int connectionTimeoutInMS) { return this; } + @Override + public DataNodeConfig setThriftMaxFrameSize(int thriftMaxFrameSize) { + return this; + } + @Override public DataNodeConfig setLoadTsFileAnalyzeSchemaMemorySizeInBytes( long loadTsFileAnalyzeSchemaMemorySizeInBytes) { diff --git a/integration-test/src/main/java/org/apache/iotdb/itbase/env/ConfigNodeConfig.java b/integration-test/src/main/java/org/apache/iotdb/itbase/env/ConfigNodeConfig.java index aec8f0f23354..a4260202a3a5 100644 --- a/integration-test/src/main/java/org/apache/iotdb/itbase/env/ConfigNodeConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/itbase/env/ConfigNodeConfig.java @@ -32,5 +32,7 @@ public interface ConfigNodeConfig { ConfigNodeConfig setLeaderDistributionPolicy(String policy); + ConfigNodeConfig setRatisLogAppenderBufferSizeMax(int size); + ConfigNodeConfig setMetadataLeaseFenceMs(long metadataLeaseFenceMs); } diff --git a/integration-test/src/main/java/org/apache/iotdb/itbase/env/DataNodeConfig.java b/integration-test/src/main/java/org/apache/iotdb/itbase/env/DataNodeConfig.java index bc045c9ba2fe..539ca012fd71 100644 --- a/integration-test/src/main/java/org/apache/iotdb/itbase/env/DataNodeConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/itbase/env/DataNodeConfig.java @@ -33,6 +33,8 @@ public interface DataNodeConfig { DataNodeConfig setConnectionTimeoutInMS(int connectionTimeoutInMS); + DataNodeConfig setThriftMaxFrameSize(int thriftMaxFrameSize); + DataNodeConfig setLoadTsFileAnalyzeSchemaMemorySizeInBytes( long loadTsFileAnalyzeSchemaMemorySizeInBytes); diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBPipeOPCUAIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBPipeOPCUAIT.java index e556de47e65c..b9ee75873589 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBPipeOPCUAIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBPipeOPCUAIT.java @@ -67,10 +67,18 @@ @Category({MultiClusterIT1.class}) public class IoTDBPipeOPCUAIT extends AbstractPipeSingleIT { + private static final int PIPE_PLUGIN_JAR_TRANSFER_LIMIT_IN_BYTES = 32 * 1024 * 1024; + @Before public void setUp() throws Exception { MultiEnvFactory.createEnv(1); env = MultiEnvFactory.getEnv(0); + env.getConfig() + .getConfigNodeConfig() + .setRatisLogAppenderBufferSizeMax(PIPE_PLUGIN_JAR_TRANSFER_LIMIT_IN_BYTES); + env.getConfig() + .getDataNodeConfig() + .setThriftMaxFrameSize(PIPE_PLUGIN_JAR_TRANSFER_LIMIT_IN_BYTES); env.getConfig() .getCommonConfig() .setAutoCreateSchemaEnabled(true) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java index fedcc73dbbe3..87c50c711fc5 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java @@ -141,10 +141,9 @@ public class IoTDBDescriptor { } protected IoTDBDescriptor() { - loadProps(); + boolean hasProperties = loadProps(); ServiceLoader propertiesLoaderServiceLoader = ServiceLoader.load(IPropertiesLoader.class); - boolean hasProperties = false; for (IPropertiesLoader loader : propertiesLoaderServiceLoader) { LOGGER.info(DataNodeMiscMessages.WILL_RELOAD_PROPERTIES, loader.getClass().getName()); hasProperties = true; @@ -227,7 +226,8 @@ else if (!urlString.endsWith(".properties")) { /** load a property file and set TsfileDBConfig variables. */ @SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning - private void loadProps() { + private boolean loadProps() { + boolean hasProperties = false; TrimProperties commonProperties = new TrimProperties(); // if new properties file exist, skip old properties files URL url = getPropsUrl(CommonConfig.SYSTEM_CONFIG_NAME); @@ -238,6 +238,7 @@ private void loadProps() { properties.load(new InputStreamReader(inputStream, StandardCharsets.UTF_8)); commonProperties.putAll(properties); loadProperties(commonProperties); + hasProperties = true; } catch (FileNotFoundException e) { LOGGER.error(DataNodeMiscMessages.FAIL_FIND_CONFIG_FILE, url, e); System.exit(-1); @@ -262,6 +263,7 @@ private void loadProps() { .MISC_LOG_COULDN_T_LOAD_THE_CONFIGURATION_FROM_ANY_OF_THE_KNOWN_SOURCES_EE3ED103, CommonConfig.SYSTEM_CONFIG_NAME); } + return hasProperties; } public void loadProperties(TrimProperties properties) throws BadNodeUrlException, IOException { diff --git a/library-pipe/opc-ua-sink-example/pom.xml b/library-pipe/opc-ua-sink-example/pom.xml index 921de8aba3ff..c0543538f588 100644 --- a/library-pipe/opc-ua-sink-example/pom.xml +++ b/library-pipe/opc-ua-sink-example/pom.xml @@ -29,14 +29,30 @@ pipe-opc-ua-sink-example IoTDB: Library-Pipe: OPC UA Sink Example + + org.eclipse.milo + milo-stack-core + org.eclipse.milo milo-sdk-client + + org.eclipse.milo + milo-transport + org.eclipse.milo milo-sdk-server + + org.bouncycastle + bcprov-jdk18on + + + org.slf4j + slf4j-api + diff --git a/library-pipe/opc-ua-sink/README.md b/library-pipe/opc-ua-sink/README.md index 90b750698d2c..90c260a5ac68 100644 --- a/library-pipe/opc-ua-sink/README.md +++ b/library-pipe/opc-ua-sink/README.md @@ -39,6 +39,15 @@ It is also included under `ext/pipe` in the standalone `library-pipe` distributi ## Register +The fat JAR is larger than the default 16 MiB ConfigNode consensus limit. Before starting the +nodes, set the following properties in `iotdb-system.properties` so the plugin can be transferred +and replicated: + +```properties +config_node_ratis_log_appender_buffer_size_max=33554432 +dn_thrift_max_frame_size=33554432 +``` + Upload the fat JAR to a URI accessible to IoTDB and register the plugin before creating an OPC UA Pipe: ```sql