This is an automated email from the ASF dual-hosted git repository. dominikriemer pushed a commit to branch improve-s7-connection-cache in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit 31e8acdb7834aab35bfd47b3c22eeedebad60cce Author: Dominik Riemer <[email protected]> AuthorDate: Thu Jun 11 22:02:07 2026 +0200 fix: Improve cleanup of stale connections --- .../management/connect/PullAdapterScheduler.java | 3 +- .../connect/PullAdapterSchedulerTest.java | 70 ++++++++++++++++++++ .../plc/cache/SpCachedPlcConnectionManager.java | 31 +++++---- .../plc/adapter/ConnectionContainerReproTest.java | 75 ++++++++++++++++++++++ 4 files changed, 166 insertions(+), 13 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 1c501fe288..38693d065f 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 @@ -26,7 +26,6 @@ import org.apache.streampipes.model.monitoring.SpLogMessage; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.concurrent.CompletionException; import java.util.concurrent.ExecutionException; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; @@ -48,7 +47,7 @@ public class PullAdapterScheduler { final Runnable task = () -> { try { pullAdapter.pullData(); - } catch (ExecutionException | InterruptedException | TimeoutException | CompletionException e) { + } catch (ExecutionException | InterruptedException | TimeoutException | RuntimeException e) { LOG.error("Error while pulling data: {}", e.getMessage()); SpMonitoringManager.INSTANCE.addErrorMessage( adapterElementId, 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 new file mode 100644 index 0000000000..c6ef7df54f --- /dev/null +++ b/streampipes-extensions-management/src/test/java/org/apache/streampipes/extensions/management/connect/PullAdapterSchedulerTest.java @@ -0,0 +1,70 @@ +/* + * 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.management.connect; + +import org.apache.streampipes.extensions.api.connect.IPollingSettings; +import org.apache.streampipes.extensions.api.connect.IPullAdapter; +import org.apache.streampipes.extensions.management.connect.adapter.util.PollingSettings; + +import org.junit.jupiter.api.Test; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertTrue; + +class PullAdapterSchedulerTest { + + @Test + void continuesSchedulingAfterRuntimeException() throws InterruptedException { + var scheduler = new PullAdapterScheduler(); + var secondInvocation = new CountDownLatch(1); + + try { + scheduler.schedule(new FailingOncePullAdapter(secondInvocation), "adapter-id"); + assertTrue(secondInvocation.await(2, TimeUnit.SECONDS)); + } finally { + scheduler.shutdown(); + } + } + + private static class FailingOncePullAdapter implements IPullAdapter { + + private final AtomicInteger invocations = new AtomicInteger(); + private final CountDownLatch secondInvocation; + + private FailingOncePullAdapter(CountDownLatch secondInvocation) { + this.secondInvocation = secondInvocation; + } + + @Override + public void pullData() { + if (invocations.incrementAndGet() == 1) { + throw new IllegalStateException("first poll failed"); + } + secondInvocation.countDown(); + } + + @Override + public IPollingSettings getPollingInterval() { + return PollingSettings.from(TimeUnit.MILLISECONDS, 10); + } + } +} diff --git a/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/cache/SpCachedPlcConnectionManager.java b/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/cache/SpCachedPlcConnectionManager.java index e3702aec10..1f702d0913 100644 --- a/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/cache/SpCachedPlcConnectionManager.java +++ b/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/cache/SpCachedPlcConnectionManager.java @@ -81,12 +81,14 @@ public class SpCachedPlcConnectionManager implements PlcConnectionManager, AutoC * @param url url of the connection that should be removed. */ public void removeCachedConnection(String url) { + SpConnectionContainer connectionContainer; synchronized (connectionContainers) { - // Make sure the connection is closed before removing it. - if (connectionContainers.containsKey(url)) { - connectionContainers.get(url).close(); - } - connectionContainers.remove(url); + connectionContainer = connectionContainers.remove(url); + } + + // Make sure the connection is closed before removing it. + if (connectionContainer != null) { + connectionContainer.close(); } } @@ -119,8 +121,13 @@ public class SpCachedPlcConnectionManager implements PlcConnectionManager, AutoC Future<PlcConnection> leaseFuture = connectionContainer.lease(); try { return leaseFuture.get(this.maxWaitTime.toMillis(), TimeUnit.MILLISECONDS); - } catch (ExecutionException | InterruptedException | TimeoutException e) { - throw new PlcConnectionException("Error acquiring lease for connection"); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new PlcConnectionException("Interrupted while acquiring lease for connection", e); + } catch (TimeoutException e) { + throw new PlcConnectionException("Timed out acquiring lease for connection", e); + } catch (ExecutionException e) { + throw new PlcConnectionException("Error acquiring lease for connection", e.getCause()); } } @@ -134,9 +141,12 @@ public class SpCachedPlcConnectionManager implements PlcConnectionManager, AutoC closed.set(true); // Tell all connections to close themselves. - connectionContainers.forEach((connectionString, connectionContainer) -> { - connectionContainer.close(); - }); + Map<String, SpConnectionContainer> containersToClose; + synchronized (connectionContainers) { + containersToClose = new HashMap<>(connectionContainers); + connectionContainers.clear(); + } + containersToClose.forEach((connectionString, connectionContainer) -> connectionContainer.close()); } public static class Builder { @@ -175,4 +185,3 @@ public class SpCachedPlcConnectionManager implements PlcConnectionManager, AutoC } } - diff --git a/streampipes-extensions/streampipes-connectors-plc/src/test/java/org/apache/streampipes/extensions/connectors/plc/adapter/ConnectionContainerReproTest.java b/streampipes-extensions/streampipes-connectors-plc/src/test/java/org/apache/streampipes/extensions/connectors/plc/adapter/ConnectionContainerReproTest.java index aa96155be1..efecf2e69c 100644 --- a/streampipes-extensions/streampipes-connectors-plc/src/test/java/org/apache/streampipes/extensions/connectors/plc/adapter/ConnectionContainerReproTest.java +++ b/streampipes-extensions/streampipes-connectors-plc/src/test/java/org/apache/streampipes/extensions/connectors/plc/adapter/ConnectionContainerReproTest.java @@ -18,6 +18,7 @@ package org.apache.streampipes.extensions.connectors.plc.adapter; +import org.apache.streampipes.extensions.connectors.plc.cache.SpCachedPlcConnectionManager; import org.apache.streampipes.extensions.connectors.plc.cache.SpConnectionContainer; import org.apache.streampipes.extensions.connectors.plc.cache.SpLeasedPlcConnection; @@ -39,7 +40,10 @@ import org.junit.jupiter.api.Test; import java.time.Duration; import java.util.Optional; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -47,6 +51,7 @@ import java.util.concurrent.atomic.AtomicInteger; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; class ConnectionContainerReproTest { @@ -134,6 +139,27 @@ class ConnectionContainerReproTest { // implement other methods as no-ops if your interface requires them } + static class BlockingCloseConnection extends DummyConnection { + private final CountDownLatch closeStarted; + private final CountDownLatch releaseClose; + + BlockingCloseConnection(CountDownLatch closeStarted, + CountDownLatch releaseClose) { + this.closeStarted = closeStarted; + this.releaseClose = releaseClose; + } + + @Override + public void close() { + closeStarted.countDown(); + try { + releaseClose.await(5, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + } + @Test void recoversAfterFailedReconnectAndServesNewLeases() throws Exception { FlakyManager mgr = new FlakyManager(); @@ -166,4 +192,53 @@ class ConnectionContainerReproTest { PlcConnection lease3 = cc.lease().get(500, TimeUnit.MILLISECONDS); assertNotNull(lease3); } + + @Test + void removingSlowConnectionDoesNotBlockLeasesForOtherUrls() throws Exception { + var closeStarted = new CountDownLatch(1); + var releaseClose = new CountDownLatch(1); + PlcConnectionManager manager = new PlcConnectionManager() { + @Override + public PlcConnection getConnection(String url) { + if ("mock://slow".equals(url)) { + return new BlockingCloseConnection(closeStarted, releaseClose); + } + return new DummyConnection(); + } + + @Override + public PlcConnection getConnection(String url, + PlcAuthentication authentication) { + return null; + } + }; + + var cachedConnectionManager = new SpCachedPlcConnectionManager( + manager, + Duration.ofSeconds(30), + Duration.ofSeconds(30), + Duration.ofSeconds(30) + ); + + cachedConnectionManager.getConnection("mock://slow"); + + ExecutorService removeExecutor = Executors.newSingleThreadExecutor(); + ExecutorService leaseExecutor = Executors.newSingleThreadExecutor(); + try { + Future<?> removeFuture = removeExecutor.submit( + () -> cachedConnectionManager.removeCachedConnection("mock://slow")); + assertTrue(closeStarted.await(500, TimeUnit.MILLISECONDS)); + + Future<PlcConnection> otherLease = leaseExecutor.submit( + () -> cachedConnectionManager.getConnection("mock://other")); + assertNotNull(otherLease.get(500, TimeUnit.MILLISECONDS)); + + releaseClose.countDown(); + removeFuture.get(500, TimeUnit.MILLISECONDS); + } finally { + releaseClose.countDown(); + removeExecutor.shutdownNow(); + leaseExecutor.shutdownNow(); + } + } }
