davsclaus commented on code in PR #27401:
URL: https://github.com/apache/camel/pull/27401#discussion_r4193881653
##########
components/camel-direct/src/main/java/org/apache/camel/component/direct/DirectProducer.java:
##########
@@ -99,9 +99,15 @@ public boolean process(Exchange exchange, AsyncCallback
callback) {
}
}
} catch (InterruptedException e) {
- LOG.info("Interrupted while processing the exchange");
+ // the only wait here is for a consumer to appear (block=true),
and what interrupts it is a forced shutdown,
+ // such as a dev mode reload that adds the consumer in the same
edit (CAMEL-25365)
+ LOG.info("Interrupted while waiting for a consumer on {}: the
route is being stopped or reloaded",
+ endpoint.getEndpointUri());
Thread.currentThread().interrupt();
- exchange.setException(e);
+ exchange.setException(new DirectConsumerNotAvailableException(
Review Comment:
Good catch, thanks. Fixed in f06fb9c720: the `InterruptedException` is now
the cause of the `DirectConsumerNotAvailableException`, and the exchange is
marked interrupted as `setException(InterruptedException)` did, so the error
handler stops routing instead of running `onException`, redelivery or the dead
letter channel. `theInterruptedExchangeSaysWhy` now asserts `isInterrupted()`
and that `getException(InterruptedException.class)` finds the cause.
_Claude Code on behalf of davsclaus_
##########
core/camel-core-processor/src/main/java/org/apache/camel/processor/errorhandler/RedeliveryErrorHandler.java:
##########
@@ -2208,4 +2208,15 @@ private static String failureOrigin(Exchange exchange) {
return sb.toString();
}
+ /**
+ * Why an exchange cannot go on: its route is being stopped. Without a
message the error reads
+ * {@code RejectedExecutionException - null}, which looks like a fault in
the route, while it is a route stop or a
+ * dev mode reload cutting the exchange off (CAMEL-25365).
+ */
+ String notAllowedReason() {
+ return shutdownStrategy.isForceShutdown()
+ ? "The exchange cannot continue: its route was forced to shut
down, as the graceful shutdown timed out"
+ + " while it was in flight (a route reload in dev mode does
this when an exchange waits for something)"
Review Comment:
Right, only a CamelContext stop forces. Applied your wording in f06fb9c720,
and `RedeliveryNotAllowedReasonTest` pins both reasons.
_Claude Code on behalf of davsclaus_
##########
core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultShutdownStrategy.java:
##########
@@ -801,6 +805,41 @@ protected static int
getPendingInflightExchanges(RouteStartupOrder order, boolea
return (int) Math.min(Integer.MAX_VALUE, inflight);
}
+ /**
+ * Where the inflight exchanges of the routes are, for the one-line
waiting message: route, node and the line in the
+ * source, such as {@code picked-lines/to1 (aggregator.camel.yaml:23)}. A
dev mode reload waits here for exchanges
+ * blocked on something only the reload itself would add, such as a direct
endpoint whose consumer is part of the
+ * same edit, and without the place the wait is a mystery (CAMEL-25365).
Empty when the inflight repository cannot
+ * be browsed.
+ */
+ static String waitingAt(CamelContext camelContext, List<RouteStartupOrder>
routes) {
+ if (!camelContext.getInflightRepository().isInflightBrowseEnabled()) {
+ return "";
+ }
+ Set<String> routeIds = new HashSet<>();
+ for (RouteStartupOrder route : routes) {
+ routeIds.add(route.getRoute().getId());
+ }
+ Set<String> places = new LinkedHashSet<>();
+ int more = 0;
+ for (InflightRepository.InflightExchange inflight :
camelContext.getInflightRepository().browse()) {
+ if (!routeIds.contains(inflight.getExchange().getFromRouteId())) {
Review Comment:
Applied in f06fb9c720: an inflight exchange counts when its from route or
the route it is in now is being stopped, and entries without a route or node id
yet are left out. `anExchangeThatCameInThroughDirectIsFoundInTheStoppedRoute`
covers a caller route feeding the stopped route through `direct:`.
_Claude Code on behalf of davsclaus_
##########
core/camel-core/src/test/java/org/apache/camel/impl/engine/ShutdownWaitingAtTest.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.impl.engine;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.direct.DirectConsumerNotAvailableException;
+import org.apache.camel.spi.CamelEvent;
+import org.apache.camel.support.EventNotifierSupport;
+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.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * CAMEL-25365: a route stopped while its exchanges wait for a direct
consumer, as a dev mode reload that adds the
+ * consumer in the same edit does. The shutdown says where they wait, and an
interrupted exchange says why it failed.
+ */
+public class ShutdownWaitingAtTest extends ContextTestSupport {
+
+ @Test
+ public void theWaitSaysWhereTheExchangesAre() throws Exception {
+ context.getInflightRepository().setInflightBrowseEnabled(true);
+ context.getShutdownStrategy().setTimeout(2);
+ context.getShutdownStrategy().setTimeUnit(TimeUnit.SECONDS);
+ context.getShutdownStrategy().setLogInflightExchangesOnTimeout(false);
+
+ template.sendBody("seda:start", "A");
+ await().atMost(5, TimeUnit.SECONDS).until(() ->
context.getInflightRepository().size() == 1);
Review Comment:
Applied in f06fb9c720: the tests now wait until an exchange is at the
asserted node (`awaitAt("shipment")`), not for the inflight count.
_Claude Code on behalf of davsclaus_
##########
core/camel-core/src/test/java/org/apache/camel/impl/engine/ShutdownWaitingAtTest.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.impl.engine;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.direct.DirectConsumerNotAvailableException;
+import org.apache.camel.spi.CamelEvent;
+import org.apache.camel.support.EventNotifierSupport;
+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.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * CAMEL-25365: a route stopped while its exchanges wait for a direct
consumer, as a dev mode reload that adds the
+ * consumer in the same edit does. The shutdown says where they wait, and an
interrupted exchange says why it failed.
+ */
+public class ShutdownWaitingAtTest extends ContextTestSupport {
+
+ @Test
+ public void theWaitSaysWhereTheExchangesAre() throws Exception {
+ context.getInflightRepository().setInflightBrowseEnabled(true);
+ context.getShutdownStrategy().setTimeout(2);
+ context.getShutdownStrategy().setTimeUnit(TimeUnit.SECONDS);
+ context.getShutdownStrategy().setLogInflightExchangesOnTimeout(false);
+
+ template.sendBody("seda:start", "A");
+ await().atMost(5, TimeUnit.SECONDS).until(() ->
context.getInflightRepository().size() == 1);
+
+ String at = DefaultShutdownStrategy.waitingAt(context,
context.getCamelContextExtension().getRouteStartupOrder());
Review Comment:
Added in f06fb9c720: `theShutdownLogSaysWhereTheExchangesAre` captures the
real `DefaultShutdownStrategy` INFO line with `ConsumingAppender` during
`stopRoute`, and `theVerboseListingLeavesThePlaceOut` checks that it is not
added when the verbose listing is on. `atMostThreePlacesAreNamed` covers the
cap with `and 1 more`, the `(source)` suffix is asserted, and
`RedeliveryNotAllowedReasonTest` covers both `RejectedExecutionException`
reasons.
_Claude Code on behalf of davsclaus_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
