oscerd commented on code in PR #27401:
URL: https://github.com/apache/camel/pull/27401#discussion_r4193696242
##########
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:
**Replacing the InterruptedException drops the exchange's interrupted flag,
so a stop or reload now runs full error handling**
`exchange.setException(e)` with the `InterruptedException` also marked the
exchange as interrupted (`AbstractExchange.setException`: `if (t instanceof
InterruptedException) setInterrupted(true)`), and `RedeliveryErrorHandler` uses
that flag to stop routing without handling a failure (`SimpleTask.run()` →
`runInterrupted()`, and `isDone()`); nothing else in core reads it.
`DirectConsumerNotAvailableException` is not an `InterruptedException`, so the
flag is no longer set. On a route stop or a dev mode reload the cut-off
exchange therefore goes through full error handling: with the default error
handler an ERROR `Failed delivery ... Exhausted after delivery attempt: 1
caught: ...DirectConsumerNotAvailableException` with message history, and with
`onException`, redelivery or a dead letter channel, redelivery attempts and the
DLC. A message that was only cut off by a reload can end up in the DLC, which
is the opposite of what CAMEL-25365 asks for.
When `DefaultAsyncProcessorAwaitManager` swaps an `InterruptedException` for
a clearer exception it sets the flag explicitly, and keeping the
`InterruptedException` as the cause also keeps
`onException(InterruptedException.class)` and
`exchange.getException(InterruptedException.class)` matching (both walk the
cause chain), so no upgrade-guide entry is needed:
```java
DirectConsumerNotAvailableException cause = new
DirectConsumerNotAvailableException(
"No consumers available on endpoint: " + endpoint
+ " (interrupted while waiting for one, as the route is being
stopped or reloaded)", exchange);
cause.initCause(e);
exchange.setException(cause);
// stay marked as interrupted, as setException(InterruptedException) did, so
the error handler stops routing
// instead of handling a failure (onException, redelivery, dead letter
channel, the ERROR log)
exchange.getExchangeExtension().setInterrupted(true);
```
Asserting `failed.getExchange().getExchangeExtension().isInterrupted()` in
`theInterruptedExchangeSaysWhy` would pin it (it fails on the current head).
##########
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:
**The forced-shutdown message mentions dev mode reloads, which never reach
this branch**
`isForceShutdown()` is only true after `shutdownForced()`, which only
`AbstractCamelContext.doStopContext()` calls;
`DefaultShutdownStrategy.doShutdown()` assigns `forceShutdown` from its
parameter, and a route stop or reload goes through `doShutdownRoute()` →
`shutdown(context, routes, timeout, unit)` with `false`. So a dev mode reload
never gets this message, even when its graceful shutdown times out and the
routes are forced; it gets the "being stopped or reloaded" one. The
parenthetical points readers to the wrong cause:
```suggestion
+ " while it was in flight (the CamelContext is being
stopped)"
```
##########
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:
**'Waiting at' filters on fromRouteId, but the count it explains is
per-route occupancy**
The `Inflights per route` count comes from
`inflightRepository.size(routeId)`, maintained for every route an exchange
enters, but this filter keeps only the exchanges *created* by the routes being
stopped. `direct:` reuses the caller's exchange and `fromRouteId` stays the
caller's route, and a reload stops routes one at a time, so when the waiting
route is fed by `direct:` (a common dev layout) the line reads `Inflights per
route: [picked-lines = 1]` with no `Waiting at`. Also accepting the route the
exchange is in now covers it:
```suggestion
if (!routeIds.contains(inflight.getExchange().getFromRouteId())
&& !routeIds.contains(inflight.getAtRouteId())) {
```
Smaller: `getNodeId()` is null between nodes and `getAtRouteId()` before the
route is pushed, so an entry can render as `picked-lines/null`; skipping
entries with a null node id would keep the line clean.
##########
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:
**The await condition stands in for the asserted state, a possible flake**
`size() == 1` becomes true when the unit of work registers the exchange,
which is before the route is pushed and before the `to` channel sets the node
id. If the seda thread is descheduled in that window, `waitingAt(...)` returns
`. Waiting at: picked-lines/null` and the exact `assertEquals` fails, the kind
of race that flakes on busy CI runners. Waiting for the state you assert
removes it:
```suggestion
await().atMost(5, TimeUnit.SECONDS).until(() ->
context.getInflightRepository().browse().stream()
.anyMatch(inflight ->
"shipment".equals(inflight.getNodeId())));
```
##########
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:
**The log wiring, the 3-place cap and the new RejectedExecutionException
messages are untested**
The tests call `DefaultShutdownStrategy.waitingAt(...)` directly, so the
production wiring (`if (!logInflightExchangesOnTimeout) msg += waitingAt(...)`)
is never run: deleting it leaves all three tests green, and
`setLogInflightExchangesOnTimeout(false)` has no effect on what is asserted.
The three-place cap / `and N more`, the `(source)` suffix and the new
`RejectedExecutionException` messages from `notAllowedReason()` have no test
either. Capturing the `DefaultShutdownStrategy` INFO line during
`stopRoute("picked-lines")` with the existing
`ConsumingAppender.newAppender(...)` helper (as
`RestProducerUnresolvedPathWarnTest` does) would cover the real message.
--
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]
