From d8dc1989b6998c8e9f5a9cc0f600d424caf0140f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Serdar=20G=C3=B6kay?= Date: Tue, 27 Oct 2026 10:50:25 +0300 Subject: [PATCH 1/3] CAMEL-26339: camel-plc4x - log a run of failed polls of the triggered consumer once The triggered consumer logged every failed poll at WARN with the stack trace, so a PLC that cannot be reached logged a stack trace on every poll. Its default exception handler now logs the first failed poll of a run at WARN with the stack trace and the following ones at DEBUG without it, and the consumer logs at INFO when a poll succeeds again. A configured exceptionHandler or bridgeErrorHandler still gets every failure, and the failed polls still count for the health check. Co-authored-by: Claude Opus 5.5 --- .../camel/catalog/docs/plc4x-component.adoc | 3 +- .../src/main/docs/plc4x-component.adoc | 3 +- .../plc4x/Plc4XTriggeredConsumer.java | 35 ++++++ .../camel/component/plc4x/LogCapture.java | 73 +++++++++++ .../plc4x/Plc4XRepeatedFailureRouteTest.java | 119 ++++++++++++++++++ 5 files changed, 231 insertions(+), 2 deletions(-) create mode 100644 components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/LogCapture.java create mode 100644 components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XRepeatedFailureRouteTest.java 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 37f10570ac510..1fc4293f60af1 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,7 +101,8 @@ 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 diff --git a/components/camel-plc4x/src/main/docs/plc4x-component.adoc b/components/camel-plc4x/src/main/docs/plc4x-component.adoc index 37f10570ac510..1fc4293f60af1 100644 --- a/components/camel-plc4x/src/main/docs/plc4x-component.adoc +++ b/components/camel-plc4x/src/main/docs/plc4x-component.adoc @@ -101,7 +101,8 @@ 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 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 a742ac2a878b2..432a285d78e7b 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,24 @@ 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.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 +55,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 +94,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; @@ -147,4 +159,27 @@ 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, 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. + */ + 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 + 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 0000000000000..d195ec71c720b --- /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 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 events() { + return List.copyOf(events); + } + + /** + * @return the events logged so far whose message contains the given text + */ + List 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/Plc4XRepeatedFailureRouteTest.java b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XRepeatedFailureRouteTest.java new file mode 100644 index 0000000000000..f9f3475b3ecf7 --- /dev/null +++ b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XRepeatedFailureRouteTest.java @@ -0,0 +1,119 @@ +/* + * 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 org.apache.camel.Exchange; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +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.PlcException; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +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. + */ +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"; + + @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 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()); + } + + @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"); + } + }; + } + + private static List messages(List events) { + return events.stream().map(event -> event.getLevel() + " " + event.getMessage().getFormattedMessage()).toList(); + } +} From b323cda28436c0208de132895dbd1bf5f0126e94 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Serdar=20G=C3=B6kay?= Date: Tue, 27 Oct 2026 10:56:54 +0300 Subject: [PATCH 2/3] CAMEL-26339: camel-plc4x - shut down a lost connection once and stop polling without autoReconnect Without autoReconnect, every call of reconnectIfNeeded() after a lost connection logged the shutdown WARN and stopped the endpoint again, while the triggered consumer kept polling. The endpoint now shuts down once per lost connection, and not when no connection could be set up, so the producer and the polling consumer fail their exchanges without repeating the WARN. The triggered consumer fails the poll that finds the connection lost and stops polling until the route is restarted, so its health check stays down; its default exception handler logs that failure at WARN, even after other failed polls. Unscheduling from the poll interrupts the polling thread, so that interrupt is cleared before the failure is reported. Co-authored-by: Claude Opus 5.5 --- .../camel/catalog/docs/plc4x-component.adoc | 7 +- .../src/main/docs/plc4x-component.adoc | 7 +- .../camel/component/plc4x/Plc4XEndpoint.java | 9 +- .../plc4x/Plc4XTriggeredConsumer.java | 23 ++++- .../plc4x/Plc4XConsumerRouteTest.java | 34 ------- .../plc4x/Plc4XRepeatedFailureRouteTest.java | 95 ++++++++++++++++++- .../camel/component/plc4x/PollFailures.java | 62 ++++++++++++ .../pages/camel-4x-upgrade-guide-4_24.adoc | 3 +- 8 files changed, 192 insertions(+), 48 deletions(-) create mode 100644 components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/PollFailures.java 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 1fc4293f60af1..c273c34409d3b 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 @@ -105,9 +105,10 @@ continues. The default exception handler logs the first failed poll of a run at 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 1fc4293f60af1..c273c34409d3b 100644 --- a/components/camel-plc4x/src/main/docs/plc4x-component.adoc +++ b/components/camel-plc4x/src/main/docs/plc4x-component.adoc @@ -105,9 +105,10 @@ continues. The default exception handler logs the first failed poll of a run at 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 c541fb8998018..c0e66c6d8266e 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 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. *

