This is an automated email from the ASF dual-hosted git repository.
Croway pushed a commit to branch camel-4.18.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.18.x by this push:
new 2b4aea645524 CAMEL-23469: Fix infinite redelivery loop when child
route removes headers (#23090)
2b4aea645524 is described below
commit 2b4aea645524f9551ade02c106270889afd81d67
Author: Federico Mariani <[email protected]>
AuthorDate: Sun May 10 12:02:02 2026 +0200
CAMEL-23469: Fix infinite redelivery loop when child route removes headers
(#23090)
* CAMEL-23469: Fix infinite redelivery loop when child route removes headers
RedeliveryErrorHandler.incrementRedeliveryCounter() read the counter from
the CamelRedeliveryCounter exchange header. When a child route with
NoErrorHandler used removeHeaders("*"), the header was wiped during each
redelivery attempt, causing the counter to reset to 1 and loop forever.
Fix incrementRedeliveryCounter() and prepareExchangeForRedelivery() to
use the internal redeliveryCounter field as the authoritative source
instead of reading from the mutable exchange header.
* Fix redelivery state resilience to header removal
Move decrementRedeliveryCounter() from the outer RedeliveryErrorHandler
class into RedeliveryTask where the internal redeliveryCounter field is
accessible. Uses the same Math.max(header, field) pattern as
incrementRedeliveryCounter() to respect nested error handler counters
while falling back to the internal field when headers are wiped.
Add redeliveryCounter and redeliveryMaxCounter fields to
ExchangeExtension so redelivery state survives removeHeaders("*").
RedeliveryErrorHandler now populates these fields alongside headers.
PollEnricher reads from ExchangeExtension instead of headers for its
save/restore pattern around aggregation with bridgeErrorHandler.
* Fix saga coordinator ID resilience to header removal
SagaProcessor stores the coordinator ID exclusively as the
Long-Running-Action message header. If a sub-route calls
removeHeaders("*"), the coordinator lookup returns null — causing
silent saga corruption (REQUIRED), hard failure (MANDATORY), or
broken compensation (REQUIRES_NEW).
Dual-store the coordinator ID in ExchangeExtension alongside the
header. getCurrentSagaCoordinator() reads from extension first, then
falls back to the header for LRA protocol interoperability.
InMemorySagaCoordinator also sets the extension field when creating
exchanges for compensation/completion.
* Address review: drop "internal" from ExchangeExtension javadocs
These are now the official API for redelivery counter and saga
coordinator state. The corresponding headers are set for backward
compatibility.
---
.../java/org/apache/camel/ExchangeExtension.java | 33 ++++++
.../org/apache/camel/processor/PollEnricher.java | 24 ++--
.../errorhandler/RedeliveryErrorHandler.java | 74 +++++++------
.../apache/camel/processor/saga/SagaProcessor.java | 12 +-
...eliverToSubRouteRemoveHeadersDecrementTest.java | 106 ++++++++++++++++++
.../RedeliverToSubRouteRemoveHeadersTest.java | 121 +++++++++++++++++++++
.../camel/processor/SagaRemoveHeadersTest.java | 90 +++++++++++++++
.../apache/camel/saga/InMemorySagaCoordinator.java | 1 +
.../org/apache/camel/support/ExchangeHelper.java | 3 +
.../camel/support/ExtendedExchangeExtension.java | 36 ++++++
10 files changed, 450 insertions(+), 50 deletions(-)
diff --git
a/core/camel-api/src/main/java/org/apache/camel/ExchangeExtension.java
b/core/camel-api/src/main/java/org/apache/camel/ExchangeExtension.java
index 7f65b63e4a87..fa889c8a94ef 100644
--- a/core/camel-api/src/main/java/org/apache/camel/ExchangeExtension.java
+++ b/core/camel-api/src/main/java/org/apache/camel/ExchangeExtension.java
@@ -128,6 +128,39 @@ public interface ExchangeExtension {
*/
void setRedeliveryExhausted(boolean redeliveryExhausted);
+ /**
+ * Gets the redelivery counter. Returns -1 if not set. The corresponding
header is set for backward compatibility.
+ */
+ int getRedeliveryCounter();
+
+ /**
+ * Sets the redelivery counter. Set to -1 to clear. The corresponding
header is set for backward compatibility.
+ */
+ void setRedeliveryCounter(int redeliveryCounter);
+
+ /**
+ * Gets the maximum redelivery counter. Returns -1 if not set. The
corresponding header is set for backward
+ * compatibility.
+ */
+ int getRedeliveryMaxCounter();
+
+ /**
+ * Sets the maximum redelivery counter. Set to -1 to clear. The
corresponding header is set for backward
+ * compatibility.
+ */
+ void setRedeliveryMaxCounter(int redeliveryMaxCounter);
+
+ /**
+ * Gets the saga long running action ID. Returns null if not set. The
corresponding header is set for backward
+ * compatibility.
+ */
+ String getSagaLongRunningAction();
+
+ /**
+ * Sets the saga long running action ID. The corresponding header is set
for backward compatibility.
+ */
+ void setSagaLongRunningAction(String sagaLongRunningAction);
+
/**
* Checks if the passed {@link Synchronization} instance is already
contained on this exchange.
*
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/PollEnricher.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/PollEnricher.java
index ecddf04a03d2..50b83dca05e0 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/PollEnricher.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/PollEnricher.java
@@ -357,10 +357,10 @@ public class PollEnricher extends BaseProcessorSupport
implements IdAware, Route
}
}
- // remember current redelivery stats
- Object redelivered = exchange.getIn().getHeader(Exchange.REDELIVERED);
- Object redeliveryCounter =
exchange.getIn().getHeader(Exchange.REDELIVERY_COUNTER);
- Object redeliveryMaxCounter =
exchange.getIn().getHeader(Exchange.REDELIVERY_MAX_COUNTER);
+ // remember current redelivery stats from internal state
+ // (headers may have been stripped by removeHeaders("*"))
+ int savedRedeliveryCounter =
exchange.getExchangeExtension().getRedeliveryCounter();
+ int savedRedeliveryMaxCounter =
exchange.getExchangeExtension().getRedeliveryMaxCounter();
// if we are bridging error handler and failed then remember the
caused exception
Throwable cause = null;
@@ -420,15 +420,15 @@ public class PollEnricher extends BaseProcessorSupport
implements IdAware, Route
// remove the exhausted marker as we want to be able to
perform redeliveries with the error handler
exchange.getExchangeExtension().setRedeliveryExhausted(false);
- // preserve the redelivery stats
- if (redelivered != null) {
- exchange.getMessage().setHeader(Exchange.REDELIVERED,
redelivered);
+ // preserve the redelivery stats from internal state
+ if (savedRedeliveryCounter >= 0) {
+
exchange.getMessage().setHeader(Exchange.REDELIVERY_COUNTER,
savedRedeliveryCounter);
+ exchange.getMessage().setHeader(Exchange.REDELIVERED,
savedRedeliveryCounter > 0);
+
exchange.getExchangeExtension().setRedeliveryCounter(savedRedeliveryCounter);
}
- if (redeliveryCounter != null) {
-
exchange.getMessage().setHeader(Exchange.REDELIVERY_COUNTER, redeliveryCounter);
- }
- if (redeliveryMaxCounter != null) {
-
exchange.getMessage().setHeader(Exchange.REDELIVERY_MAX_COUNTER,
redeliveryMaxCounter);
+ if (savedRedeliveryMaxCounter >= 0) {
+
exchange.getMessage().setHeader(Exchange.REDELIVERY_MAX_COUNTER,
savedRedeliveryMaxCounter);
+
exchange.getExchangeExtension().setRedeliveryMaxCounter(savedRedeliveryMaxCounter);
}
}
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/errorhandler/RedeliveryErrorHandler.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/errorhandler/RedeliveryErrorHandler.java
index 199b38395e58..67ac00995dd9 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/errorhandler/RedeliveryErrorHandler.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/errorhandler/RedeliveryErrorHandler.java
@@ -543,6 +543,8 @@ public abstract class RedeliveryErrorHandler extends
ErrorHandlerSupport
exchange.getIn().removeHeader(Exchange.REDELIVERY_COUNTER);
exchange.getIn().removeHeader(Exchange.REDELIVERY_MAX_COUNTER);
exchange.getExchangeExtension().setRedeliveryExhausted(false);
+ exchange.getExchangeExtension().setRedeliveryCounter(-1);
+ exchange.getExchangeExtension().setRedeliveryMaxCounter(-1);
// and remove traces of rollback only and uow exhausted markers
exchange.setRollbackOnly(false);
@@ -1275,29 +1277,24 @@ public abstract class RedeliveryErrorHandler extends
ErrorHandlerSupport
// clear rollback flags
exchange.setRollbackOnly(false);
- // TODO: We may want to store these as state on RedeliveryData so
we keep them in case end user messes with Exchange
- // and then put these on the exchange when doing a redelivery /
fault processor
-
- // preserve these headers
- Integer redeliveryCounter =
exchange.getIn().getHeader(Exchange.REDELIVERY_COUNTER, Integer.class);
- Integer redeliveryMaxCounter =
exchange.getIn().getHeader(Exchange.REDELIVERY_MAX_COUNTER, Integer.class);
- Boolean redelivered =
exchange.getIn().getHeader(Exchange.REDELIVERED, Boolean.class);
-
// we are redelivering so copy from original back to exchange
exchange.getIn().copyFrom(this.original.getIn());
exchange.setOut(null);
// reset cached streams so they can be read again
MessageHelper.resetStreamCache(exchange.getIn());
- // put back headers
- if (redeliveryCounter != null) {
+ // restore redelivery headers from internal state (not from the
exchange headers,
+ // which user routes may have removed via removeHeaders("*"))
+ if (redeliveryCounter > 0) {
exchange.getIn().setHeader(Exchange.REDELIVERY_COUNTER,
redeliveryCounter);
- }
- if (redeliveryMaxCounter != null) {
- exchange.getIn().setHeader(Exchange.REDELIVERY_MAX_COUNTER,
redeliveryMaxCounter);
- }
- if (redelivered != null) {
- exchange.getIn().setHeader(Exchange.REDELIVERED, redelivered);
+ exchange.getIn().setHeader(Exchange.REDELIVERED, Boolean.TRUE);
+
exchange.getExchangeExtension().setRedeliveryCounter(redeliveryCounter);
+ if (currentRedeliveryPolicy.getMaximumRedeliveries() > 0) {
+ exchange.getIn().setHeader(Exchange.REDELIVERY_MAX_COUNTER,
+ currentRedeliveryPolicy.getMaximumRedeliveries());
+ exchange.getExchangeExtension().setRedeliveryMaxCounter(
+ currentRedeliveryPolicy.getMaximumRedeliveries());
+ }
}
}
@@ -1459,6 +1456,8 @@ public abstract class RedeliveryErrorHandler extends
ErrorHandlerSupport
exchange.getIn().removeHeader(Exchange.REDELIVERY_COUNTER);
exchange.getIn().removeHeader(Exchange.REDELIVERY_MAX_COUNTER);
exchange.getExchangeExtension().setRedeliveryExhausted(false);
+ exchange.getExchangeExtension().setRedeliveryCounter(-1);
+ exchange.getExchangeExtension().setRedeliveryMaxCounter(-1);
// and remove traces of rollback only and uow exhausted markers
exchange.setRollbackOnly(false);
@@ -1468,7 +1467,7 @@ public abstract class RedeliveryErrorHandler extends
ErrorHandlerSupport
} else {
// must decrement the redelivery counter as we didn't process
the redelivery but is
// handling by the failure handler. So we must -1 to not let
the counter be out-of-sync
- decrementRedeliveryCounter(exchange);
+ decrementRedeliveryCounter();
}
// we should allow using the failure processor if we should not
continue
@@ -1854,17 +1853,38 @@ public abstract class RedeliveryErrorHandler extends
ErrorHandlerSupport
*/
private int incrementRedeliveryCounter(Exchange exchange) {
Message in = exchange.getIn();
+ // use the internal field as the authoritative counter, not the
header,
+ // because user routes may remove headers (e.g. removeHeaders("*"))
+ // however, if the header carries a higher value from a nested
error handler
+ // (e.g. per-endpoint redelivery in routing slip), respect that
accumulated count
Integer counter = in.getHeader(Exchange.REDELIVERY_COUNTER,
Integer.class);
- int next = counter != null ? counter + 1 : 1;
+ int base = counter != null ? Math.max(redeliveryCounter, counter)
: redeliveryCounter;
+ int next = base + 1;
in.setHeader(Exchange.REDELIVERY_COUNTER, next);
in.setHeader(Exchange.REDELIVERED, Boolean.TRUE);
+ exchange.getExchangeExtension().setRedeliveryCounter(next);
// if maximum redeliveries is used, then provide that information
as well
if (currentRedeliveryPolicy.getMaximumRedeliveries() > 0) {
in.setHeader(Exchange.REDELIVERY_MAX_COUNTER,
currentRedeliveryPolicy.getMaximumRedeliveries());
+
exchange.getExchangeExtension().setRedeliveryMaxCounter(currentRedeliveryPolicy.getMaximumRedeliveries());
}
return next;
}
+ /**
+ * Decrements the redelivery counter when the failure handler takes
over without processing the redelivery.
+ */
+ private void decrementRedeliveryCounter() {
+ Message in = exchange.getIn();
+ // use the same Math.max(header, field) pattern as
incrementRedeliveryCounter:
+ // a nested error handler may have bumped the counter in the
header beyond our field
+ Integer counter = in.getHeader(Exchange.REDELIVERY_COUNTER,
Integer.class);
+ int base = counter != null ? Math.max(redeliveryCounter, counter)
: redeliveryCounter;
+ int prev = Math.max(base - 1, 0);
+ in.setHeader(Exchange.REDELIVERY_COUNTER, prev);
+ in.setHeader(Exchange.REDELIVERED, prev > 0 ? Boolean.TRUE :
Boolean.FALSE);
+ }
+
/**
* Method for sleeping during redelivery attempts.
* <p/>
@@ -1914,24 +1934,6 @@ public abstract class RedeliveryErrorHandler extends
ErrorHandlerSupport
return same;
}
- /**
- * Prepares the redelivery counter and boolean flag for the failure handle
processor
- */
- private void decrementRedeliveryCounter(Exchange exchange) {
- Message in = exchange.getIn();
- Integer counter = in.getHeader(Exchange.REDELIVERY_COUNTER,
Integer.class);
- if (counter != null) {
- int prev = counter - 1;
- in.setHeader(Exchange.REDELIVERY_COUNTER, prev);
- // set boolean flag according to counter
- in.setHeader(Exchange.REDELIVERED, prev > 0 ? Boolean.TRUE :
Boolean.FALSE);
- } else {
- // not redelivered
- in.setHeader(Exchange.REDELIVERY_COUNTER, 0);
- in.setHeader(Exchange.REDELIVERED, Boolean.FALSE);
- }
- }
-
@Override
public boolean determineIfRedeliveryIsEnabled() throws Exception {
// determine if redeliver is enabled either on error handler
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/saga/SagaProcessor.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/saga/SagaProcessor.java
index b76050b47b5a..1e28a1f3a4fb 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/saga/SagaProcessor.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/saga/SagaProcessor.java
@@ -52,7 +52,12 @@ public abstract class SagaProcessor extends
BaseDelegateProcessorSupport impleme
}
protected CompletableFuture<CamelSagaCoordinator>
getCurrentSagaCoordinator(Exchange exchange) {
- String currentSaga =
exchange.getIn().getHeader(Exchange.SAGA_LONG_RUNNING_ACTION, String.class);
+ // try internal state first (survives removeHeaders("*"))
+ String currentSaga =
exchange.getExchangeExtension().getSagaLongRunningAction();
+ if (currentSaga == null) {
+ // fall back to header for interoperability (e.g., LRA protocol)
+ currentSaga =
exchange.getIn().getHeader(Exchange.SAGA_LONG_RUNNING_ACTION, String.class);
+ }
if (currentSaga != null) {
return sagaService.getSaga(currentSaga);
}
@@ -62,10 +67,13 @@ public abstract class SagaProcessor extends
BaseDelegateProcessorSupport impleme
protected void setCurrentSagaCoordinator(Exchange exchange,
CamelSagaCoordinator coordinator) {
if (coordinator != null) {
- exchange.getIn().setHeader(Exchange.SAGA_LONG_RUNNING_ACTION,
coordinator.getId());
+ String id = coordinator.getId();
+ exchange.getIn().setHeader(Exchange.SAGA_LONG_RUNNING_ACTION, id);
+ exchange.getExchangeExtension().setSagaLongRunningAction(id);
} else {
exchange.getIn().removeHeader(Exchange.SAGA_LONG_RUNNING_ACTION);
exchange.getMessage().removeHeader(Exchange.SAGA_LONG_RUNNING_ACTION);
+ exchange.getExchangeExtension().setSagaLongRunningAction(null);
}
}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/RedeliverToSubRouteRemoveHeadersDecrementTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/RedeliverToSubRouteRemoveHeadersDecrementTest.java
new file mode 100644
index 000000000000..d9297ea2aea2
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/RedeliverToSubRouteRemoveHeadersDecrementTest.java
@@ -0,0 +1,106 @@
+/*
+ * 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.processor;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.NotifyBuilder;
+import org.apache.camel.builder.RouteBuilder;
+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.assertTrue;
+
+/**
+ * Tests that decrementRedeliveryCounter correctly uses the internal field
when the failure handler receives the
+ * exchange after removeHeaders("*") was called in a child route.
+ */
+public class RedeliverToSubRouteRemoveHeadersDecrementTest extends
ContextTestSupport {
+
+ private static final AtomicInteger failureCounter = new AtomicInteger(-1);
+
+ @Override
+ @BeforeEach
+ public void setUp() throws Exception {
+ super.setUp();
+ failureCounter.set(-1);
+ }
+
+ @Test
+ public void testDecrementCounterWithRemoveHeaders() throws Exception {
+ getMockEndpoint("mock:failure").expectedMessageCount(1);
+
+ NotifyBuilder notify = new NotifyBuilder(context).whenDone(1).create();
+
+ try {
+ template.sendBody("direct:start", "Hello World");
+ } catch (Exception e) {
+ // expected: handled(false) propagates the exception
+ }
+
+ assertTrue(notify.matches(10, TimeUnit.SECONDS), "Exchange did not
complete in time");
+
+ assertMockEndpointsSatisfied();
+
+ // With maximumRedeliveries(3), redeliveryCounter reaches 4 (initial +
3 redeliveries),
+ // then decrementRedeliveryCounter brings it to 3. Without the fix,
removeHeaders("*")
+ // in the child route would cause the header to be null and the
counter would be set to 0.
+ assertEquals(3, failureCounter.get());
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ onException(IOException.class)
+ .maximumRedeliveries(3)
+ .redeliveryDelay(0)
+ .handled(false)
+ .process(new Processor() {
+ @Override
+ public void process(Exchange exchange) {
+ Integer counter =
exchange.getIn().getHeader(Exchange.REDELIVERY_COUNTER, Integer.class);
+ if (counter != null) {
+ failureCounter.set(counter);
+ }
+ }
+ })
+ .to("mock:failure");
+
+ from("direct:start")
+ .to("direct:sub");
+
+ from("direct:sub")
+ .errorHandler(noErrorHandler())
+ .removeHeaders("*")
+ .process(new Processor() {
+ @Override
+ public void process(Exchange exchange) throws
Exception {
+ throw new IOException("Forced");
+ }
+ });
+ }
+ };
+ }
+}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/RedeliverToSubRouteRemoveHeadersTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/RedeliverToSubRouteRemoveHeadersTest.java
new file mode 100644
index 000000000000..d989445afb8b
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/RedeliverToSubRouteRemoveHeadersTest.java
@@ -0,0 +1,121 @@
+/*
+ * 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.processor;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.NotifyBuilder;
+import org.apache.camel.builder.RouteBuilder;
+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.assertTrue;
+
+/**
+ * Tests that redelivery works correctly when a child route with
NoErrorHandler uses removeHeaders("*"), which strips
+ * internal CamelRedeliveryCounter headers during each redelivery attempt.
+ */
+public class RedeliverToSubRouteRemoveHeadersTest extends ContextTestSupport {
+
+ @Override
+ @BeforeEach
+ public void setUp() throws Exception {
+ super.setUp();
+ OnRedeliveryRecorder.counters.clear();
+ }
+
+ @Test
+ public void testRedeliveryCounterWithRemoveHeaders() throws Exception {
+ // The sub-route always throws, so after 3 redelivery attempts the
onException handler should kick in.
+ // mock:dead receives the handled failure; mock:sub is hit 1 (initial)
+ 3 (redeliveries) = 4 times.
+ getMockEndpoint("mock:dead").expectedMessageCount(1);
+ getMockEndpoint("mock:sub").expectedMessageCount(4);
+
+ NotifyBuilder notify = new NotifyBuilder(context).whenDone(1).create();
+
+ template.sendBody("direct:start", "Hello World");
+
+ assertTrue(notify.matches(10, TimeUnit.SECONDS), "Exchange did not
complete in time — redelivery may be looping");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Test
+ public void testOnRedeliverySeesCorrectCounter() throws Exception {
+ // Verify that the onRedelivery processor sees the correct,
incrementing
+ // CamelRedeliveryCounter header even when removeHeaders("*") is used
in the child route.
+ getMockEndpoint("mock:dead").expectedMessageCount(1);
+
+ NotifyBuilder notify = new NotifyBuilder(context).whenDone(1).create();
+
+ template.sendBody("direct:start", "Hello World");
+
+ assertTrue(notify.matches(10, TimeUnit.SECONDS), "Exchange did not
complete in time — redelivery may be looping");
+
+ assertMockEndpointsSatisfied();
+
+ assertEquals(List.of(1, 2, 3), OnRedeliveryRecorder.counters);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ onException(IOException.class)
+ .maximumRedeliveries(3)
+ .redeliveryDelay(0)
+ .handled(true)
+ .onRedelivery(new OnRedeliveryRecorder())
+ .to("mock:dead");
+
+ from("direct:start")
+ .to("direct:sub");
+
+ from("direct:sub")
+ .errorHandler(noErrorHandler())
+ .removeHeaders("*")
+ .to("mock:sub")
+ .process(new AlwaysFailProcessor());
+ }
+ };
+ }
+
+ public static class AlwaysFailProcessor implements Processor {
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ throw new IOException("Forced");
+ }
+ }
+
+ public static class OnRedeliveryRecorder implements Processor {
+ static final List<Integer> counters = new ArrayList<>();
+
+ @Override
+ public void process(Exchange exchange) {
+ Integer counter =
exchange.getIn().getHeader(Exchange.REDELIVERY_COUNTER, Integer.class);
+ counters.add(counter);
+ }
+ }
+}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/SagaRemoveHeadersTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/SagaRemoveHeadersTest.java
new file mode 100644
index 000000000000..c690d8bc8540
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/SagaRemoveHeadersTest.java
@@ -0,0 +1,90 @@
+/*
+ * 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.processor;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.saga.InMemorySagaService;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+
+/**
+ * Tests that saga coordination survives removeHeaders("*") in a sub-route.
The coordinator ID is stored internally in
+ * ExchangeExtension, so it is resilient to header removal.
+ */
+public class SagaRemoveHeadersTest extends ContextTestSupport {
+
+ @Test
+ public void testSagaRequiredSurvivesRemoveHeaders() throws Exception {
+ getMockEndpoint("mock:saga-step2").expectedMinimumMessageCount(1);
+ getMockEndpoint("mock:saga-step2")
+ .expectedMessagesMatches(ex ->
ex.getIn().getHeader(Exchange.SAGA_LONG_RUNNING_ACTION) != null);
+
+ template.sendBody("direct:saga-start", "Hello Saga");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Test
+ public void testSagaCompensationAfterRemoveHeaders() throws Exception {
+ getMockEndpoint("mock:compensate").expectedMinimumMessageCount(1);
+ getMockEndpoint("mock:compensate")
+ .expectedMessagesMatches(ex ->
ex.getIn().getHeader(Exchange.SAGA_LONG_RUNNING_ACTION) != null);
+
+ try {
+ template.sendBody("direct:saga-fail", "Hello Saga Fail");
+ } catch (Exception e) {
+ // expected
+ }
+
+ await().atMost(5, TimeUnit.SECONDS)
+ .untilAsserted(() -> assertMockEndpointsSatisfied());
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ InMemorySagaService sagaService = new InMemorySagaService();
+ context.addService(sagaService);
+
+ from("direct:saga-start")
+ .saga()
+ .removeHeaders("*")
+ .to("direct:saga-step2");
+
+ from("direct:saga-step2")
+ .saga()
+ .to("mock:saga-step2");
+
+ from("direct:saga-fail")
+ .saga()
+ .compensation("direct:compensate")
+ .removeHeaders("*")
+ .throwException(new RuntimeException("Saga failure"));
+
+ from("direct:compensate")
+ .to("mock:compensate");
+ }
+ };
+ }
+}
diff --git
a/core/camel-support/src/main/java/org/apache/camel/saga/InMemorySagaCoordinator.java
b/core/camel-support/src/main/java/org/apache/camel/saga/InMemorySagaCoordinator.java
index 067744af5369..a049aeca26a5 100644
---
a/core/camel-support/src/main/java/org/apache/camel/saga/InMemorySagaCoordinator.java
+++
b/core/camel-support/src/main/java/org/apache/camel/saga/InMemorySagaCoordinator.java
@@ -231,6 +231,7 @@ public class InMemorySagaCoordinator implements
CamelSagaCoordinator {
private Exchange createExchange(Exchange parent, Endpoint endpoint,
CamelSagaStep step) {
Exchange answer = endpoint.createExchange();
answer.getMessage().setHeader(Exchange.SAGA_LONG_RUNNING_ACTION,
getId());
+ answer.getExchangeExtension().setSagaLongRunningAction(getId());
// preserve span from parent, so we can link this new exchange to the
parent span for distributed tracing
Object span = parent != null ?
parent.getProperty(ExchangePropertyKey.OTEL_ACTIVE_SPAN) : null;
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/ExchangeHelper.java
b/core/camel-support/src/main/java/org/apache/camel/support/ExchangeHelper.java
index 81c0f466b153..1e0b486d6dc0 100644
---
a/core/camel-support/src/main/java/org/apache/camel/support/ExchangeHelper.java
+++
b/core/camel-support/src/main/java/org/apache/camel/support/ExchangeHelper.java
@@ -391,6 +391,9 @@ public final class ExchangeHelper {
result.setRollbackOnlyLast(source.isRollbackOnlyLast());
resultExtension.setNotifyEvent(sourceExtension.isNotifyEvent());
resultExtension.setRedeliveryExhausted(sourceExtension.isRedeliveryExhausted());
+
resultExtension.setRedeliveryCounter(sourceExtension.getRedeliveryCounter());
+
resultExtension.setRedeliveryMaxCounter(sourceExtension.getRedeliveryMaxCounter());
+
resultExtension.setSagaLongRunningAction(sourceExtension.getSagaLongRunningAction());
resultExtension.setErrorHandlerHandled(sourceExtension.getErrorHandlerHandled());
resultExtension.setFailureHandled(sourceExtension.isFailureHandled());
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/ExtendedExchangeExtension.java
b/core/camel-support/src/main/java/org/apache/camel/support/ExtendedExchangeExtension.java
index 202c8e63e7ff..16dad05f146b 100644
---
a/core/camel-support/src/main/java/org/apache/camel/support/ExtendedExchangeExtension.java
+++
b/core/camel-support/src/main/java/org/apache/camel/support/ExtendedExchangeExtension.java
@@ -41,6 +41,9 @@ public class ExtendedExchangeExtension implements
ExchangeExtension {
private String fromRouteId;
private boolean streamCacheDisabled;
private boolean redeliveryExhausted;
+ private int redeliveryCounter = -1;
+ private int redeliveryMaxCounter = -1;
+ private String sagaLongRunningAction;
private String historyNodeId;
private String historyNodeSource;
private String historyNodeLabel;
@@ -139,6 +142,36 @@ public class ExtendedExchangeExtension implements
ExchangeExtension {
this.redeliveryExhausted = redeliveryExhausted;
}
+ @Override
+ public int getRedeliveryCounter() {
+ return this.redeliveryCounter;
+ }
+
+ @Override
+ public void setRedeliveryCounter(int redeliveryCounter) {
+ this.redeliveryCounter = redeliveryCounter;
+ }
+
+ @Override
+ public int getRedeliveryMaxCounter() {
+ return this.redeliveryMaxCounter;
+ }
+
+ @Override
+ public void setRedeliveryMaxCounter(int redeliveryMaxCounter) {
+ this.redeliveryMaxCounter = redeliveryMaxCounter;
+ }
+
+ @Override
+ public String getSagaLongRunningAction() {
+ return this.sagaLongRunningAction;
+ }
+
+ @Override
+ public void setSagaLongRunningAction(String sagaLongRunningAction) {
+ this.sagaLongRunningAction = sagaLongRunningAction;
+ }
+
@Override
public void handoverCompletions(Exchange target) {
if (onCompletions != null) {
@@ -377,6 +410,9 @@ public class ExtendedExchangeExtension implements
ExchangeExtension {
setInterrupted(false);
setInterruptable(true);
setRedeliveryExhausted(false);
+ setRedeliveryCounter(-1);
+ setRedeliveryMaxCounter(-1);
+ setSagaLongRunningAction(null);
setErrorHandlerHandled(null);
setStreamCacheDisabled(false);
setRollbackOnly(false);
