CAMEL-26339: camel-plc4x - log a run of failed polls once and stop polling a lost connection without autoReconnect

catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/plc4x-component.adoc

@@ -101,12 +101,14 @@ may use. While the trigger stays `true`, no further exchange is delivered.
 A poll that fails, for example because a tag cannot be read, the PLC cannot be reached or does not answer within five
 seconds, or the trigger reads as a value that is not a boolean (such as a number or an array), is reported to the
 exception handler of the consumer (or to the error handler of the route with `bridgeErrorHandler=true`), and polling
-continues.
+continues. The default exception handler logs the first failed poll of a run at WARN with the stack trace and the
+following ones at DEBUG without it; the first successful poll after a run of failed polls is logged at INFO.
 The trigger value read before the failure is forgotten, so a trigger that is `true` at the first successful poll after
 a failure counts as a rising edge. Automatic recovery after a lost connection needs `autoReconnect=true`: a poll then
-replaces a lost connection with a new one. With the default `autoReconnect=false`, the endpoint stops on the first lost
-connection and does not recover. Likewise, a connection that cannot be set up when the route starts is retried by the
-polls with `autoReconnect=true`, and fails the start of the route without it.
+replaces a lost connection with a new one. With the default `autoReconnect=false`, the endpoint stops on each lost
+connection, and the consumer fails that poll and stops polling until the route is restarted. Likewise, a connection
+that cannot be set up when the route starts is retried by the polls with `autoReconnect=true`, and fails the start of
+the route without it.
 
 The following route reads a register of a Modbus TCP device each time coil 1 goes from `false` to `true`, checking the
 coil every 100 milliseconds:

components/camel-plc4x/src/main/docs/plc4x-component.adoc

@@ -101,12 +101,14 @@ may use. While the trigger stays `true`, no further exchange is delivered.
 A poll that fails, for example because a tag cannot be read, the PLC cannot be reached or does not answer within five
 seconds, or the trigger reads as a value that is not a boolean (such as a number or an array), is reported to the
 exception handler of the consumer (or to the error handler of the route with `bridgeErrorHandler=true`), and polling
-continues.
+continues. The default exception handler logs the first failed poll of a run at WARN with the stack trace and the
+following ones at DEBUG without it; the first successful poll after a run of failed polls is logged at INFO.
 The trigger value read before the failure is forgotten, so a trigger that is `true` at the first successful poll after
 a failure counts as a rising edge. Automatic recovery after a lost connection needs `autoReconnect=true`: a poll then
-replaces a lost connection with a new one. With the default `autoReconnect=false`, the endpoint stops on the first lost
-connection and does not recover. Likewise, a connection that cannot be set up when the route starts is retried by the
-polls with `autoReconnect=true`, and fails the start of the route without it.
+replaces a lost connection with a new one. With the default `autoReconnect=false`, the endpoint stops on each lost
+connection, and the consumer fails that poll and stops polling until the route is restarted. Likewise, a connection
+that cannot be set up when the route starts is retried by the polls with `autoReconnect=true`, and fails the start of
+the route without it.
 
 The following route reads a register of a Modbus TCP device each time coil 1 goes from `false` to `true`, checking the
 coil every 100 milliseconds:

components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java

