This is an automated email from the ASF dual-hosted git repository.
dominikriemer pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to refs/heads/dev by this push:
new 807f4e3918 chore: Improve timeout handling of OPC UA adapter (#4723)
807f4e3918 is described below
commit 807f4e391803f63c26885c0c60b9358eacb87e6a
Author: Dominik Riemer <[email protected]>
AuthorDate: Thu Jul 16 16:16:40 2026 +0200
chore: Improve timeout handling of OPC UA adapter (#4723)
---
.../management/connect/PullAdapterScheduler.java | 19 ++-
.../connect/PullAdapterSchedulerTest.java | 55 +++++++
.../connectors/opcua/adapter/OpcUaAdapter.java | 13 +-
.../opcua/alarms/OpcUaAlarmEventSubscriber.java | 32 +++-
.../opcua/client/ConnectedOpcUaClient.java | 37 +++++
.../alarms/OpcUaAlarmEventSubscriberTest.java | 68 ++++++++
.../opcua/client/ConnectedOpcUaClientTest.java | 66 ++++++++
.../OpcUaAdapterReadTimeoutReproductionTest.java | 177 +++++++++++++++++++++
.../adapters/opcua/OpcUaAdapterTestHarness.java | 34 +++-
.../containers/OpcUaDemoServerContainer.java | 7 +-
10 files changed, 487 insertions(+), 21 deletions(-)
diff --git
a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/PullAdapterScheduler.java
b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/PullAdapterScheduler.java
index 38693d065f..c8caf1f1b0 100644
---
a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/PullAdapterScheduler.java
+++
b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/PullAdapterScheduler.java
@@ -31,6 +31,7 @@ import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
public class PullAdapterScheduler {
@@ -44,15 +45,23 @@ public class PullAdapterScheduler {
public void schedule(IPullAdapter pullAdapter,
String adapterElementId) {
+ var pullFailed = new AtomicBoolean();
final Runnable task = () -> {
try {
pullAdapter.pullData();
+ if (pullFailed.compareAndSet(true, false)) {
+ LOG.info("Adapter {} recovered from pull failures",
adapterElementId);
+ }
} catch (ExecutionException | InterruptedException | TimeoutException |
RuntimeException e) {
- LOG.error("Error while pulling data: {}", e.getMessage());
- SpMonitoringManager.INSTANCE.addErrorMessage(
- adapterElementId,
- SpLogEntry.from(System.currentTimeMillis(), SpLogMessage.from(e))
- );
+ if (pullFailed.compareAndSet(false, true)) {
+ LOG.warn("Adapter {} failed while pulling data: {}",
adapterElementId, e.getMessage());
+ SpMonitoringManager.INSTANCE.addErrorMessage(
+ adapterElementId,
+ SpLogEntry.from(System.currentTimeMillis(), SpLogMessage.from(e))
+ );
+ } else {
+ LOG.debug("Adapter {} still failing while pulling data: {}",
adapterElementId, e.getMessage());
+ }
if (e instanceof InterruptedException) {
Thread.currentThread().interrupt();
}
diff --git
a/streampipes-extensions-management/src/test/java/org/apache/streampipes/extensions/management/connect/PullAdapterSchedulerTest.java
b/streampipes-extensions-management/src/test/java/org/apache/streampipes/extensions/management/connect/PullAdapterSchedulerTest.java
index c6ef7df54f..d96e0fa71b 100644
---
a/streampipes-extensions-management/src/test/java/org/apache/streampipes/extensions/management/connect/PullAdapterSchedulerTest.java
+++
b/streampipes-extensions-management/src/test/java/org/apache/streampipes/extensions/management/connect/PullAdapterSchedulerTest.java
@@ -20,6 +20,7 @@ package org.apache.streampipes.extensions.management.connect;
import org.apache.streampipes.extensions.api.connect.IPollingSettings;
import org.apache.streampipes.extensions.api.connect.IPullAdapter;
+import org.apache.streampipes.extensions.api.monitoring.SpMonitoringManager;
import
org.apache.streampipes.extensions.management.connect.adapter.util.PollingSettings;
import org.junit.jupiter.api.Test;
@@ -28,6 +29,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
class PullAdapterSchedulerTest {
@@ -45,6 +47,30 @@ class PullAdapterSchedulerTest {
}
}
+ @Test
+ void recordsOnlyFirstFailureUntilAdapterRecovers() throws
InterruptedException {
+ var adapterId = "repeatedly-failing-adapter";
+ var scheduler = new PullAdapterScheduler();
+ var fourthInvocation = new CountDownLatch(1);
+ SpMonitoringManager.INSTANCE.remove(adapterId);
+
+ try {
+ scheduler.schedule(new RepeatedlyFailingPullAdapter(fourthInvocation),
adapterId);
+ assertTrue(fourthInvocation.await(2, TimeUnit.SECONDS));
+ } finally {
+ scheduler.shutdown();
+ }
+
+ try {
+ var logEntries = SpMonitoringManager.INSTANCE.getMonitoringInfo()
+ .getLogInfos()
+ .get(adapterId);
+ assertEquals(2, logEntries.size());
+ } finally {
+ SpMonitoringManager.INSTANCE.remove(adapterId);
+ }
+ }
+
private static class FailingOncePullAdapter implements IPullAdapter {
private final AtomicInteger invocations = new AtomicInteger();
@@ -67,4 +93,33 @@ class PullAdapterSchedulerTest {
return PollingSettings.from(TimeUnit.MILLISECONDS, 10);
}
}
+
+ private static class RepeatedlyFailingPullAdapter implements IPullAdapter {
+
+ private final AtomicInteger invocations = new AtomicInteger();
+ private final CountDownLatch fourthInvocation;
+
+ private RepeatedlyFailingPullAdapter(CountDownLatch fourthInvocation) {
+ this.fourthInvocation = fourthInvocation;
+ }
+
+ @Override
+ public void pullData() {
+ int invocation = invocations.incrementAndGet();
+ try {
+ if (invocation == 1 || invocation == 2 || invocation == 4) {
+ throw new IllegalStateException("poll failed");
+ }
+ } finally {
+ if (invocation == 4) {
+ fourthInvocation.countDown();
+ }
+ }
+ }
+
+ @Override
+ public IPollingSettings getPollingInterval() {
+ return PollingSettings.from(TimeUnit.MILLISECONDS, 10);
+ }
+ }
}
diff --git
a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/OpcUaAdapter.java
b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/OpcUaAdapter.java
index 20799195a6..81a922af1b 100644
---
a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/OpcUaAdapter.java
+++
b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/OpcUaAdapter.java
@@ -50,7 +50,6 @@ import org.apache.streampipes.sdk.helpers.Locales;
import org.eclipse.milo.opcua.sdk.client.subscriptions.OpcUaMonitoredItem;
import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue;
import org.eclipse.milo.opcua.stack.core.types.builtin.StatusCode;
-import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -176,14 +175,16 @@ public class OpcUaAdapter implements StreamPipesAdapter,
IPullAdapter, SupportsR
LOG.debug("Reading {} OPC UA nodes: {}", nodeIds.size(), nodeIds);
var response =
- this.connectedClient.getClient().readValuesAsync(
- 0,
- TimestampsToReturn.Both,
- nodeIds);
+ this.connectedClient.readValuesAsync(
+ nodeIds,
+
this.getPollingInterval().timeUnit().toMillis(this.getPollingInterval().value()));
boolean badStatusCodeReceived = false;
boolean emptyValueReceived = false;
List<DataValue> returnValues =
- response.get(this.getPollingInterval().value(),
this.getPollingInterval().timeUnit());
+ response.get(
+ 2L * this.getPollingInterval().value(),
+ this.getPollingInterval().timeUnit()
+ );
if (returnValues == null) {
emptyValueReceived = true;
LOG.debug("Null value object returned for OPC UA nodes {} - event will
not be sent", nodeIds);
diff --git
a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/alarms/OpcUaAlarmEventSubscriber.java
b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/alarms/OpcUaAlarmEventSubscriber.java
index 866bae83a1..ea06e69302 100644
---
a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/alarms/OpcUaAlarmEventSubscriber.java
+++
b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/alarms/OpcUaAlarmEventSubscriber.java
@@ -70,11 +70,25 @@ public class OpcUaAlarmEventSubscriber implements
AutoCloseable {
OpcUaAlarmEventSubscriber(ConnectedOpcUaClient connectedClient,
OpcUaAlarmAdapterConfig config,
Consumer<Map<String, Object>> eventConsumer) {
+ this(
+ connectedClient,
+ config,
+ eventConsumer,
+ OpcUaAlarmEventMapper.create(connectedClient.getClient(), config),
+ new OpcUaAlarmEventFilter(config)
+ );
+ }
+
+ OpcUaAlarmEventSubscriber(ConnectedOpcUaClient connectedClient,
+ OpcUaAlarmAdapterConfig config,
+ Consumer<Map<String, Object>> eventConsumer,
+ OpcUaAlarmEventMapper eventMapper,
+ OpcUaAlarmEventFilter eventFilter) {
this.connectedClient = connectedClient;
this.config = config;
this.eventConsumer = eventConsumer;
- this.eventMapper =
OpcUaAlarmEventMapper.create(connectedClient.getClient(), config);
- this.eventFilter = new OpcUaAlarmEventFilter(config);
+ this.eventMapper = eventMapper;
+ this.eventFilter = eventFilter;
this.sessionActivityListener = new SessionActivityListener() {
@Override
public void onSessionActive(org.eclipse.milo.opcua.sdk.client.UaSession
session) {
@@ -119,6 +133,17 @@ public class OpcUaAlarmEventSubscriber implements
AutoCloseable {
private void createSubscription() throws UaException {
setLastSubscriptionOperation("create-subscription");
OpcUaSubscription newSubscription = createManagedSubscription();
+ this.subscription = newSubscription;
+
+ try {
+ initializeSubscription(newSubscription);
+ } catch (UaException | RuntimeException e) {
+ deleteSubscriptionQuietly();
+ throw e;
+ }
+ }
+
+ private void initializeSubscription(OpcUaSubscription newSubscription)
throws UaException {
newSubscription.setSubscriptionListener(new
OpcUaSubscription.SubscriptionListener() {
@Override
public void onKeepAliveReceived(OpcUaSubscription subscription) {
@@ -196,14 +221,13 @@ public class OpcUaAlarmEventSubscriber implements
AutoCloseable {
results.get(0).operationResult().orElse(results.get(0).serviceResult())
);
- this.subscription = newSubscription;
requestConditionRefresh(newSubscription);
setLastSubscriptionOperation(
"subscription-ready(subscriptionId=%s)".formatted(newSubscription.getSubscriptionId().orElse(null))
);
}
- private OpcUaSubscription createManagedSubscription() throws UaException {
+ OpcUaSubscription createManagedSubscription() throws UaException {
var subscription = new OpcUaSubscription(connectedClient.getClient(),
PUBLISHING_INTERVAL_MS);
subscription.create();
return subscription;
diff --git
a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/client/ConnectedOpcUaClient.java
b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/client/ConnectedOpcUaClient.java
index 1441695cbe..a8f0c7dd06 100644
---
a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/client/ConnectedOpcUaClient.java
+++
b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/client/ConnectedOpcUaClient.java
@@ -25,15 +25,24 @@ import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
import
org.eclipse.milo.opcua.sdk.client.subscriptions.MonitoredItemServiceOperationResult;
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.UaException;
+import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue;
import org.eclipse.milo.opcua.stack.core.types.builtin.NodeId;
+import org.eclipse.milo.opcua.stack.core.types.builtin.QualifiedName;
import org.eclipse.milo.opcua.stack.core.types.builtin.StatusCode;
+import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn;
+import org.eclipse.milo.opcua.stack.core.types.structured.ReadRequest;
+import org.eclipse.milo.opcua.stack.core.types.structured.ReadResponse;
+import org.eclipse.milo.opcua.stack.core.types.structured.ReadValueId;
import org.jspecify.annotations.NonNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
+import java.util.concurrent.CompletableFuture;
import static
org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.Unsigned.uint;
@@ -122,6 +131,34 @@ public class ConnectedOpcUaClient {
return this.client;
}
+ public CompletableFuture<List<DataValue>> readValuesAsync(List<NodeId>
nodeIds,
+ long
requestTimeoutMillis) {
+ var readValueIds = nodeIds.stream()
+ .map(nodeId -> new ReadValueId(
+ nodeId,
+ AttributeId.Value.uid(),
+ null,
+ QualifiedName.NULL_VALUE
+ ))
+ .toArray(ReadValueId[]::new);
+
+ return client.getSessionAsync().thenCompose(session -> {
+ var request = new ReadRequest(
+ client.newRequestHeader(session.getAuthenticationToken(),
uint(requestTimeoutMillis)),
+ 0.0,
+ TimestampsToReturn.Both,
+ readValueIds
+ );
+
+ return client.sendRequestAsync(request)
+ .thenApply(ReadResponse.class::cast)
+ .thenApply(response -> {
+ var results = response.getResults();
+ return results == null ? List.of() : Arrays.asList(results);
+ });
+ });
+ }
+
public void disconnect() {
try {
client.disconnect();
diff --git
a/streampipes-extensions/streampipes-connectors-opcua/src/test/java/org/apache/streampipes/extensions/connectors/opcua/alarms/OpcUaAlarmEventSubscriberTest.java
b/streampipes-extensions/streampipes-connectors-opcua/src/test/java/org/apache/streampipes/extensions/connectors/opcua/alarms/OpcUaAlarmEventSubscriberTest.java
new file mode 100644
index 0000000000..16cd04f89a
--- /dev/null
+++
b/streampipes-extensions/streampipes-connectors-opcua/src/test/java/org/apache/streampipes/extensions/connectors/opcua/alarms/OpcUaAlarmEventSubscriberTest.java
@@ -0,0 +1,68 @@
+/*
+ * 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.streampipes.extensions.connectors.opcua.alarms;
+
+import
org.apache.streampipes.extensions.connectors.opcua.client.ConnectedOpcUaClient;
+
+import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
+import org.eclipse.milo.opcua.sdk.client.subscriptions.OpcUaSubscription;
+import org.eclipse.milo.opcua.stack.core.types.structured.EventFilter;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class OpcUaAlarmEventSubscriberTest {
+
+ @Test
+ void deletesSubscriptionWhenMonitoredItemCreationFails() throws Exception {
+ var client = mock(OpcUaClient.class);
+ var connectedClient = mock(ConnectedOpcUaClient.class);
+ var config = new OpcUaAlarmAdapterConfig();
+ var eventMapper = mock(OpcUaAlarmEventMapper.class);
+ var eventFilter = mock(OpcUaAlarmEventFilter.class);
+ var subscription = mock(OpcUaSubscription.class);
+
+ when(connectedClient.getClient()).thenReturn(client);
+ when(eventMapper.makeEventFilter(client.getStaticEncodingContext()))
+ .thenReturn(mock(EventFilter.class));
+ when(subscription.createMonitoredItems()).thenReturn(List.of());
+
+ var subscriber = spy(new OpcUaAlarmEventSubscriber(
+ connectedClient,
+ config,
+ event -> { },
+ eventMapper,
+ eventFilter
+ ));
+ doReturn(subscription).when(subscriber).createManagedSubscription();
+
+ assertThrows(org.eclipse.milo.opcua.stack.core.UaException.class,
subscriber::start);
+ subscriber.close();
+
+ verify(subscription, times(1)).delete();
+ }
+}
diff --git
a/streampipes-extensions/streampipes-connectors-opcua/src/test/java/org/apache/streampipes/extensions/connectors/opcua/client/ConnectedOpcUaClientTest.java
b/streampipes-extensions/streampipes-connectors-opcua/src/test/java/org/apache/streampipes/extensions/connectors/opcua/client/ConnectedOpcUaClientTest.java
new file mode 100644
index 0000000000..554c14a348
--- /dev/null
+++
b/streampipes-extensions/streampipes-connectors-opcua/src/test/java/org/apache/streampipes/extensions/connectors/opcua/client/ConnectedOpcUaClientTest.java
@@ -0,0 +1,66 @@
+/*
+ * 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.streampipes.extensions.connectors.opcua.client;
+
+import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
+import org.eclipse.milo.opcua.sdk.client.OpcUaSession;
+import org.eclipse.milo.opcua.stack.core.types.UaResponseMessageType;
+import org.eclipse.milo.opcua.stack.core.types.builtin.NodeId;
+import org.eclipse.milo.opcua.stack.core.types.structured.ReadRequest;
+import org.eclipse.milo.opcua.stack.core.types.structured.RequestHeader;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+import static
org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.Unsigned.uint;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class ConnectedOpcUaClientTest {
+
+ @Test
+ void usesPullTimeoutAsRequestTimeoutHint() {
+ var client = mock(OpcUaClient.class);
+ var session = mock(OpcUaSession.class);
+ var authenticationToken = new NodeId(1, "authentication-token");
+ var requestHeader = mock(RequestHeader.class);
+ var responseFuture = new CompletableFuture<UaResponseMessageType>();
+
+ when(session.getAuthenticationToken()).thenReturn(authenticationToken);
+
when(client.getSessionAsync()).thenReturn(CompletableFuture.completedFuture(session));
+ when(client.newRequestHeader(authenticationToken,
uint(250))).thenReturn(requestHeader);
+ when(client.sendRequestAsync(any())).thenReturn(responseFuture);
+
+ var connectedClient = new ConnectedOpcUaClient(client);
+ connectedClient.readValuesAsync(List.of(new NodeId(2, "value")), 250);
+
+ var requestCaptor =
ArgumentCaptor.forClass(org.eclipse.milo.opcua.stack.core.types.UaRequestMessageType.class);
+ verify(client).sendRequestAsync(requestCaptor.capture());
+
+ var request = (ReadRequest) requestCaptor.getValue();
+ assertSame(requestHeader, request.getRequestHeader());
+ assertEquals(1, request.getNodesToRead().length);
+ }
+}
diff --git
a/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/adapters/OpcUaAdapterReadTimeoutReproductionTest.java
b/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/adapters/OpcUaAdapterReadTimeoutReproductionTest.java
new file mode 100644
index 0000000000..beb54341fa
--- /dev/null
+++
b/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/adapters/OpcUaAdapterReadTimeoutReproductionTest.java
@@ -0,0 +1,177 @@
+/*
+ * 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.streampipes.integration.adapters;
+
+import org.apache.streampipes.extensions.api.monitoring.SpMonitoringManager;
+import
org.apache.streampipes.extensions.connectors.opcua.client.OpcUaClientProvider;
+import
org.apache.streampipes.integration.adapters.opcua.OpcUaAdapterTestHarness;
+import
org.apache.streampipes.integration.adapters.opcua.OpcUaAdapterTestHarness.RunningOpcUaAdapter;
+import org.apache.streampipes.integration.containers.OpcUaDemoServerContainer;
+
+import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
+import org.eclipse.milo.opcua.stack.core.security.SecurityPolicy;
+import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue;
+import org.eclipse.milo.opcua.stack.core.types.builtin.NodeId;
+import org.eclipse.milo.opcua.stack.core.types.builtin.Variant;
+import org.eclipse.milo.opcua.stack.core.types.enumerated.MessageSecurityMode;
+import org.eclipse.milo.opcua.stack.core.types.structured.EndpointDescription;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assumptions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.lang.management.ManagementFactory;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Opt-in load test for OPC UA pull reads whose server-side processing exceeds
the request timeout.
+ */
+public class OpcUaAdapterReadTimeoutReproductionTest {
+
+ private static final String ENABLE_PROPERTY =
"streampipes.opcua.heap-reproduction";
+ private static final String IMAGE_PROPERTY = "streampipes.opcua.demo.image";
+ private static final String DELAYED_VALUE_NODE =
+ "ns=2;s=Demo.StreamPipesTestCases.ReadFailures.DelayedValue";
+ private static final NodeId DELAY_ENABLED_NODE =
+
NodeId.parse("ns=2;s=Demo.StreamPipesTestCases.ReadFailures.DelayEnabled");
+
+ private static OpcUaDemoServerContainer opcUaContainer;
+
+ @BeforeAll
+ public static void startContainer() {
+ Assumptions.assumeTrue(
+ Boolean.getBoolean(ENABLE_PROPERTY),
+ "Enable explicitly with -D" + ENABLE_PROPERTY + "=true"
+ );
+ String image = System.getProperty(IMAGE_PROPERTY,
"opc-ua-demo-server:heap-repro");
+ opcUaContainer = new OpcUaDemoServerContainer(image);
+ opcUaContainer.start();
+ }
+
+ @AfterAll
+ public static void stopContainer() {
+ if (opcUaContainer != null) {
+ opcUaContainer.stop();
+ }
+ }
+
+ @Test
+ public void expiresDelayedReadsAtTransportLevel() throws Exception {
+ int adapterCount =
Integer.getInteger("streampipes.opcua.reproduction.adapters", 25);
+ int pollingIntervalMillis =
+ Integer.getInteger("streampipes.opcua.reproduction.poll-interval-ms",
100);
+ int durationSeconds =
Integer.getInteger("streampipes.opcua.reproduction.duration-seconds", 120);
+
+ var harness = new OpcUaAdapterTestHarness();
+ var clientProvider = new OpcUaClientProvider();
+ var adapters = new ArrayList<RunningOpcUaAdapter>();
+
+ try {
+ for (int i = 0; i < adapterCount; i++) {
+ String adapterId = "opcua-timeout-reproduction-" + i;
+ adapters.add(
+ harness.startPullAdapter(
+ clientProvider,
+ opcUaContainer.getEndpointUrl(),
+ List.of(DELAYED_VALUE_NODE),
+ pollingIntervalMillis,
+ adapterId
+ )
+ );
+ }
+
+ enableDelayedReads();
+ sampleHeapAndWait(Duration.ofSeconds(durationSeconds));
+
+ var monitoring =
SpMonitoringManager.INSTANCE.getMonitoringInfo().getLogInfos();
+ long adaptersWithTimeout = monitoring.entrySet().stream()
+ .filter(entry ->
entry.getKey().startsWith("opcua-timeout-reproduction-"))
+ .filter(entry -> entry.getValue().stream().anyMatch(logEntry ->
+
logEntry.getErrorMessage().getFullStackTrace().contains("Bad_Timeout")))
+ .count();
+
+ assertTrue(
+ adaptersWithTimeout == adapterCount,
+ () -> "Expected timeout errors for all adapters, but found "
+ + adaptersWithTimeout + " of " + adapterCount
+ );
+ } finally {
+ for (int i = adapters.size() - 1; i >= 0; i--) {
+ adapters.get(i).close();
+ }
+ }
+ }
+
+ private void enableDelayedReads() throws Exception {
+ OpcUaClient controlClient = OpcUaClient.create(
+ opcUaContainer.getEndpointUrl(),
+ endpoints -> endpoints.stream()
+ .filter(endpoint -> endpoint.getSecurityMode() ==
MessageSecurityMode.None)
+ .filter(endpoint ->
SecurityPolicy.None.getUri().equals(endpoint.getSecurityPolicyUri()))
+ .findFirst()
+ .map(this::useMappedEndpoint),
+ transportConfig -> { },
+ clientConfig -> { }
+ );
+ try {
+ controlClient.connect();
+ var statuses = controlClient.writeValues(
+ List.of(DELAY_ENABLED_NODE),
+ List.of(DataValue.valueOnly(Variant.ofBoolean(true)))
+ );
+ assertTrue(statuses.get(0).isGood(), "Could not enable delayed reads in
demo server");
+ } finally {
+ controlClient.disconnect();
+ }
+ }
+
+ private EndpointDescription useMappedEndpoint(EndpointDescription endpoint) {
+ return new EndpointDescription(
+ opcUaContainer.getEndpointUrl(),
+ endpoint.getServer(),
+ endpoint.getServerCertificate(),
+ endpoint.getSecurityMode(),
+ endpoint.getSecurityPolicyUri(),
+ endpoint.getUserIdentityTokens(),
+ endpoint.getTransportProfileUri(),
+ endpoint.getSecurityLevel()
+ );
+ }
+
+ private void sampleHeapAndWait(Duration duration) throws
InterruptedException {
+ var memoryBean = ManagementFactory.getMemoryMXBean();
+ Instant deadline = Instant.now().plus(duration);
+
+ while (Instant.now().isBefore(deadline)) {
+ var heap = memoryBean.getHeapMemoryUsage();
+ System.out.printf(
+ "OPC-UA heap regression: used=%d MiB, committed=%d MiB, max=%d
MiB%n",
+ heap.getUsed() / 1024 / 1024,
+ heap.getCommitted() / 1024 / 1024,
+ heap.getMax() / 1024 / 1024
+ );
+ Thread.sleep(1000);
+ }
+ }
+}
diff --git
a/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/adapters/opcua/OpcUaAdapterTestHarness.java
b/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/adapters/opcua/OpcUaAdapterTestHarness.java
index 4522f2af5d..d332ddf310 100644
---
a/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/adapters/opcua/OpcUaAdapterTestHarness.java
+++
b/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/adapters/opcua/OpcUaAdapterTestHarness.java
@@ -53,7 +53,7 @@ public class OpcUaAdapterTestHarness {
public Map<String, Object> readSingleEvent(String endpointUrl, List<String>
selectedNodeIds) throws Exception {
var collectorQueue = new LinkedBlockingQueue<Map<String, Object>>();
- var extractor = makeExtractor(endpointUrl, selectedNodeIds);
+ var extractor = makeExtractor(endpointUrl, selectedNodeIds, 1000,
"opcua-adapter-it");
var runtimeContext = makeRuntimeContext();
var adapter = new OpcUaAdapter(new OpcUaClientProvider());
var started = false;
@@ -72,7 +72,22 @@ public class OpcUaAdapterTestHarness {
}
}
- private IAdapterParameterExtractor makeExtractor(String endpointUrl,
List<String> selectedNodeIds) {
+ public RunningOpcUaAdapter startPullAdapter(OpcUaClientProvider
clientProvider,
+ String endpointUrl,
+ List<String> selectedNodeIds,
+ int pollingIntervalMillis,
+ String adapterId) throws
Exception {
+ var extractor = makeExtractor(endpointUrl, selectedNodeIds,
pollingIntervalMillis, adapterId);
+ var runtimeContext = makeRuntimeContext();
+ var adapter = new OpcUaAdapter(clientProvider);
+ adapter.onAdapterStarted(extractor, event -> { }, runtimeContext);
+ return new RunningOpcUaAdapter(adapter, extractor, runtimeContext);
+ }
+
+ private IAdapterParameterExtractor makeExtractor(String endpointUrl,
+ List<String>
selectedNodeIds,
+ int pollingIntervalMillis,
+ String adapterId) {
IStaticPropertyExtractor staticExtractor =
mock(IStaticPropertyExtractor.class);
when(staticExtractor.selectedAlternativeInternalId(ADAPTER_TYPE.name()))
@@ -90,7 +105,7 @@ public class OpcUaAdapterTestHarness {
when(staticExtractor.singleValueParameter(OPC_SERVER_URL.name(),
String.class))
.thenReturn(endpointUrl);
when(staticExtractor.singleValueParameter(PULLING_INTERVAL.name(),
Integer.class))
- .thenReturn(1000);
+ .thenReturn(pollingIntervalMillis);
when(staticExtractor.selectedSingleValueInternalName(
SharedUserConfiguration.INCOMPLETE_EVENT_HANDLING_KEY,
String.class
@@ -99,7 +114,7 @@ public class OpcUaAdapterTestHarness {
.thenReturn(OpcUaNamingStrategy.DISPLAY_NAME.name());
AdapterDescription adapterDescription = new AdapterDescription();
- adapterDescription.setElementId("opcua-adapter-it");
+ adapterDescription.setElementId(adapterId);
IAdapterParameterExtractor extractor =
mock(IAdapterParameterExtractor.class);
when(extractor.getStaticPropertyExtractor()).thenReturn(staticExtractor);
@@ -113,5 +128,14 @@ public class OpcUaAdapterTestHarness {
when(runtimeContext.getStreamPipesClient()).thenReturn(streamPipesClient);
return runtimeContext;
}
-}
+ public record RunningOpcUaAdapter(OpcUaAdapter adapter,
+ IAdapterParameterExtractor extractor,
+ IAdapterRuntimeContext runtimeContext)
implements AutoCloseable {
+
+ @Override
+ public void close() throws Exception {
+ adapter.onAdapterStopped(extractor, runtimeContext);
+ }
+ }
+}
diff --git
a/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/containers/OpcUaDemoServerContainer.java
b/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/containers/OpcUaDemoServerContainer.java
index 50ab2dd932..e3c9e7fcec 100644
---
a/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/containers/OpcUaDemoServerContainer.java
+++
b/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/containers/OpcUaDemoServerContainer.java
@@ -27,9 +27,14 @@ import java.time.Duration;
public class OpcUaDemoServerContainer extends
GenericContainer<OpcUaDemoServerContainer> {
public static final int OPC_UA_PORT = 4840;
+ public static final String DEFAULT_IMAGE =
"digitalpetri/opc-ua-demo-server:latest";
public OpcUaDemoServerContainer() {
- super(DockerImageName.parse("digitalpetri/opc-ua-demo-server:latest"));
+ this(DEFAULT_IMAGE);
+ }
+
+ public OpcUaDemoServerContainer(String imageName) {
+ super(DockerImageName.parse(imageName));
}
@Override
