fix

Author: Caideyipi

integration-test/pom.xml

@@ -195,6 +195,11 @@
             <artifactId>junit-jupiter-api</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>io.github.Caideyipi</groupId>
+            <version>0.0.1</version>
+            <artifactId>iotdb_opc_server</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.awaitility</groupId>
             <artifactId>awaitility</artifactId>

integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBPipeOPCUAIT.java

@@ -118,7 +118,8 @@ public void testOPCUAServerSink() throws Exception {
       final long startTime = System.currentTimeMillis();
       while (true) {
         try {
-          value = opcUaClient.readValue(0, TimestampsToReturn.Both, new NodeId(2, "root/db/opc")).get();
+          value =
+              opcUaClient.readValue(0, TimestampsToReturn.Both, new NodeId(2, "root/db/opc")).get();
           Assert.assertEquals(new DateTime(timestampToUtc(2)), value.getSourceTime());
           Assert.assertEquals(new Variant(2.0), value.getValue());
           Assert.assertEquals(StatusCode.UNCERTAIN, value.getStatusCode());
@@ -167,6 +168,9 @@ public void testOPCUAServerSink() throws Exception {
     }
   }
 
+  @Test
+  public void testOPCUAClientSink() throws Exception {}
+
   @Test
   public void testOPCUASinkInTableModel() throws Exception {
     try (final SyncConfigNodeIServiceClient client =

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSink.java

@@ -449,6 +449,9 @@ public void transfer(final TabletInsertionEvent tabletInsertionEvent) throws Exc
             nameSpace.transfer(tablet, isTableModel, this);
           } else if (Objects.nonNull(client)) {
             client.transfer(tablet, this);
+          } else {
+            throw new PipeException(
+                "No OPC client or server is specified when transferring tablet");
           }
         });
   }