(streampipes) branch dev updated: fix(#3099): Remove broken connection from connection pool (#3100)

[riemer]

This is an automated email from the ASF dual-hosted git repository.

riemer 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 ddcbc18b02 fix(#3099): Remove broken connection from connection pool 
(#3100)
ddcbc18b02 is described below

commit ddcbc18b026de76fef37ca47f60e71d24c5987f1
Author: Dominik Riemer <dominik.rie...@gmail.com>
AuthorDate: Sat Aug 3 21:01:35 2024 +0200

    fix(#3099): Remove broken connection from connection pool (#3100)
---
 .../management/connect/PullAdapterScheduler.java   | 38 ++++++++++++++--------
 .../connection/ContinuousPlcRequestReader.java     |  4 +++
 2 files changed, 29 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 de9a1b4146..7795d40bfb 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
@@ -33,27 +33,39 @@ import java.util.concurrent.TimeoutException;
 
 public class PullAdapterScheduler {
 
-  protected static final Logger LOGGER = 
LoggerFactory.getLogger(PullAdapterScheduler.class);
+  protected static final Logger LOG = 
LoggerFactory.getLogger(PullAdapterScheduler.class);
 
   private ScheduledExecutorService scheduler;
 
   public void schedule(IPullAdapter pullAdapter,
                        String adapterElementId) {
-    final Runnable task = () -> {
-      try {
-        pullAdapter.pullData();
-      } catch (ExecutionException | InterruptedException e) {
-        SpMonitoringManager.INSTANCE.addErrorMessage(
-            adapterElementId,
-            SpLogEntry.from(System.currentTimeMillis(), SpLogMessage.from(e)));
-      } catch (TimeoutException e) {
-        LOGGER.warn("Timeout occurred", e);
+    scheduler = Executors.newSingleThreadScheduledExecutor();
+    final Runnable task = new Runnable() {
+      @Override
+      public void run() {
+        try {
+          pullAdapter.pullData();
+        } catch (ExecutionException | InterruptedException | TimeoutException 
e) {
+          LOG.error("Error while pulling data", e);
+          SpMonitoringManager.INSTANCE.addErrorMessage(
+              adapterElementId,
+              SpLogEntry.from(System.currentTimeMillis(), SpLogMessage.from(e))
+          );
+        } finally {
+          scheduler.schedule(
+              this,
+              pullAdapter.getPollingInterval().value(),
+              pullAdapter.getPollingInterval().timeUnit()
+          );
+        }
       }
     };
 
-    scheduler = Executors.newScheduledThreadPool(1);
-    scheduler.scheduleAtFixedRate(task, 1,
-        pullAdapter.getPollingInterval().value(), 
pullAdapter.getPollingInterval().timeUnit());
+    scheduler.schedule(
+        task,
+        pullAdapter.getPollingInterval().value(),
+        pullAdapter.getPollingInterval().timeUnit()
+    );
   }
 
   public void shutdown() {
diff --git 
a/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/generic/connection/ContinuousPlcRequestReader.java
 
b/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/generic/connection/ContinuousPlcRequestReader.java
index befc54a234..80086d20d2 100644
--- 
a/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/generic/connection/ContinuousPlcRequestReader.java
+++ 
b/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/generic/connection/ContinuousPlcRequestReader.java
@@ -55,6 +55,10 @@ public class ContinuousPlcRequestReader
       var event = eventGenerator.makeEvent(readResponse);
       collector.collect(event);
     } catch (Exception e) {
+      // ensure that the cached connection manager removes the broken 
connection
+      if (connectionManager instanceof CachedPlcConnectionManager) {
+        ((CachedPlcConnectionManager) 
connectionManager).removeCachedConnection(settings.connectionString());
+      }
       LOG.error("Error while reading from PLC with connection string {} ", 
settings.connectionString(), e);
     }
   }