Merge master and address review comments

Author: Caideyipi

.github/workflows/sonar-codecov.yml

@@ -71,11 +71,11 @@ jobs:
     if: github.event_name == 'pull_request' && github.event.pull_request.head.repo.full_name == 'apache/iotdb' || github.event_name == 'push'
     steps:
       - uses: actions/checkout@v5
-      - name: Set up JDK 17
+      - name: Set up JDK 21
         uses: actions/setup-java@v5
         with:
           distribution: corretto
-          java-version: 17
+          java-version: 21
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
       - name: Cache Maven packages

LICENSE-binary

@@ -230,7 +230,7 @@ io.swagger.core.v3:swagger-integration-jakarta:2.2.50
 io.swagger.core.v3:swagger-jaxrs2-jakarta:2.2.50
 io.swagger.core.v3:swagger-models-jakarta:2.2.50
 jakarta.inject:jakarta.inject-api:2.0.1
-at.yawk.lz4:lz4-java:1.10.0
+at.yawk.lz4:lz4-java:1.11.1
 com.github.ben-manes.caffeine:caffeine:2.9.3
 org.eclipse.jetty:jetty-http:12.0.36
 org.eclipse.jetty:jetty-io:12.0.36
@@ -246,17 +246,17 @@ io.dropwizard.metrics:metrics-core:4.2.19
 io.dropwizard.metrics:metrics-jvm:3.2.2
 com.librato.metrics:metrics-librato:5.1.0
 com.github.moquette-io.moquette:moquette-broker:0.18
-io.netty:netty-buffer:4.1.134.Final
-io.netty:netty-codec:4.1.134.Final
-io.netty:netty-codec-http:4.1.134.Final
-io.netty:netty-codec-mqtt:4.1.134.Final
-io.netty:netty-common:4.1.134.Final
-io.netty:netty-handler:4.1.134.Final
-io.netty:netty-resolver:4.1.134.Final
-io.netty:netty-transport:4.1.134.Final
-io.netty:netty-transport-native-epoll:4.1.134.Final:linux-aarch_64
-io.netty:netty-transport-native-epoll:4.1.134.Final:linux-x86_64
-io.netty:netty-transport-native-unix-common:4.1.134.Final
+io.netty:netty-buffer:4.1.136.Final
+io.netty:netty-codec:4.1.136.Final
+io.netty:netty-codec-http:4.1.136.Final
+io.netty:netty-codec-mqtt:4.1.136.Final
+io.netty:netty-common:4.1.136.Final
+io.netty:netty-handler:4.1.136.Final
+io.netty:netty-resolver:4.1.136.Final
+io.netty:netty-transport:4.1.136.Final
+io.netty:netty-transport-native-epoll:4.1.136.Final:linux-aarch_64
+io.netty:netty-transport-native-epoll:4.1.136.Final:linux-x86_64
+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

dependencies.json

@@ -28,7 +28,6 @@
         "com.google.j2objc:j2objc-annotations",
         "com.h2database:h2-mvstore",
         "com.sun.activation:jakarta.activation",
-        "com.sun.istack:istack-commons-runtime",
         "com.zaxxer:HikariCP",
         "commons-cli:commons-cli",
         "commons-codec:commons-codec",
@@ -116,14 +115,11 @@
         "org.eclipse.jetty:jetty-session",
         "org.eclipse.jetty:jetty-util",
         "org.eclipse.jetty.ee10:jetty-ee10-servlet",
-        "org.eclipse.milo:bsd-core",
-        "org.eclipse.milo:bsd-generator",
-        "org.eclipse.milo:sdk-client",
-        "org.eclipse.milo:sdk-core",
-        "org.eclipse.milo:sdk-server",
-        "org.eclipse.milo:stack-client",
-        "org.eclipse.milo:stack-core",
-        "org.eclipse.milo:stack-server",
+        "org.eclipse.milo:milo-sdk-client",
+        "org.eclipse.milo:milo-sdk-core",
+        "org.eclipse.milo:milo-sdk-server",
+        "org.eclipse.milo:milo-stack-core",
+        "org.eclipse.milo:milo-transport",
         "org.fusesource.hawtbuf:hawtbuf",
         "org.fusesource.hawtdispatch:hawtdispatch",
         "org.fusesource.hawtdispatch:hawtdispatch-transport",
@@ -133,8 +129,6 @@
         "org.glassfish.hk2:hk2-utils",
         "org.glassfish.hk2:osgi-resource-locator",
         "org.glassfish.hk2.external:aopalliance-repackaged",