* * @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"); 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 432a285d78e7b..29784c4d66545 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 @@ -27,6 +27,7 @@ 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; @@ -123,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.connection == null) { + // 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 future = plc4XEndpoint.buildPlcReadRequest(Map.of(Constants.TRIGGER, trigger)).execute(); PlcReadResponse response; @@ -162,8 +175,8 @@ public String toString() { /** * 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, 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. + * 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. */ private final class FailedPollExceptionHandler extends LoggingExceptionHandler { FailedPollExceptionHandler(CamelContext camelContext) { @@ -173,8 +186,8 @@ private final class FailedPollExceptionHandler extends LoggingExceptionHandler { @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 - if (getErrorCounter() == 0) { + // error counter of 0, and the one that stops the polling finds its task unscheduled + if (getErrorCounter() == 0 || !isSchedulerStarted()) { super.handleException(message, exception); } else { // the first failed poll of the run is logged with the stack trace already 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 6a94252023fda..6ad5fff4ecd99 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 @@ 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 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 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 index f9f3475b3ecf7..3d1092fd8d0e3 100644 --- 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 @@ -17,19 +17,26 @@ 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; @@ -37,12 +44,19 @@ /** * 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. + * {@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() { @@ -100,6 +114,79 @@ public void bridgeErrorHandlerRoutesEveryFailedPollToTheErrorHandlerOfTheRoute() () -> "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())); + } + } + + @Test + public void aProducerReportsALostConnectionWithoutAutoReconnectOnce() throws Exception { + Map> 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() { @@ -109,10 +196,16 @@ public void configure() { 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 messages(List 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 0000000000000..d6abf39f764bf --- /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 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 failures() { + return failures; + } + + int count() { + return failures.size(); + } + + List 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 0be69c592ec19..43be76d680975 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 @@ -3628,7 +3628,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 From f26f939b804c57f3d4cbba41235c6fc76b0c8043 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Serdar=20G=C3=B6kay?= Date: Tue, 27 Oct 2026 12:00:02 +0300 Subject: [PATCH 3/3] CAMEL-26339: camel-plc4x - say that polling stops in the log line of the poll that stops it The failed poll that stops the polling of the triggered consumer logged the message of the scheduled poll consumer, which announces a next poll. It now says that polling stops until the route is restarted. The consumer asks the endpoint whether it has a connection through hasConnection() instead of reading the connection field. Co-authored-by: Claude Opus 5.5 --- .../apache/camel/component/plc4x/Plc4XEndpoint.java | 7 +++++++ .../camel/component/plc4x/Plc4XTriggeredConsumer.java | 11 ++++++++--- .../plc4x/Plc4XRepeatedFailureRouteTest.java | 5 +++++ 3 files changed, 20 insertions(+), 3 deletions(-) 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 c0e66c6d8266e..2fd507d342a75 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 @@ -229,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); 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 29784c4d66545..09af4edcdb5d5 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 @@ -130,7 +130,7 @@ protected int poll() throws Exception { 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.connection == null) { + 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(); @@ -176,7 +176,8 @@ public String toString() { /** * 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. + * 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) { @@ -187,7 +188,11 @@ private final class FailedPollExceptionHandler extends LoggingExceptionHandler { 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 (getErrorCounter() == 0 || !isSchedulerStarted()) { + 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 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 index 3d1092fd8d0e3..4e897e9762a3f 100644 --- 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 @@ -168,6 +168,11 @@ public void logsTheFailedPollThatStopsPollingAtWarnAfterAFailedRead() throws Exc 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); } }