This is an automated email from the ASF dual-hosted git repository.
dominikriemer pushed a commit to branch fix-plc-retry-failed-connection
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to
refs/heads/fix-plc-retry-failed-connection by this push:
new af6f2e1bc9 Fix connection recovery
new 423a21a7c6 Merge branch 'fix-plc-retry-failed-connection' of
github.com:apache/streampipes into fix-plc-retry-failed-connection
af6f2e1bc9 is described below
commit af6f2e1bc9109908643440c93fe2ee308fa9e5b3
Author: Dominik Riemer <[email protected]>
AuthorDate: Mon Jun 22 10:16:48 2026 +0200
Fix connection recovery
---
.../plc/cache/SpConnectionContainer.java | 50 +++++++++++-----------
.../plc/adapter/ConnectionContainerReproTest.java | 39 +++++++++++++++++
2 files changed, 63 insertions(+), 26 deletions(-)
diff --git
a/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/cache/SpConnectionContainer.java
b/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/cache/SpConnectionContainer.java
index 67eda0f222..ec1f0371a7 100644
---
a/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/cache/SpConnectionContainer.java
+++
b/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/cache/SpConnectionContainer.java
@@ -169,27 +169,23 @@ public class SpConnectionContainer {
return;
}
- // If something happened while using the connection, invalidate this one
and create a new connection.
+ // If something happened while using the connection, invalidate it.
if (invalidateConnection) {
// Close the old connection.
- try {
- connection.close();
- } catch (Exception e) {
- // We're ignoring this as we have no idea, what state the connection
is in.
- // Nevertheless, it is polite to say something in logs about this
situation.
- LOGGER.warn("Exception while closing connection", e);
- }
+ closeConnection(connection);
+ connection = null;
- // Try to get a new connection.
- try {
- connection = connectionManager.getConnection(connectionUrl);
- } catch (PlcConnectionException e) {
- // If something goes wrong, close all waiting futures exceptionally.
- LOGGER.warn("Can't get connection for {} complete queue items
exceptionally", connectionUrl, e);
- queue.forEach(future -> future.completeExceptionally(e));
- queue.clear();
- leasedConnection = null;
- connection = null;
+ // Only reconnect immediately when another client is waiting for the
connection.
+ if (!queue.isEmpty()) {
+ try {
+ connection = connectionManager.getConnection(connectionUrl);
+ } catch (PlcConnectionException e) {
+ // If something goes wrong, close all waiting futures exceptionally.
+ LOGGER.warn("Can't get connection for {} complete queue items
exceptionally", connectionUrl, e);
+ queue.forEach(future -> future.completeExceptionally(e));
+ queue.clear();
+ leasedConnection = null;
+ }
}
}
@@ -197,14 +193,16 @@ public class SpConnectionContainer {
if (queue.isEmpty()) {
leasedConnection = null;
- // Start a timer to invalidate this connection if it's idle for too long.
- idleTimer = new Timer("CC-Idle-Timer-" + Thread.currentThread().getId());
- idleTimer.schedule(new TimerTask() {
- @Override
- public void run() {
- closeIdleConnection();
- }
- }, maxIdleTime.toMillis());
+ if (connection != null) {
+ // Start a timer to invalidate this connection if it's idle for too
long.
+ idleTimer = new Timer("CC-Idle-Timer-" +
Thread.currentThread().getId());
+ idleTimer.schedule(new TimerTask() {
+ @Override
+ public void run() {
+ closeIdleConnection();
+ }
+ }, maxIdleTime.toMillis());
+ }
return;
}
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 a4573832b6..49decd93cd 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
@@ -265,6 +265,45 @@ class ConnectionContainerReproTest {
connectionContainer.close();
}
+ @Test
+ void doesNotEagerlyReplaceInvalidConnectionWithoutWaitingClient() throws
Exception {
+ var firstConnection = new MutableConnection(true);
+ var secondConnection = new MutableConnection(true);
+ var managerCalls = new AtomicInteger();
+ PlcConnectionManager manager = new PlcConnectionManager() {
+ @Override
+ public PlcConnection getConnection(String url) {
+ return managerCalls.incrementAndGet() == 1 ? firstConnection :
secondConnection;
+ }
+
+ @Override
+ public PlcConnection getConnection(String url,
+ PlcAuthentication authentication) {
+ return null;
+ }
+ };
+ var connectionContainer = new SpConnectionContainer(
+ manager,
+ "mock://plc",
+ Duration.ofSeconds(30),
+ Duration.ofSeconds(30),
+ url -> null
+ );
+
+ SpLeasedPlcConnection firstLease =
+ (SpLeasedPlcConnection) connectionContainer.lease().get(500,
TimeUnit.MILLISECONDS);
+ connectionContainer.returnConnection(firstLease, true);
+
+ assertEquals(1, managerCalls.get());
+ assertEquals(1, firstConnection.closeCalls());
+
+ SpLeasedPlcConnection secondLease =
+ (SpLeasedPlcConnection) connectionContainer.lease().get(500,
TimeUnit.MILLISECONDS);
+ assertEquals(2, managerCalls.get());
+ connectionContainer.returnConnection(secondLease, false);
+ connectionContainer.close();
+ }
+
@Test
void removingSlowConnectionDoesNotBlockLeasesForOtherUrls() throws Exception
{
var closeStarted = new CountDownLatch(1);