-        "org.glassfish.jaxb:jaxb-runtime",
-        "org.glassfish.jaxb:txw2",
         "org.glassfish.jersey.containers:jersey-container-servlet-core",
         "org.glassfish.jersey.core:jersey-client",
         "org.glassfish.jersey.core:jersey-common",
@@ -145,6 +139,7 @@
         "org.java-websocket:Java-WebSocket",
         "org.javassist:javassist",
         "org.jline:jline",
+        "org.jspecify:jspecify",
         "org.jvnet.mimepull:mimepull",
         "org.latencyutils:LatencyUtils",
         "org.ops4j.pax.jdbc:pax-jdbc-common",

example/pipe-opc-ua-sink/pom.xml

@@ -31,12 +31,12 @@
     <dependencies>
         <dependency>
             <groupId>org.eclipse.milo</groupId>
-            <artifactId>sdk-client</artifactId>
+            <artifactId>milo-sdk-client</artifactId>
             <version>${milo.version}</version>
         </dependency>
         <dependency>
             <groupId>org.eclipse.milo</groupId>
-            <artifactId>sdk-server</artifactId>
+            <artifactId>milo-sdk-server</artifactId>
         </dependency>
     </dependencies>
     <profiles>

example/pipe-opc-ua-sink/src/main/java/org/apache/iotdb/opcua/ClientExample.java

@@ -20,8 +20,8 @@
 package org.apache.iotdb.opcua;
 
 import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
-import org.eclipse.milo.opcua.sdk.client.api.identity.AnonymousProvider;
-import org.eclipse.milo.opcua.sdk.client.api.identity.IdentityProvider;
+import org.eclipse.milo.opcua.sdk.client.identity.AnonymousProvider;
+import org.eclipse.milo.opcua.sdk.client.identity.IdentityProvider;
 import org.eclipse.milo.opcua.stack.core.security.SecurityPolicy;
 import org.eclipse.milo.opcua.stack.core.types.structured.EndpointDescription;
 

example/pipe-opc-ua-sink/src/main/java/org/apache/iotdb/opcua/ClientExampleRunner.java

@@ -21,13 +21,14 @@
 
 import org.bouncycastle.jce.provider.BouncyCastleProvider;
 import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
-import org.eclipse.milo.opcua.stack.client.security.DefaultClientCertificateValidator;
 import org.eclipse.milo.opcua.stack.core.Stack;
-import org.eclipse.milo.opcua.stack.core.security.DefaultTrustListManager;
+import org.eclipse.milo.opcua.stack.core.security.DefaultClientCertificateValidator;
+import org.eclipse.milo.opcua.stack.core.security.FileBasedCertificateQuarantine;
+import org.eclipse.milo.opcua.stack.core.security.FileBasedTrustListManager;
 import org.eclipse.milo.opcua.stack.core.types.builtin.LocalizedText;
 import org.slf4j.LoggerFactory;
 