@@ -169,7 +169,8 @@ public void setupConnection() throws PlcConnectionException {
     }
 
     /**
-     * Reconnects if needed. If connection is lost and auto-reconnect is turned off, endpoint will be shutdown.
+     * Reconnects if needed. If connection is lost and auto-reconnect is turned off, endpoint will be shutdown, once for
+     * each lost connection.
      * <p>
      *
      * @throws PlcConnectionException If reconnect failed and auto-reconnect is turned on
@@ -189,6 +190,12 @@ public void reconnectIfNeeded() throws PlcConnectionException {
                 LOGGER.debug("Successfully reconnected");
                 return;
             }
+            if (connection == null) {
+                // no connection left to lose, as when an earlier call shut the lost one down or none could be set up
+                return;
+            }
+            // closed under the lock, so that only one caller shuts the endpoint down for the lost connection
+            closeConnection();
         }
         // stop() is called outside the lock, as a concurrent stop() holds the service lock while doStop() waits for it
         LOGGER.warn("Connection lost and auto-reconnect is turned off, shutting down Plc4XEndpoint");
@@ -222,6 +229,13 @@ public boolean isConnected() {
         return current != null && current.isConnected();
     }
 
+    /**
+     * @return false if the lost connection was shut down, or none was set up
+     */
+    boolean hasConnection() {
+        return connection != null;
+    }
+
     @Override
     public Producer createProducer() {
         return new Plc4XProducer(this);

components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XTriggeredConsumer.java

@@ -22,18 +22,25 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
+import org.apache.camel.CamelContext;
 import org.apache.camel.Exchange;
 import org.apache.camel.Processor;
+import org.apache.camel.support.LoggingExceptionHandler;
 import org.apache.camel.support.ScheduledPollConsumer;
+import org.apache.plc4x.java.api.exceptions.PlcConnectionException;
 import org.apache.plc4x.java.api.exceptions.PlcException;
 import org.apache.plc4x.java.api.messages.PlcReadResponse;
 import org.apache.plc4x.java.api.types.PlcResponseCode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Reads the trigger tag together with the tags every {@code period} milliseconds and emits the tags once on each rising
  * edge of the trigger.
  */
 public class Plc4XTriggeredConsumer extends ScheduledPollConsumer {
+    private static final Logger LOGGER = LoggerFactory.getLogger(Plc4XTriggeredConsumer.class);
+
     private final Plc4XEndpoint plc4XEndpoint;
     private final String trigger;
     private final int period;
@@ -49,6 +56,8 @@ public Plc4XTriggeredConsumer(Plc4XEndpoint endpoint, Processor processor) {
         setInitialDelay(0);
         setDelay(period);
         setTimeUnit(TimeUnit.MILLISECONDS);
+        // the endpoint replaces it when an exceptionHandler or bridgeErrorHandler is configured
+        setExceptionHandler(new FailedPollExceptionHandler(endpoint.getCamelContext()));
     }
 
     @Override
@@ -86,6 +95,10 @@ protected int poll() throws Exception {
             // the consumer stopped while the read was in progress
             return 0;
         }
+        if (getErrorCounter() > 0) {
+            // the error counter is reset after this poll, so it still counts the failed polls before it
+            LOGGER.info("Polling endpoint: {} succeeded again after {} failed polls", getEndpoint(), getErrorCounter());
+        }
         boolean triggered = (Boolean) response.getObject(Constants.TRIGGER);
         boolean risingEdge = triggered && !Boolean.TRUE.equals(lastTrigger);
         lastTrigger = triggered;
@@ -111,10 +124,22 @@ protected int poll() throws Exception {
     }
 
     /**
-     * Reads the trigger and the tags in one request; fails unless every tag was read and the trigger is a boolean.
+     * Reads the trigger and the tags in one request; fails unless every tag was read and the trigger is a boolean. A
+     * connection that is lost without autoReconnect also stops the polling.
      */
     private PlcReadResponse read() throws Exception {
         plc4XEndpoint.reconnectIfNeeded();
+        // a connection that is lost after reconnectIfNeeded() fails the read instead, and the next poll shuts it down
+        if (!plc4XEndpoint.isAutoReconnect() && !plc4XEndpoint.hasConnection()) {
+            // the endpoint shut the lost connection down, so no later poll could read: polling stops until the route is
+            // restarted, and this poll fails, so the health check of the consumer stays down
+            getScheduler().unscheduleTask();
+            // unscheduling cancels the task this poll runs in, which interrupts this thread: clear it, so that the
+            // failure is reported as for any other poll
+            Thread.interrupted();
+            throw new PlcConnectionException(
+                    "Connection lost and auto-reconnect is turned off, polling stops until the route is restarted");
+        }
         CompletableFuture<? extends PlcReadResponse> future
                 = plc4XEndpoint.buildPlcReadRequest(Map.of(Constants.TRIGGER, trigger)).execute();
         PlcReadResponse response;
@@ -147,4 +172,32 @@ private PlcReadResponse read() throws Exception {
     public String toString() {
         return "Plc4XTriggeredConsumer[" + plc4XEndpoint + "]";
     }
+
+    /**
+     * Logs like the default exception handler of a consumer, but a failed poll only at WARN with the stack trace when
+     * it starts a run of failed polls or stops the polling, and the following failed polls of the run at DEBUG without
+     * it, so that a PLC that cannot be reached is not logged on every poll. The failed poll that stops the polling says
+     * so, instead of announcing a next poll.
+     */
+    private final class FailedPollExceptionHandler extends LoggingExceptionHandler {
+        FailedPollExceptionHandler(CamelContext camelContext) {
+            super(camelContext, Plc4XTriggeredConsumer.class);
+        }
+
+        @Override
+        public void handleException(String message, Throwable exception) {
+            // a failed poll is reported without an exchange and before it is counted, so the first one of a run finds an
+            // error counter of 0, and the one that stops the polling finds its task unscheduled
+            if (!isSchedulerStarted()) {
+                super.handleException(
+                        "Failed polling endpoint: " + getEndpoint() + ". Polling stops until the route is restarted",
+                        exception);
+            } else if (getErrorCounter() == 0) {
+                super.handleException(message, exception);
+            } else {
+                // the first failed poll of the run is logged with the stack trace already
+                LOGGER.debug("{}. Caused by: {}", message, exception.getMessage());
+            }
+        }
+    }
 }

components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/LogCapture.java

@@ -0,0 +1,73 @@
+/*
+ * 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.camel.component.plc4x;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.core.LogEvent;
+import org.apache.logging.log4j.core.Logger;
+import org.apache.logging.log4j.core.appender.AbstractAppender;
+import org.apache.logging.log4j.core.config.Property;
+
+/**
+ * Captures the log events of one logger, down to DEBUG, while it is open. Only one capture is open at a time: opening
+ * another one resets the level of the loggers.
+ */
+final class LogCapture extends AbstractAppender implements AutoCloseable {
+
+    private final List<LogEvent> events = new CopyOnWriteArrayList<>();
+    private final Logger logger;
+    private final Level previousLevel;
+
+    LogCapture(Class<?> loggerClass) {
+        super("LogCapture-" + loggerClass.getSimpleName(), null, null, true, Property.EMPTY_ARRAY);
+        this.logger = (Logger) LogManager.getLogger(loggerClass);
+        this.previousLevel = logger.getLevel();
+        start();
+        logger.addAppender(this);
+        logger.setLevel(Level.DEBUG);
+    }
+
+    @Override
+    public void append(LogEvent event) {
+        events.add(event.toImmutable());
+    }
+
+    /**
+     * @return the events logged so far, in the order they were logged
+     */
+    List<LogEvent> events() {
+        return List.copyOf(events);
+    }
+
+    /**
+     * @return the events logged so far whose message contains the given text
+     */
+    List<LogEvent> eventsContaining(String text) {
+        return events.stream().filter(e -> e.getMessage().getFormattedMessage().contains(text)).toList();
+    }
+
+    @Override
+    public void close() {
+        logger.removeAppender(this);
+        logger.setLevel(previousLevel);
+        stop();
+    }
+}

components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerRouteTest.java

@@ -19,13 +19,10 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.TimeUnit;
 
-import org.apache.camel.Exchange;
 import org.apache.camel.builder.RouteBuilder;
 import org.apache.camel.component.mock.MockEndpoint;
-import org.apache.camel.spi.ExceptionHandler;
 import org.apache.camel.spi.Registry;
 import org.apache.camel.test.junit6.CamelTestSupport;
 import org.apache.plc4x.java.api.exceptions.PlcConnectionException;
@@ -363,35 +360,4 @@ private static Throwable causeOfType(Throwable exception, Class<? extends Throwa
         return cause;
     }
 
-    /**
-     * Collects the failures the consumer reports to its exception handler.
-     */
-    private static final class PollFailures implements ExceptionHandler {
-
-        private final List<Throwable> causes = new CopyOnWriteArrayList<>();
-
-        @Override
-        public void handleException(Throwable exception) {
-            causes.add(exception);
-        }
-
-        @Override
-        public void handleException(String message, Throwable exception) {
-            causes.add(exception);
-        }
-
-        @Override
-        public void handleException(String message, Exchange exchange, Throwable exception) {
-            causes.add(exception);
-        }
-
-        int count() {
-            return causes.size();
-        }
-
-        List<Throwable> causes() {
-            return causes;
-        }
-    }
-
 }