This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new b18e4520df39 CAMEL-26339: camel-plc4x - log a run of failed polls once
and stop polling a lost connection without autoReconnect (#28556)
b18e4520df39 is described below
commit b18e4520df39341f5134d79eae85d320023fd8f1
Author: Serdar Gökay <[email protected]>
AuthorDate: Tue Oct 27 13:17:43 2026 +0300
CAMEL-26339: camel-plc4x - log a run of failed polls once and stop polling
a lost connection without autoReconnect (#28556)
The triggered plc4x consumer now installs its own default exception
handler: the
first failed poll of a run is logged at WARN with the stack trace, the
following
ones at DEBUG with the message only, and the first poll that reads the PLC
again
logs one INFO line with the number of failed polls. A configured
exceptionHandler
or bridgeErrorHandler=true still replaces it, and every poll still fails,
so the
health check sees each failure.
With autoReconnect=false, Plc4XEndpoint.reconnectIfNeeded() shuts a lost
connection down once instead of on every call, and the triggered consumer
then
stops polling (unscheduleTask, as repeatCount does) until the route is
restarted. The upgrade guide documents the new behaviour.
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/catalog/docs/plc4x-component.adoc | 10 +-
.../camel-plc4x/src/main/docs/plc4x-component.adoc | 10 +-
.../camel/component/plc4x/Plc4XEndpoint.java | 16 +-
.../component/plc4x/Plc4XTriggeredConsumer.java | 55 +++++-
.../apache/camel/component/plc4x/LogCapture.java | 73 +++++++
.../component/plc4x/Plc4XConsumerRouteTest.java | 34 ----
.../plc4x/Plc4XRepeatedFailureRouteTest.java | 217 +++++++++++++++++++++
.../apache/camel/component/plc4x/PollFailures.java | 62 ++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 3 +-
9 files changed, 435 insertions(+), 45 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/plc4x-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/plc4x-component.adoc
index 37f10570ac51..c273c34409d3 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/plc4x-component.adoc
+++
b/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:
diff --git a/components/camel-plc4x/src/main/docs/plc4x-component.adoc
b/components/camel-plc4x/src/main/docs/plc4x-component.adoc
index 37f10570ac51..c273c34409d3 100644
--- a/components/camel-plc4x/src/main/docs/plc4x-component.adoc
+++ b/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:
diff --git
a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java
b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java
index c541fb899801..2fd507d342a7 100644
---
a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java
+++
b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java
@@ -169,7 +169,8 @@ public class Plc4XEndpoint extends DefaultEndpoint
implements EndpointServiceLoc
}
/**
- * 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 class Plc4XEndpoint extends DefaultEndpoint
implements EndpointServiceLoc
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 class Plc4XEndpoint extends DefaultEndpoint
implements EndpointServiceLoc
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);
diff --git
a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XTriggeredConsumer.java
b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XTriggeredConsumer.java
index a742ac2a878b..09af4edcdb5d 100644
---
a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XTriggeredConsumer.java
+++
b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XTriggeredConsumer.java
@@ -22,18 +22,25 @@ import java.util.concurrent.ExecutionException;
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 class Plc4XTriggeredConsumer extends
ScheduledPollConsumer {
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 @@ public class Plc4XTriggeredConsumer extends
ScheduledPollConsumer {
// 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 @@ public class Plc4XTriggeredConsumer extends
ScheduledPollConsumer {
}
/**
- * 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 @@ public class Plc4XTriggeredConsumer extends
ScheduledPollConsumer {
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());
+ }
+ }
+ }
}
diff --git
a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/LogCapture.java
b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/LogCapture.java
new file mode 100644
index 000000000000..d195ec71c720
--- /dev/null
+++
b/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();
+ }
+}
diff --git
a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerRouteTest.java
b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerRouteTest.java
index 6a94252023fd..6ad5fff4ecd9 100644
---
a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerRouteTest.java
+++
b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerRouteTest.java
@@ -19,13 +19,10 @@ package org.apache.camel.component.plc4x;
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 @@ public class Plc4XConsumerRouteTest extends
CamelTestSupport {
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;
- }
- }
-
}
diff --git
a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XRepeatedFailureRouteTest.java
b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XRepeatedFailureRouteTest.java
new file mode 100644
index 000000000000..4e897e9762a3
--- /dev/null
+++
b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XRepeatedFailureRouteTest.java
@@ -0,0 +1,217 @@
+/*
+ * 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.Map;
+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.Registry;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.core.LogEvent;
+import org.apache.plc4x.java.api.exceptions.PlcConnectionException;
+import org.apache.plc4x.java.api.exceptions.PlcException;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * How a route reports a PLC that keeps failing its requests: the triggered
consumer logs the first failed poll of a run
+ * at WARN and the end of the run at INFO, instead of a WARN with a stack
trace on every poll, while
+ * {@code bridgeErrorHandler} still gets every failure. Without {@code
autoReconnect}, a lost connection shuts the
+ * endpoint down once, and the triggered consumer stops polling until the
route is restarted.
+ */
+public class Plc4XRepeatedFailureRouteTest extends CamelTestSupport {
+
+ private static final String PLC =
"plc4x:mock:10.10.10.1/1/1?trigger=TRIG&period=50&tag.temperature=TEMP";
+ private static final String BRIDGED = PLC + "&bridgeErrorHandler=true";
+ private static final String REPORTED = PLC +
"&exceptionHandler=#pollFailures";
+ private static final String WRITE = "plc4x:mock:10.10.10.2/1/1";
+ private static final String SHUTDOWN = "Connection lost and auto-reconnect
is turned off";
+ private static final String POLLING_STOPS = "polling stops until the route
is restarted";
+
+ private final PollFailures pollFailures = new PollFailures();
+
+ @BeforeEach
+ void resetMockDriver() {
+ MockDriver.reset();
+ }
+
+ @AfterAll
+ static void leaveMockDriverUnprogrammed() {
+ MockDriver.reset();
+ }
+
+ @Test
+ public void logsARunOfFailedPollsOnceAtWarnAndItsEndOnceAtInfo() throws
Exception {
+ Object failed = MockDriver.failedRead();
+ // two runs of failed polls, each followed by successful polls, and a
rising edge after them
+ MockDriver.programRead("TRIG", failed, failed, failed, false, false,
false, failed, failed, false, false, true);
+ MockDriver.programRead("TEMP", 21);
+ MockEndpoint result = getMockEndpoint("mock:logged");
+ result.expectedMessageCount(1);
+
+ try (LogCapture log = new LogCapture(Plc4XTriggeredConsumer.class)) {
+ context.getRouteController().startRoute("logged");
+ result.assertIsSatisfied(10000);
+
+ List<LogEvent> events = log.events();
+ assertEquals(List.of(Level.WARN, Level.DEBUG, Level.DEBUG,
Level.INFO, Level.WARN, Level.DEBUG, Level.INFO),
+ events.stream().map(LogEvent::getLevel).toList(), () ->
"Unexpected log: " + messages(events));
+ assertEquals(5, log.eventsContaining("Failed polling
endpoint").size(),
+ () -> "Unexpected log: " + messages(events));
+ assertNotNull(events.get(0).getThrown(), "the first failed poll of
a run is logged with its stack trace");
+ assertNotNull(events.get(4).getThrown(), "the first failed poll of
a run is logged with its stack trace");
+ assertNull(events.get(1).getThrown(), "a following failed poll of
a run is logged without its stack trace");
+ // the error counter of the consumer still counts each failed poll
of a run
+ String firstEnd = events.get(3).getMessage().getFormattedMessage();
+ assertTrue(firstEnd.endsWith("after 3 failed polls"), firstEnd);
+ String secondEnd =
events.get(6).getMessage().getFormattedMessage();
+ assertTrue(secondEnd.endsWith("after 2 failed polls"), secondEnd);
+ }
+ }
+
+ @Test
+ public void
bridgeErrorHandlerRoutesEveryFailedPollToTheErrorHandlerOfTheRoute() throws
Exception {
+ Object failed = MockDriver.failedRead();
+ MockDriver.programRead("TRIG", failed, failed, failed, false);
+ MockDriver.programRead("TEMP", 21);
+ MockEndpoint failures = getMockEndpoint("mock:failures");
+ failures.expectedMessageCount(3);
+ failures.setAssertPeriod(500);
+
+ context.getRouteController().startRoute("bridged");
+
+ failures.assertIsSatisfied(10000);
+ assertTrue(failures.getExchanges().stream()
+ .allMatch(exchange ->
exchange.getProperty(Exchange.EXCEPTION_CAUGHT) instanceof PlcException),
+ () -> "Unexpected failures: " + failures.getExchanges());
+ }
+
+ @Test
+ public void
aLostConnectionWithoutAutoReconnectStopsPollingUntilTheRouteIsRestarted()
throws Exception {
+ MockDriver.programRead("TRIG", false);
+ MockDriver.programRead("TEMP", 21);
+ MockEndpoint result = getMockEndpoint("mock:reported");
+
+ try (LogCapture log = new LogCapture(Plc4XEndpoint.class)) {
+ context.getRouteController().startRoute("reported");
+ Plc4XTriggeredConsumer consumer = (Plc4XTriggeredConsumer)
context.getRoute("reported").getConsumer();
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
consumer.getSuccessCounter() >= 2);
+
+ MockDriver.dropConnections();
+
+ // the poll that finds the connection lost fails, and no poll
follows it
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
!pollFailures.failures().isEmpty());
+ await().during(500, TimeUnit.MILLISECONDS).atMost(1,
TimeUnit.SECONDS)
+ .untilAsserted(() -> assertEquals(1,
pollFailures.failures().size(), "the failed polls"));
+ assertEquals(1, log.eventsContaining(SHUTDOWN).size(), "the
endpoint is shut down once");
+ assertEquals(1, consumer.getErrorCounter(), "the failed poll keeps
the health check of the consumer down");
+ PollFailures.Failure failure = pollFailures.failures().get(0);
+ assertInstanceOf(PlcConnectionException.class, failure.cause());
+ assertFalse(failure.interrupted(), "the failure is reported on a
thread that is not interrupted");
+
+ // the restarted route sets up a new connection and polls again
+ MockDriver.programRead("TRIG", false, true);
+ result.expectedMessageCount(1);
+ context.getRouteController().stopRoute("reported");
+ context.getRouteController().startRoute("reported");
+ result.assertIsSatisfied(10000);
+
+ // and a loss of the new connection shuts the endpoint down again
+ MockDriver.dropConnections();
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
log.eventsContaining(SHUTDOWN).size() == 2);
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
pollFailures.failures().size() == 2);
+ }
+ }
+
+ @Test
+ public void logsTheFailedPollThatStopsPollingAtWarnAfterAFailedRead()
throws Exception {
+ MockDriver.programRead("TRIG", false, MockDriver.failedRead());
+ MockDriver.programRead("TEMP", 21);
+
+ try (LogCapture log = new LogCapture(Plc4XTriggeredConsumer.class)) {
+ context.getRouteController().startRoute("logged");
+ Plc4XTriggeredConsumer consumer = (Plc4XTriggeredConsumer)
context.getRoute("logged").getConsumer();
+ // a read fails as the PLC goes away, and a later poll finds the
connection lost within the same run
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
consumer.getErrorCounter() >= 1);
+
+ MockDriver.dropConnections();
+
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
!log.eventsContaining(POLLING_STOPS).isEmpty());
+ assertEquals(List.of(Level.WARN),
+
log.eventsContaining(POLLING_STOPS).stream().map(LogEvent::getLevel).toList(),
+ () -> "Unexpected log: " + messages(log.events()));
+ // the log line of the poll says that no next poll follows it
+ String stopped =
log.eventsContaining(POLLING_STOPS).get(0).getMessage().getFormattedMessage();
+ assertTrue(stopped.startsWith(
+ "Failed polling endpoint: " + consumer.getEndpoint() + ".
Polling stops until the route is restarted"),
+ stopped);
+ }
+ }
+
+ @Test
+ public void aProducerReportsALostConnectionWithoutAutoReconnectOnce()
throws Exception {
+ Map<String, Map<String, Object>> tags = Map.of("temperature",
Map.of("%DB1:DINT", 21));
+ template.sendBody(WRITE, tags);
+
+ try (LogCapture log = new LogCapture(Plc4XEndpoint.class)) {
+ MockDriver.dropConnections();
+
+ for (int i = 0; i < 3; i++) {
+ Exchange exchange = template.send(WRITE, e ->
e.getIn().setBody(tags));
+ assertInstanceOf(PlcConnectionException.class,
exchange.getException(), "each write fails");
+ }
+ assertEquals(1, log.eventsContaining(SHUTDOWN).size(), "the
endpoint is shut down once");
+ }
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from(PLC).routeId("logged").autoStartup(false).to("mock:logged");
+ from(BRIDGED).routeId("bridged").autoStartup(false)
+
.onException(Exception.class).handled(true).to("mock:failures").end()
+ .to("mock:bridged");
+
from(REPORTED).routeId("reported").autoStartup(false).to("mock:reported");
+ }
+ };
+ }
+
+ @Override
+ protected void bindToRegistry(Registry registry) {
+ registry.bind("pollFailures", pollFailures);
+ }
+
+ private static List<String> messages(List<LogEvent> events) {
+ return events.stream().map(event -> event.getLevel() + " " +
event.getMessage().getFormattedMessage()).toList();
+ }
+}
diff --git
a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/PollFailures.java
b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/PollFailures.java
new file mode 100644
index 000000000000..d6abf39f764b
--- /dev/null
+++
b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/PollFailures.java
@@ -0,0 +1,62 @@
+/*
+ * 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.camel.Exchange;
+import org.apache.camel.spi.ExceptionHandler;
+
+/**
+ * Collects the failures a consumer reports to its exception handler, and
whether the thread that reported each one was
+ * interrupted.
+ */
+final class PollFailures implements ExceptionHandler {
+
+ private final List<Failure> failures = new CopyOnWriteArrayList<>();
+
+ @Override
+ public void handleException(Throwable exception) {
+ handleException(null, null, exception);
+ }
+
+ @Override
+ public void handleException(String message, Throwable exception) {
+ handleException(message, null, exception);
+ }
+
+ @Override
+ public void handleException(String message, Exchange exchange, Throwable
exception) {
+ failures.add(new Failure(exception,
Thread.currentThread().isInterrupted()));
+ }
+
+ List<Failure> failures() {
+ return failures;
+ }
+
+ int count() {
+ return failures.size();
+ }
+
+ List<Throwable> causes() {
+ return failures.stream().map(Failure::cause).toList();
+ }
+
+ record Failure(Throwable cause, boolean interrupted) {
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index a2e88f6e9c37..64ee1374f3fd 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -3743,7 +3743,8 @@ and read the tags in a later, separate request on another
connection. A failed r
handler of the consumer (or to the error handler of the route with
`bridgeErrorHandler=true`), and polling continues;
before, the consumer stopped triggering after the first failed read of the
trigger. After a failed poll the remembered
trigger value is unknown, so a trigger that is still `true` when the PLC
returns emits once. Since the consumer now uses
-the connection of the endpoint, `autoReconnect` governs its reconnects. A
connection that cannot be set up when the
+the connection of the endpoint, `autoReconnect` governs its reconnects:
without it, a lost connection stops
+the endpoint and the polling until the route is restarted. A connection that
cannot be set up when the
route starts fails the start of the route, unless `autoReconnect=true`, in
which case the polls retry it. A route that
starts with the CamelContext then fails the start of the CamelContext too,
unless the route is supervised by the
xref:route-controller.adoc[supervising route controller] or has
`autoStartup=false`. As before, `period` must be