-import java.io.File;
+import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
@@ -48,6 +49,7 @@ public class ClientExampleRunner {
   private final CompletableFuture<OpcUaClient> future = new CompletableFuture<>();
 
   private final ClientExample clientExample;
+  private FileBasedTrustListManager trustListManager;
 
   public ClientExampleRunner(ClientExample clientExample) {
     this.clientExample = clientExample;
@@ -61,21 +63,24 @@ private OpcUaClient createClient() throws Exception {
       throw new Exception("unable to create security dir: " + securityTempDir);
     }
 
-    final File pkiDir = securityTempDir.resolve("pki").toFile();
+    final Path pkiDir = securityTempDir.resolve("pki");
 
     System.out.println("security dir: " + securityTempDir.toAbsolutePath());
-    LoggerFactory.getLogger(getClass()).info("security pki dir: {}", pkiDir.getAbsolutePath());
+    LoggerFactory.getLogger(getClass()).info("security pki dir: {}", pkiDir.toAbsolutePath());
 
     final IoTDBKeyStoreLoaderClient loader = new IoTDBKeyStoreLoaderClient().load(securityTempDir);
 
-    final DefaultTrustListManager trustListManager = new DefaultTrustListManager(pkiDir);
+    trustListManager = FileBasedTrustListManager.createAndInitialize(pkiDir);
+    final FileBasedCertificateQuarantine certificateQuarantine =
+        FileBasedCertificateQuarantine.create(pkiDir.resolve("rejected").resolve("certs"));
 
     final DefaultClientCertificateValidator certificateValidator =
-        new DefaultClientCertificateValidator(trustListManager);
+        new DefaultClientCertificateValidator(trustListManager, certificateQuarantine);
 
     return OpcUaClient.create(
         clientExample.getEndpointUrl(),
         endpoints -> endpoints.stream().filter(clientExample.endpointFilter()).findFirst(),
+        transportBuilder -> {},
         configBuilder ->
             configBuilder
                 .setApplicationName(LocalizedText.english("eclipse milo opc-ua client"))
@@ -85,8 +90,7 @@ private OpcUaClient createClient() throws Exception {
                 .setCertificateChain(loader.getClientCertificateChain())
                 .setCertificateValidator(certificateValidator)
                 .setIdentityProvider(clientExample.getIdentityProvider())
-                .setRequestTimeout(uint(5000))
-                .build());
+                .setRequestTimeout(uint(5000)));
   }
 
   public void run() {
@@ -100,11 +104,13 @@ public void run() {
             }
 
             try {
-              client.disconnect().get();
-              Stack.releaseSharedResources();
+              client.disconnectAsync().get();
             } catch (InterruptedException | ExecutionException e) {
               Thread.currentThread().interrupt();
               System.out.println("Error disconnecting: {}" + e.getMessage());
+            } finally {
+              closeTrustListManager();
+              Stack.releaseSharedResources();
             }
 
             try {
@@ -126,6 +132,7 @@ public void run() {
     } catch (Throwable t) {
       System.out.println("Error getting client: {}" + t.getMessage());
 
+      closeTrustListManager();
       future.completeExceptionally(t);
 
       try {
@@ -144,4 +151,16 @@ public void run() {
       e.printStackTrace();
     }
   }
+
+  private void closeTrustListManager() {
+    if (trustListManager != null) {
+      try {
+        trustListManager.close();
+      } catch (IOException e) {
+        e.printStackTrace();
+      } finally {
+        trustListManager = null;
+      }
+    }
+  }
 }

example/pipe-opc-ua-sink/src/main/java/org/apache/iotdb/opcua/ClientTest.java

@@ -20,27 +20,18 @@
 package org.apache.iotdb.opcua;
 
 import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
-import org.eclipse.milo.opcua.sdk.client.api.subscriptions.UaMonitoredItem;
-import org.eclipse.milo.opcua.sdk.client.api.subscriptions.UaSubscription;
+import org.eclipse.milo.opcua.sdk.client.subscriptions.OpcUaMonitoredItem;
+import org.eclipse.milo.opcua.sdk.client.subscriptions.OpcUaSubscription;
 import org.eclipse.milo.opcua.stack.core.AttributeId;
 import org.eclipse.milo.opcua.stack.core.Identifiers;
-import org.eclipse.milo.opcua.stack.core.types.builtin.ExtensionObject;
 import org.eclipse.milo.opcua.stack.core.types.builtin.QualifiedName;
-import org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.UInteger;
 import org.eclipse.milo.opcua.stack.core.types.enumerated.MonitoringMode;
-import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn;
 import org.eclipse.milo.opcua.stack.core.types.structured.ContentFilter;
 import org.eclipse.milo.opcua.stack.core.types.structured.EventFilter;
-import org.eclipse.milo.opcua.stack.core.types.structured.MonitoredItemCreateRequest;
-import org.eclipse.milo.opcua.stack.core.types.structured.MonitoringParameters;
 import org.eclipse.milo.opcua.stack.core.types.structured.ReadValueId;
 import org.eclipse.milo.opcua.stack.core.types.structured.SimpleAttributeOperand;
 
-import java.util.Collections;
-import java.util.List;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
 
 import static org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.Unsigned.uint;
 
@@ -52,24 +43,19 @@ public static void main(String[] args) {
     new ClientExampleRunner(example).run();
   }
 
-  private final AtomicLong clientHandles = new AtomicLong(1L);
-
   @Override
   public void run(OpcUaClient client, CompletableFuture<OpcUaClient> future) throws Exception {
     // synchronous connect
-    client.connect().get();
+    client.connect();
 
     // create a subscription and a monitored item
-    final UaSubscription subscription =
-        client.getSubscriptionManager().createSubscription(200.0).get();
+    final OpcUaSubscription subscription = new OpcUaSubscription(client, 200.0);
+    subscription.create();
 
     final ReadValueId readValueId =
         new ReadValueId(
             Identifiers.Server, AttributeId.EventNotifier.uid(), null, QualifiedName.NULL_VALUE);
 
-    // client handle must be unique per item
-    final UInteger clientHandle = uint(clientHandles.getAndIncrement());
-
     final EventFilter eventFilter =
         new EventFilter(
             new SimpleAttributeOperand[] {
@@ -96,30 +82,18 @@ public void run(OpcUaClient client, CompletableFuture<OpcUaClient> future) throw
             },
             new ContentFilter(null));
 
-    final MonitoringParameters parameters =
-        new MonitoringParameters(
-            clientHandle,
-            0.0,
-            ExtensionObject.encode(client.getStaticSerializationContext(), eventFilter),
-            uint(10000),
-            true);
-
-    final MonitoredItemCreateRequest request =
-        new MonitoredItemCreateRequest(readValueId, MonitoringMode.Reporting, parameters);
-
-    final List<UaMonitoredItem> items =
-        subscription
-            .createMonitoredItems(TimestampsToReturn.Both, Collections.singletonList(request))
-            .get();
+    final OpcUaMonitoredItem monitoredItem =
+        new OpcUaMonitoredItem(readValueId, MonitoringMode.Reporting);
+    monitoredItem.setSamplingInterval(0.0);
+    monitoredItem.setFilter(eventFilter);
+    monitoredItem.setQueueSize(uint(10000));
+    monitoredItem.setDiscardOldest(true);
+    subscription.addMonitoredItem(monitoredItem);
+    subscription.synchronizeMonitoredItems();
 
     // do something with the value updates
-    final UaMonitoredItem monitoredItem = items.get(0);
-
-    final AtomicInteger eventCount = new AtomicInteger(0);
-
-    monitoredItem.setEventConsumer(
+    monitoredItem.setEventValueListener(
         (item, vs) -> {
-          eventCount.incrementAndGet();
           System.out.println("Event Received from " + item.getReadValueId().getNodeId());
 
           for (int i = 0; i < vs.length; i++) {

example/session/src/main/java/org/apache/iotdb/ConsensusSubscriptionWalFileAnalyzer.java

@@ -23,10 +23,10 @@
 
 import java.io.File;
 import java.io.IOException;
-import java.io.RandomAccessFile;
 import java.nio.ByteBuffer;
 import java.nio.channels.FileChannel;
 import java.nio.charset.StandardCharsets;
+import java.nio.file.StandardOpenOption;
 import java.util.Locale;
 
 /**
@@ -79,8 +79,7 @@ private static void printUsage() {
   }
 
   private static WalFileAnalysis analyze(final File walFile) throws IOException {
-    try (RandomAccessFile raf = new RandomAccessFile(walFile, "r");
-        FileChannel channel = raf.getChannel()) {
+    try (FileChannel channel = FileChannel.open(walFile.toPath(), StandardOpenOption.READ)) {
       final long totalBytes = channel.size();
       final String version = detectVersion(channel, totalBytes);
       final int headMagicBytes = getHeadMagicBytes(version);

example/subscription/src/main/java/org/apache/iotdb/ConsensusSubscriptionSessionExample.java

@@ -109,7 +109,7 @@ private static void createConsensusTopic(final String topicName, final String pa
       session.open();
 
       final Properties config = new Properties();
-      config.put(TopicConstant.MODE_KEY, TopicConstant.MODE_CONSENSUS_VALUE);
+      config.put(TopicConstant.MODE_KEY, TopicConstant.MODE_INCREMENTAL_VALUE);
       config.put(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_RECORD_HANDLER_VALUE);
       config.put(TopicConstant.PATH_KEY, path);
       config.put(TopicConstant.ORDER_MODE_KEY, TopicConstant.ORDER_MODE_PER_WRITER_VALUE);

example/subscription/src/main/java/org/apache/iotdb/ConsensusTableModelSubscriptionSessionExample.java

@@ -107,7 +107,7 @@ private static void createConsensusTopic(
             .password(PASSWORD)
             .build()) {
       final Properties config = new Properties();
-      config.put(TopicConstant.MODE_KEY, TopicConstant.MODE_CONSENSUS_VALUE);
+      config.put(TopicConstant.MODE_KEY, TopicConstant.MODE_INCREMENTAL_VALUE);
       config.put(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_RECORD_HANDLER_VALUE);
       config.put(TopicConstant.DATABASE_KEY, database);
       config.put(TopicConstant.TABLE_KEY, table);