This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26607-multi-hook in repository https://gitbox.apache.org/repos/asf/camel.git
commit fb01bf65336cd19cc32dee41eb72d5accb842767 Author: Claus Ibsen <[email protected]> AuthorDate: Sat Nov 7 11:17:27 2026 +0100 CAMEL-26607: camel-chaos - a duplicate fault that sends a message twice, to test idempotent consumers A new SendHook.SEND_TWICE (-2): SendHelper copies the message before the first send (a stream body is cached first), and when the first send succeeded it sends the copy again, on the thread that completed the first send. A failed first send is not repeated, the second reply wins, afterSend and the sending/sent events run once. With several hooks, any SEND_TWICE sends twice, and a SKIP still wins. camel-chaos gets the duplicate fault (endpoint=kafka:orders* duplicate rate=0.05) at the send hook, in the parser, the model (with the recipe "Is delivered twice") and the docs. Co-Authored-By: Claude Opus 5.5 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../org/apache/camel/catalog/chaos-model.json | 70 +++++- .../org/apache/camel/catalog/docs/chaos.adoc | 14 +- .../org/apache/camel/chaos/chaos-model.json | 70 +++++- components/camel-chaos/src/main/docs/chaos.adoc | 14 +- .../component/chaos/ChaosContextServicePlugin.java | 3 +- .../apache/camel/component/chaos/ChaosFaults.java | 5 + .../apache/camel/component/chaos/ChaosModel.java | 21 ++ .../apache/camel/component/chaos/ChaosRule.java | 3 + .../camel/component/chaos/ChaosRuleParser.java | 13 +- .../camel/component/chaos/ChaosSendHook.java | 10 +- .../camel/component/chaos/ChaosDuplicateTest.java | 139 +++++++++++ .../camel/component/chaos/ChaosModelTest.java | 7 +- .../camel/component/chaos/ChaosRuleParserTest.java | 20 +- .../main/java/org/apache/camel/spi/SendHook.java | 36 ++- .../org/apache/camel/processor/SendTwiceTest.java | 264 +++++++++++++++++++++ .../apache/camel/support/CompositeHookTest.java | 35 +++ .../org/apache/camel/support/SendHelperTest.java | 15 +- .../apache/camel/support/CompositeSendHook.java | 17 +- .../java/org/apache/camel/support/SendHelper.java | 115 ++++++++- design/chaos.adoc | 9 +- .../modules/ROOT/pages/security-model.adoc | 14 +- 21 files changed, 839 insertions(+), 55 deletions(-) diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/chaos-model.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/chaos-model.json index 75d82165406c..34095d99a9b8 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/chaos-model.json +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/chaos-model.json @@ -40,7 +40,8 @@ "error", "delay", "drop", - "corrupt" + "corrupt", + "duplicate" ], "onlyWithAt": false }, @@ -295,6 +296,28 @@ "within" ] }, + { + "name": "duplicate", + "description": "Sends the message twice, as a sender that delivers a message again; the reply of the second send is the reply of the send", + "targets": [ + "component", + "endpoint" + ], + "hooks": [ + "send" + ], + "parameters": [ + + ], + "options": [ + "at", + "expect", + "from", + "rate", + "until", + "within" + ] + }, { "name": "lifecycle", "description": "Suspends, stops or removes the route of a route=<id> target, and brings it back after the duration; repeat= takes it down more than once", @@ -936,6 +959,51 @@ } ] }, + { + "id": "duplicate", + "group": "Messages", + "title": "Is delivered twice", + "description": "The message is sent to the endpoint twice, as a sender or a broker that delivers a message again. The receiver must not do the work twice", + "uncovers": [ + "Duplicates: an order booked twice, a payment taken twice", + "A receiving route without an idempotent consumer, or with a key that is not the same for both copies", + "An aggregator or a counter that counts the message twice" + ], + "avoid": [ + "Against a real service where a duplicate costs, such as one that charges or sends mail to people", + "To test that a failed send is sent again: use Did the work, then failed" + ], + "watch": [ + "What the other side received: two of the same", + "The messages the idempotent consumer of the receiver dropped as duplicates" + ], + "rule": "endpoint=${endpoint} duplicate rate=${rate} until=${until}", + "placeholders": [ + { + "name": "endpoint", + "type": "endpoint", + "description": "The endpoint the route sends to", + "example": "kafka:orders", + "pick": { + "from": "producers" + } + }, + { + "name": "rate", + "type": "rate", + "description": "How often: a probability, every:N or burst:<time>\/<period>", + "default": "0.1", + "example": "0.1" + }, + { + "name": "until", + "type": "countOrDuration", + "description": "When the rule ends: a number of matches or a time", + "default": "10m", + "example": "10m" + } + ] + }, { "id": "poll-error", "group": "Consumers and routes", diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/chaos.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/chaos.adoc index 54671d202bf6..04e85dae28e1 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/chaos.adoc +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/chaos.adoc @@ -172,7 +172,7 @@ are a quote and a backslash). [source] ---- target: route=<id> | endpoint=<uri> | component=<name> | step=<id> (* matches any characters) -fault: error | delay | drop | corrupt | stall | lifecycle, with its parameters +fault: error | delay | drop | corrupt | duplicate | stall | lifecycle, with its parameters at: send | reply | dlq | poll | entry | step (optional) rate: a probability (rate=0.3), every Nth match (rate=every:5), or bursts (rate=burst:5s/30s: every match during the first 5s of each 30s); without it every match fires @@ -228,7 +228,8 @@ target offers, is rejected when it is added, with an error that names the hooks |The sends to a matching endpoint (`send`), the reply of such a send after the producer succeeded (`reply`, only with `at=reply`), the send of the dead letter channel to it (`dlq`, only with `at=dlq`), and the polls of its consumer (`poll`). -|`send` and `dlq`: `error`, `delay`, `drop`, `corrupt`. `reply`: `error`, `corrupt`. `poll`: `error`, `delay`, `drop`. +|`send`: `error`, `delay`, `drop`, `corrupt`, `duplicate`. `dlq`: `error`, `delay`, `drop`, `corrupt`. `reply`: +`error`, `corrupt`. `poll`: `error`, `delay`, `drop`. |`route=` |The messages that enter the route (`entry`), and the route itself. @@ -300,6 +301,15 @@ delays again. |none |A send that reports success without calling the producer, or a poll that is skipped. +|`duplicate` +|none +|Sends the message twice, as a sender or a broker that delivers a message again (`endpoint=kafka:orders* duplicate +rate=0.05`), to test that the receiver is idempotent, such as a route with an `idempotentConsumer`. The message is +copied before the first send (a stream body is cached first, so both sends read all of it), and when the first send +succeeded the copy is sent again; a first send that failed is not sent again. With a reply (InOut), the reply of the +second send is the one the route goes on with. The sending and sent events fire once, and the report counts one +fault. A fault of the sends only (`at=send`): not at the reply, the polls or the dead letter channel. + |`corrupt` |`mode=headers\|charset\|truncate\|empty\|remove-header`, and `header=<name>` for `remove-header` |Changes the message: at a send, the exchange that is sent, so the endpoint gets the changed message, and after a `to` diff --git a/components/camel-chaos/src/generated/resources/META-INF/org/apache/camel/chaos/chaos-model.json b/components/camel-chaos/src/generated/resources/META-INF/org/apache/camel/chaos/chaos-model.json index 75d82165406c..34095d99a9b8 100644 --- a/components/camel-chaos/src/generated/resources/META-INF/org/apache/camel/chaos/chaos-model.json +++ b/components/camel-chaos/src/generated/resources/META-INF/org/apache/camel/chaos/chaos-model.json @@ -40,7 +40,8 @@ "error", "delay", "drop", - "corrupt" + "corrupt", + "duplicate" ], "onlyWithAt": false }, @@ -295,6 +296,28 @@ "within" ] }, + { + "name": "duplicate", + "description": "Sends the message twice, as a sender that delivers a message again; the reply of the second send is the reply of the send", + "targets": [ + "component", + "endpoint" + ], + "hooks": [ + "send" + ], + "parameters": [ + + ], + "options": [ + "at", + "expect", + "from", + "rate", + "until", + "within" + ] + }, { "name": "lifecycle", "description": "Suspends, stops or removes the route of a route=<id> target, and brings it back after the duration; repeat= takes it down more than once", @@ -936,6 +959,51 @@ } ] }, + { + "id": "duplicate", + "group": "Messages", + "title": "Is delivered twice", + "description": "The message is sent to the endpoint twice, as a sender or a broker that delivers a message again. The receiver must not do the work twice", + "uncovers": [ + "Duplicates: an order booked twice, a payment taken twice", + "A receiving route without an idempotent consumer, or with a key that is not the same for both copies", + "An aggregator or a counter that counts the message twice" + ], + "avoid": [ + "Against a real service where a duplicate costs, such as one that charges or sends mail to people", + "To test that a failed send is sent again: use Did the work, then failed" + ], + "watch": [ + "What the other side received: two of the same", + "The messages the idempotent consumer of the receiver dropped as duplicates" + ], + "rule": "endpoint=${endpoint} duplicate rate=${rate} until=${until}", + "placeholders": [ + { + "name": "endpoint", + "type": "endpoint", + "description": "The endpoint the route sends to", + "example": "kafka:orders", + "pick": { + "from": "producers" + } + }, + { + "name": "rate", + "type": "rate", + "description": "How often: a probability, every:N or burst:<time>\/<period>", + "default": "0.1", + "example": "0.1" + }, + { + "name": "until", + "type": "countOrDuration", + "description": "When the rule ends: a number of matches or a time", + "default": "10m", + "example": "10m" + } + ] + }, { "id": "poll-error", "group": "Consumers and routes", diff --git a/components/camel-chaos/src/main/docs/chaos.adoc b/components/camel-chaos/src/main/docs/chaos.adoc index 54671d202bf6..04e85dae28e1 100644 --- a/components/camel-chaos/src/main/docs/chaos.adoc +++ b/components/camel-chaos/src/main/docs/chaos.adoc @@ -172,7 +172,7 @@ are a quote and a backslash). [source] ---- target: route=<id> | endpoint=<uri> | component=<name> | step=<id> (* matches any characters) -fault: error | delay | drop | corrupt | stall | lifecycle, with its parameters +fault: error | delay | drop | corrupt | duplicate | stall | lifecycle, with its parameters at: send | reply | dlq | poll | entry | step (optional) rate: a probability (rate=0.3), every Nth match (rate=every:5), or bursts (rate=burst:5s/30s: every match during the first 5s of each 30s); without it every match fires @@ -228,7 +228,8 @@ target offers, is rejected when it is added, with an error that names the hooks |The sends to a matching endpoint (`send`), the reply of such a send after the producer succeeded (`reply`, only with `at=reply`), the send of the dead letter channel to it (`dlq`, only with `at=dlq`), and the polls of its consumer (`poll`). -|`send` and `dlq`: `error`, `delay`, `drop`, `corrupt`. `reply`: `error`, `corrupt`. `poll`: `error`, `delay`, `drop`. +|`send`: `error`, `delay`, `drop`, `corrupt`, `duplicate`. `dlq`: `error`, `delay`, `drop`, `corrupt`. `reply`: +`error`, `corrupt`. `poll`: `error`, `delay`, `drop`. |`route=` |The messages that enter the route (`entry`), and the route itself. @@ -300,6 +301,15 @@ delays again. |none |A send that reports success without calling the producer, or a poll that is skipped. +|`duplicate` +|none +|Sends the message twice, as a sender or a broker that delivers a message again (`endpoint=kafka:orders* duplicate +rate=0.05`), to test that the receiver is idempotent, such as a route with an `idempotentConsumer`. The message is +copied before the first send (a stream body is cached first, so both sends read all of it), and when the first send +succeeded the copy is sent again; a first send that failed is not sent again. With a reply (InOut), the reply of the +second send is the one the route goes on with. The sending and sent events fire once, and the report counts one +fault. A fault of the sends only (`at=send`): not at the reply, the polls or the dead letter channel. + |`corrupt` |`mode=headers\|charset\|truncate\|empty\|remove-header`, and `header=<name>` for `remove-header` |Changes the message: at a send, the exchange that is sent, so the endpoint gets the changed message, and after a `to` diff --git a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosContextServicePlugin.java b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosContextServicePlugin.java index 00dca6303eb9..c08cb9938002 100644 --- a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosContextServicePlugin.java +++ b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosContextServicePlugin.java @@ -129,7 +129,8 @@ public class ChaosContextServicePlugin implements ContextServicePlugin { chaos.install(); context.getCamelContextExtension().addContextPlugin(ChaosManager.class, chaos); manager = chaos; - LOG.warn("Chaos is switched on ({}): rules can fail, delay, drop and corrupt the messages of this application." + LOG.warn( + "Chaos is switched on ({}): rules can fail, delay, drop, duplicate and corrupt the messages of this application." + " Do not use in production. Seed: {}, rules from {}: {}", enabled != null ? ENABLED + "=true" : "profile dev", chaos.getSeed(), RULES, startup.size()); } diff --git a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosFaults.java b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosFaults.java index c3d7af72a0eb..4ca68f6f7091 100644 --- a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosFaults.java +++ b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosFaults.java @@ -111,6 +111,11 @@ final class ChaosFaults { corrupt(rule, exchange, hook, null, endpoint.getEndpointUri()); return SendHook.PROCEED; } + case DUPLICATE -> { + // the message is sent twice, as a sender that delivers it again + rule.fired(hook, exchange, null, endpoint.getEndpointUri()); + return SendHook.SEND_TWICE; + } default -> throw new IllegalStateException("Unknown fault " + rule.fault); } } diff --git a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosModel.java b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosModel.java index 0da00fcf3912..958a65f6dabe 100644 --- a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosModel.java +++ b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosModel.java @@ -165,6 +165,8 @@ final class ChaosModel { case DELAY -> "Waits before the send, the poll or the step, as for a slow endpoint"; case DROP -> "A send that reports success without calling the producer, or a poll that is skipped"; case CORRUPT -> "Changes the message: its headers, the charset of its body, or its body"; + case DUPLICATE -> "Sends the message twice, as a sender that delivers a message again; the reply" + + " of the second send is the reply of the send"; default -> throw new IllegalStateException("Unknown fault " + fault); })); } @@ -439,6 +441,12 @@ final class ChaosModel { List.of(producer(), new Placeholder( "header", "string", "The header, or a pattern with *", "X-Trace-Id", null, pick("headers", null, null)), rate("0.25"), until("10m")))); + answer.add(new Recipe( + "duplicate", "Messages", "Is delivered twice", + "The message is sent to the endpoint twice, as a sender or a broker that delivers " + + "a message again. The receiver must not do the work twice", + "endpoint=${endpoint} duplicate rate=${rate} until=${until}", + List.of(producer(), rate("0.1"), until("10m")))); // consumers and routes answer.add(new Recipe( "poll-error", "Consumers and routes", "Polls fail", @@ -610,6 +618,19 @@ final class ChaosModel { + "changes"), List.of( "The route or the service that receives it"))); + answer.put("duplicate", new Guide( + List.of( + "Duplicates: an order booked twice, a payment taken twice", + "A receiving route without an idempotent consumer, or with a key that is not " + + "the same for both copies", + "An aggregator or a counter that counts the message twice"), + List.of( + "Against a real service where a duplicate costs, such as one that charges or " + + "sends mail to people", + "To test that a failed send is sent again: use Did the work, then failed"), + List.of( + "What the other side received: two of the same", + "The messages the idempotent consumer of the receiver dropped as duplicates"))); answer.put("poll-error", new Guide( List.of( "A consumer without a backoff, which fails every poll and fills the log", diff --git a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRule.java b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRule.java index 3eaa586e74e7..cbd56bb4cd64 100644 --- a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRule.java +++ b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRule.java @@ -72,6 +72,8 @@ final class ChaosRule { DELAY("delay"), DROP("drop"), CORRUPT("corrupt"), + // a send that happens twice + DUPLICATE("duplicate"), // stall and lifecycle: they act once on the route, at no hook LIFECYCLE("lifecycle"); @@ -426,6 +428,7 @@ final class ChaosRule { + (durationMax > duration ? ".." + ChaosRuleParser.printTime(durationMax) : ""); case DROP -> "drop"; case CORRUPT -> "corrupt " + mode.key + (header != null ? " " + header : ""); + case DUPLICATE -> "duplicate"; case LIFECYCLE -> (stall ? "stall" : action.key) + " " + ChaosRuleParser.printTime(duration); }; } diff --git a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRuleParser.java b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRuleParser.java index 32165cb3a225..16ab1146432e 100644 --- a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRuleParser.java +++ b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosRuleParser.java @@ -88,6 +88,7 @@ final class ChaosRuleParser { case DELAY -> Set.of("duration"); case DROP -> Set.of(); case CORRUPT -> Set.of("mode", "header"); + case DUPLICATE -> Set.of(); case LIFECYCLE -> stall ? Set.of("duration", "repeat", "every") : Set.of("action", "duration", "repeat", "every"); }; } @@ -219,7 +220,8 @@ final class ChaosRuleParser { + (unknown != null ? " (" + unknown + "= is not a target)" : "")); } if (b.fault == null) { - throw new IllegalArgumentException("a rule needs a fault: error, delay, drop, corrupt, stall or lifecycle"); + throw new IllegalArgumentException( + "a rule needs a fault: error, delay, drop, corrupt, duplicate, stall or lifecycle"); } if (b.target == Target.COMPONENT && b.pattern.indexOf(':') >= 0) { // matched against the scheme only, so a pattern with a colon would never match @@ -278,6 +280,7 @@ final class ChaosRuleParser { } } case DROP -> checkKeys(params, faultKeys(Fault.DROP, false), b.fault); + case DUPLICATE -> checkKeys(params, faultKeys(Fault.DUPLICATE, false), b.fault); case CORRUPT -> { checkKeys(params, faultKeys(Fault.CORRUPT, false), b.fault); String mode = params.get("mode"); @@ -436,10 +439,11 @@ final class ChaosRuleParser { case "delay" -> Fault.DELAY; case "drop" -> Fault.DROP; case "corrupt" -> Fault.CORRUPT; + case "duplicate" -> Fault.DUPLICATE; case "stall", "lifecycle" -> Fault.LIFECYCLE; default -> throw new IllegalArgumentException( "unknown fault " + word - + " (the faults are error, delay, drop, corrupt, stall and lifecycle;" + + " (the faults are error, delay, drop, corrupt, duplicate, stall and lifecycle;" + " a parameter is name=value)"); }; } @@ -776,6 +780,9 @@ final class ChaosRuleParser { if (target == Target.STEP && fault == Fault.DROP) { return "; drop is a fault of the sends and polls: target them with endpoint= or component="; } + if (target == Target.STEP && fault == Fault.DUPLICATE) { + return "; duplicate is a fault of the sends: target them with endpoint= or component="; + } if (target == Target.STEP && fault == Fault.CORRUPT) { return "; corrupt is a fault of the sends (endpoint=, component=) and of the entry of a route (route=)"; } @@ -800,7 +807,7 @@ final class ChaosRuleParser { Map<Hook, Set<Fault>> answer = new LinkedHashMap<>(); switch (target) { case ENDPOINT, COMPONENT -> { - answer.put(Hook.SEND, EnumSet.of(Fault.ERROR, Fault.DELAY, Fault.DROP, Fault.CORRUPT)); + answer.put(Hook.SEND, EnumSet.of(Fault.ERROR, Fault.DELAY, Fault.DROP, Fault.CORRUPT, Fault.DUPLICATE)); answer.put(Hook.REPLY, EnumSet.of(Fault.ERROR, Fault.CORRUPT)); answer.put(Hook.DLQ, EnumSet.of(Fault.ERROR, Fault.DELAY, Fault.DROP, Fault.CORRUPT)); answer.put(Hook.POLL, EnumSet.of(Fault.ERROR, Fault.DELAY, Fault.DROP)); diff --git a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosSendHook.java b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosSendHook.java index b013e4b29946..74e9c11abacf 100644 --- a/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosSendHook.java +++ b/components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosSendHook.java @@ -26,11 +26,11 @@ import org.apache.camel.util.TimeSource; /** * Raises the faults of the rules at the sends to an endpoint: {@code error} (the producer is not called), {@code delay} * (the sending thread waits, through {@link TimeSource}, before it calls the producer), {@code drop} (the send reports - * success without calling the producer) and {@code corrupt} (the message that is sent). The first rule that fires - * decides the fault of a send; the rules are asked in the order they were added. The send of the dead letter channel - * meets only the rules with {@code at=dlq}: the event notifier marks the exchange while the channel delivers it. The - * rules with {@code at=reply} are asked after the producer succeeded: {@code error} fails the send after its side - * effect happened, and {@code corrupt} changes the reply. + * success without calling the producer), {@code corrupt} (the message that is sent) and {@code duplicate} (the message + * is sent twice). The first rule that fires decides the fault of a send; the rules are asked in the order they were + * added. The send of the dead letter channel meets only the rules with {@code at=dlq}: the event notifier marks the + * exchange while the channel delivers it. The rules with {@code at=reply} are asked after the producer succeeded: + * {@code error} fails the send after its side effect happened, and {@code corrupt} changes the reply. */ final class ChaosSendHook implements SendHook, Ordered { diff --git a/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosDuplicateTest.java b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosDuplicateTest.java new file mode 100644 index 000000000000..76e20fdae680 --- /dev/null +++ b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosDuplicateTest.java @@ -0,0 +1,139 @@ +/* + * 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.chaos; + +import java.io.ByteArrayInputStream; +import java.nio.charset.StandardCharsets; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.camel.Exchange; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * The {@code duplicate} fault (CAMEL-26607): a send that happens twice, to test that the receiver is idempotent. + */ +class ChaosDuplicateTest extends ChaosTestSupport { + + private final AtomicInteger replies = new AtomicInteger(); + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("direct:start").routeId("start").to("mock:target").to("mock:after"); + + from("direct:inout").routeId("inout").to("direct:reply"); + from("direct:reply").routeId("reply") + .process(e -> e.getMessage().setBody("reply-" + replies.incrementAndGet())); + + from("direct:stream").routeId("stream").to("direct:read"); + from("direct:read").routeId("read").convertBodyTo(String.class).to("mock:read"); + + from("direct:orders").routeId("orders").to("direct:booking"); + from("direct:booking").routeId("booking") + .idempotentConsumer(header("orderId"), MemoryIdempotentRepository.memoryIdempotentRepository()) + .to("mock:booked"); + } + }; + } + + @Test + void inOnlySendsTheMessageTwice() throws Exception { + String id = chaos().addRule("endpoint=mock:target duplicate"); + MockEndpoint target = getMockEndpoint("mock:target"); + target.expectedBodiesReceived("hello", "hello"); + getMockEndpoint("mock:after").expectedBodiesReceived("hello"); + + Exchange out = template.send("direct:start", e -> e.getMessage().setBody("hello")); + + assertThat(out.getException()).isNull(); + MockEndpoint.assertIsSatisfied(context); + // one fault for the two sends + assertThat(fired(id, "send")).isEqualTo(1); + } + + @Test + void inOutTheSecondReplyWins() { + String id = chaos().addRule("endpoint=direct:reply duplicate"); + + Object reply = template.requestBody("direct:inout", "hello"); + + assertThat(replies.get()).isEqualTo(2); + assertThat(reply).isEqualTo("reply-2"); + assertThat(fired(id, "send")).isEqualTo(1); + } + + @Test + void aFailedFirstSendIsNotRepeated() throws Exception { + chaos().addRule("endpoint=mock:target duplicate"); + MockEndpoint target = getMockEndpoint("mock:target"); + target.whenAnyExchangeReceived(e -> { + throw new IllegalStateException("down"); + }); + target.expectedMessageCount(1); + getMockEndpoint("mock:after").expectedMessageCount(0); + + Exchange out = template.send("direct:start", e -> e.getMessage().setBody("hello")); + + assertThat(out.getException()).isInstanceOf(IllegalStateException.class).hasMessage("down"); + MockEndpoint.assertIsSatisfied(context); + } + + @Test + void aStreamBodyIsSentTwiceIntact() throws Exception { + chaos().addRule("endpoint=direct:read duplicate"); + getMockEndpoint("mock:read").expectedBodiesReceived("hello stream", "hello stream"); + + template.sendBody("direct:stream", new ByteArrayInputStream("hello stream".getBytes(StandardCharsets.UTF_8))); + + MockEndpoint.assertIsSatisfied(context); + } + + @Test + void anIdempotentConsumerDropsTheDuplicate() throws Exception { + String id = chaos().addRule("endpoint=direct:booking duplicate"); + getMockEndpoint("mock:booked").expectedBodiesReceived("order 1", "order 2"); + + template.sendBodyAndHeader("direct:orders", "order 1", "orderId", "1"); + template.sendBodyAndHeader("direct:orders", "order 2", "orderId", "2"); + + MockEndpoint.assertIsSatisfied(context); + assertThat(fired(id, "send")).isEqualTo(2); + // the two orders completed once each: the duplicates are dropped by the idempotent consumer + assertThat(routeCount("orders", "completed")).isEqualTo(2); + } + + @Test + void theRateDecidesWhichSendsAreDuplicated() throws Exception { + String id = chaos().addRule("endpoint=mock:target duplicate rate=every:2"); + getMockEndpoint("mock:target").expectedMessageCount(6); + + for (int i = 0; i < 4; i++) { + template.sendBody("direct:start", "hello " + i); + } + + MockEndpoint.assertIsSatisfied(context); + assertThat(fired(id, "send")).isEqualTo(2); + assertThat(chaos().report().toText()).contains("duplicate"); + } +} diff --git a/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosModelTest.java b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosModelTest.java index f14a97807bd3..0c2d80da559f 100644 --- a/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosModelTest.java +++ b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosModelTest.java @@ -85,7 +85,8 @@ class ChaosModelTest { assertThat(new HashSet<>((JsonArray) hook.get("faults"))).containsAll(faults); } } - assertThat(names("faults")).containsExactly("error", "delay", "drop", "corrupt", "lifecycle", "stall"); + assertThat(names("faults")).containsExactly("error", "delay", "drop", "corrupt", "duplicate", "lifecycle", + "stall"); JsonObject stall = byName("faults", "stall"); assertThat(stall.getString("shortFor")).isEqualTo("lifecycle action=suspend"); assertThat(paramNames(stall)).containsExactly("duration", "every", "repeat"); @@ -94,6 +95,10 @@ class ChaosModelTest { .containsExactly("after", "body", "exception", "header.<name>", "message", "status"); assertThat(strings(byName("faults", "lifecycle").get("options"))).containsExactly("expect", "from", "within"); assertThat(strings(byName("faults", "error").get("hooks"))).containsExactly("send", "reply", "dlq", "poll", "step"); + // duplicate is a fault of the sends only + assertThat(strings(byName("faults", "duplicate").get("hooks"))).containsExactly("send"); + assertThat(strings(byName("faults", "duplicate").get("targets"))).containsExactly("component", "endpoint"); + assertThat(paramNames(byName("faults", "duplicate"))).isEmpty(); assertThat(names("options")).containsExactly("at", "expect", "from", "rate", "until", "within"); } diff --git a/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosRuleParserTest.java b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosRuleParserTest.java index 85d0eab3a840..81f7efcbca7f 100644 --- a/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosRuleParserTest.java +++ b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/ChaosRuleParserTest.java @@ -120,6 +120,17 @@ class ChaosRuleParserTest { assertThat(parse("step=s1 error exception=" + IOException.class.getName()).exceptionType).isEqualTo(IOException.class); } + @Test + void duplicate() { + ChaosRule.Builder duplicate = parse("endpoint=kafka:orders* duplicate rate=0.05"); + assertThat(duplicate.fault).isEqualTo(Fault.DUPLICATE); + assertThat(duplicate.target).isEqualTo(Target.ENDPOINT); + assertThat(duplicate.pattern).isEqualTo("kafka:orders*"); + // a fault of the sends only: not at the polls, the reply or the dead letter channel + assertThat(duplicate.hooks).containsExactly(Hook.SEND); + assertThat(parse("component=kafka duplicate at=send until=10").hooks).containsExactly(Hook.SEND); + } + @Test void lifecycle() { ChaosRule.Builder stall = parse("route=poller stall duration=20s"); @@ -288,6 +299,12 @@ class ChaosRuleParserTest { "route=orders error|route= does not reach the sends and steps of the route: target them with endpoint=," + " component= or step=", "step=s1 drop|drop is a fault of the sends and polls: target them with endpoint= or component=", + "step=s1 duplicate|duplicate is a fault of the sends: target them with endpoint= or component=", + "route=orders duplicate|route= does not reach the sends and steps of the route", + "endpoint=x duplicate at=poll|the fault duplicate is not offered at=poll", + "endpoint=x duplicate at=reply|the fault duplicate is not offered at=reply", + "endpoint=x duplicate at=dlq|the fault duplicate is not offered at=dlq", + "endpoint=x duplicate rate=0.5 mode=headers|unknown parameter mode for the fault duplicate", "Endpoint=x error|(Endpoint= is not a target)", "endpoint=x|needs a fault", "endpoint=x error drop|has one fault", @@ -501,7 +518,8 @@ class ChaosRuleParserTest { // random rules built from the tokens and characters of the syntax: every input either parses or is rejected // with an IllegalArgumentException that has a reason, and nothing hangs String[] pieces = { - "route=", "endpoint=", "component=", "step=", "error", "delay", "drop", "corrupt", "stall", "at=", "send", + "route=", "endpoint=", "component=", "step=", "error", "delay", "drop", "corrupt", "duplicate", "stall", "at=", + "send", "poll", "entry", "step", "rate=", "every:", "until=", "expect=", "within=", "duration=", "mode=", "headers", "charset", "truncate", "exception=", "java.io.IOException", "message=", "status=", "503", "header.", "body=", "\"", "'", "\\", " ", " ", "\t", "=", ":", ";", "*", "0", "1", "0.5", "1m", "ms", diff --git a/core/camel-api/src/main/java/org/apache/camel/spi/SendHook.java b/core/camel-api/src/main/java/org/apache/camel/spi/SendHook.java index 59e3e97ba7f9..9ec271307b81 100644 --- a/core/camel-api/src/main/java/org/apache/camel/spi/SendHook.java +++ b/core/camel-api/src/main/java/org/apache/camel/spi/SendHook.java @@ -36,12 +36,13 @@ import org.apache.camel.Exchange; * {@link org.apache.camel.Ordered} has order 0, and hooks with the same order are asked in the order they were added). * A hook that stops the send (it throws an exception or sets one on the exchange), skips it ({@link #SKIP}, so the * first {@link #SKIP} wins) or returns a reserved value ends it: the hooks after it are not asked, and the send does - * not wait for a delay. Otherwise the longest of the delays wins, and the send waits for it once. - * {@link #afterSend(Exchange, Endpoint)} is called only on the hooks whose {@link #beforeSend(Exchange, Endpoint)} - * returned, in the reverse order: the hook asked first before the send is called last after it, so the hooks nest. A - * hook that only observes the sends takes a low order, so it is outside the hooks that change the outcome and sees the - * send as the route sees it; a hook that changes the outcome of a send (such as the faults of camel-chaos, with - * {@link org.apache.camel.Ordered#LOWEST}) takes a high order, to run innermost, right next to the producer. + * not wait for a delay. Otherwise the longest of the delays wins, and the send waits for it once; when a hook returned + * {@link #SEND_TWICE}, the message is then sent twice. {@link #afterSend(Exchange, Endpoint)} is called only on the + * hooks whose {@link #beforeSend(Exchange, Endpoint)} returned, in the reverse order: the hook asked first before the + * send is called last after it, so the hooks nest. A hook that only observes the sends takes a low order, so it is + * outside the hooks that change the outcome and sees the send as the route sees it; a hook that changes the outcome of + * a send (such as the faults of camel-chaos, with {@link org.apache.camel.Ordered#LOWEST}) takes a high order, to run + * innermost, right next to the producer. * <p/> * <b>Where it runs.</b> The hook runs right around the call of the producer, inside the error handler that handles a * failure of that producer: the error handler of the route for {@code to}, {@code toD} and {@code enrich}, the error @@ -84,6 +85,22 @@ public interface SendHook { */ long SKIP = -1; + /** + * Returned by {@link #beforeSend(Exchange, Endpoint)} to send the message twice, as a sender that delivers a + * message again (to test that the receiver copes with a duplicate). The message is copied before the first send (a + * stream body is cached first, so it can be read twice), and when the first send succeeds the copy is sent again, + * on the thread that completed the first send. When the first send fails, there is no second send. A reply of the + * second send is the reply of the send. {@link #afterSend(Exchange, Endpoint)} is called once, after the second + * send (or after the first one when it failed), and the {@link CamelEvent.ExchangeSendingEvent} and + * {@link CamelEvent.ExchangeSentEvent} are fired once. + * <p/> + * With more than one hook, a hook that returns it makes the send happen twice, unless another hook stops or skips + * the send; the longest delay of the hooks is waited once, before the first send. + * + * @since 4.24 + */ + long SEND_TWICE = -2; + /** * Called before the producer of the endpoint is called. * <p/> @@ -103,9 +120,10 @@ public interface SendHook { * * @param exchange the exchange that is sent * @param endpoint the endpoint it is sent to - * @return {@link #PROCEED} to call the producer now, a delay in milliseconds to wait before calling it, or - * {@link #SKIP} to complete the send without calling it. Other negative values are reserved: the - * send fails with an {@link IllegalArgumentException}, without calling the producer (and + * @return {@link #PROCEED} to call the producer now, a delay in milliseconds to wait before calling it, + * {@link #SKIP} to complete the send without calling it, or {@link #SEND_TWICE} to send the + * message twice. Other negative values are reserved: the send fails with an + * {@link IllegalArgumentException}, without calling the producer (and * {@link #afterSend(Exchange, Endpoint)} is called). The value is ignored when the hook sets an * exception on the exchange. */ diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/SendTwiceTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/SendTwiceTest.java new file mode 100644 index 000000000000..8e945c5d0e67 --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/processor/SendTwiceTest.java @@ -0,0 +1,264 @@ +/* + * 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.ByteArrayInputStream; +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.camel.CamelContext; +import org.apache.camel.ContextTestSupport; +import org.apache.camel.Endpoint; +import org.apache.camel.Exchange; +import org.apache.camel.ExchangePattern; +import org.apache.camel.Ordered; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.spi.CamelEvent; +import org.apache.camel.spi.CamelEvent.ExchangeSendingEvent; +import org.apache.camel.spi.CamelEvent.ExchangeSentEvent; +import org.apache.camel.spi.SendHook; +import org.apache.camel.support.EventNotifierSupport; +import org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository; +import org.junit.jupiter.api.Test; + +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.assertNull; + +/** + * {@link SendHook#SEND_TWICE}: a send that happens twice (CAMEL-26607), as the duplicate fault of camel-chaos. + */ +class SendTwiceTest extends ContextTestSupport { + + // the endpoints the hook sends to twice + private final Set<String> twice = ConcurrentHashMap.newKeySet(); + // the sending and sent events per endpoint + private final Map<String, AtomicInteger> sending = new ConcurrentHashMap<>(); + private final Map<String, AtomicInteger> sent = new ConcurrentHashMap<>(); + private final Map<String, AtomicInteger> after = new ConcurrentHashMap<>(); + private final AtomicInteger replies = new AtomicInteger(); + private final TwiceHook hook = new TwiceHook(); + + @Override + protected CamelContext createCamelContext() throws Exception { + CamelContext context = super.createCamelContext(); + // a stream body stays a stream, so the hook must cache it to send it twice + context.setStreamCaching(false); + context.getManagementStrategy().addEventNotifier(new EventNotifierSupport() { + @Override + public void notify(CamelEvent event) { + if (event instanceof ExchangeSendingEvent e) { + sending.computeIfAbsent(e.getEndpoint().getEndpointUri(), k -> new AtomicInteger()).incrementAndGet(); + } else if (event instanceof ExchangeSentEvent e) { + sent.computeIfAbsent(e.getEndpoint().getEndpointUri(), k -> new AtomicInteger()).incrementAndGet(); + } + } + }); + context.getCamelContextExtension().addSendHook(hook); + return context; + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("direct:start").routeId("start").to("mock:target").to("mock:after"); + + from("direct:inout").routeId("inout").to("direct:reply"); + from("direct:reply").routeId("reply") + .process(e -> e.getMessage().setBody("reply-" + replies.incrementAndGet())); + + from("direct:fail").routeId("fail").to("mock:fail"); + + from("direct:stream").routeId("stream").to("direct:read"); + from("direct:read").routeId("read").convertBodyTo(String.class).to("mock:read"); + + from("direct:orders").routeId("orders").to("direct:idempotent"); + from("direct:idempotent").routeId("idempotent") + .idempotentConsumer(header("orderId"), MemoryIdempotentRepository.memoryIdempotentRepository()) + .to("mock:booked"); + } + }; + } + + @Test + void testInOnlySendsTwice() throws Exception { + twice.add("mock://target"); + MockEndpoint target = getMockEndpoint("mock:target"); + target.expectedBodiesReceived("Hello", "Hello"); + target.expectedHeaderValuesReceivedInAnyOrder("foo", "bar", "bar"); + getMockEndpoint("mock:after").expectedBodiesReceived("Hello"); + + template.sendBodyAndHeader("direct:start", "Hello", "foo", "bar"); + + assertMockEndpointsSatisfied(); + // the events fire once, and the hook is called once after the two sends + assertEquals(1, sending.get("mock://target").get()); + assertEquals(1, sent.get("mock://target").get()); + assertEquals(1, after.get("mock://target").get()); + } + + @Test + void testInOutTheSecondReplyWins() { + twice.add("direct://reply"); + + Object reply = template.requestBody("direct:inout", "Hello"); + + assertEquals(2, replies.get()); + assertEquals("reply-2", reply); + assertEquals(1, after.get("direct://reply").get()); + } + + @Test + void testFailedFirstSendIsNotRepeated() throws Exception { + twice.add("mock://fail"); + MockEndpoint fail = getMockEndpoint("mock:fail"); + fail.whenAnyExchangeReceived(e -> { + throw new IllegalStateException("Down"); + }); + fail.expectedMessageCount(1); + + Exchange out = template.send("direct:fail", e -> e.getMessage().setBody("Hello")); + + MockEndpoint.assertIsSatisfied(context); + assertInstanceOf(IllegalStateException.class, out.getException()); + assertEquals(1, after.get("mock://fail").get()); + } + + @Test + void testStreamBodyIsSentTwiceIntact() throws Exception { + twice.add("direct://read"); + getMockEndpoint("mock:read").expectedBodiesReceived("Hello stream", "Hello stream"); + + template.sendBody("direct:stream", new ByteArrayInputStream("Hello stream".getBytes(StandardCharsets.UTF_8))); + + assertMockEndpointsSatisfied(); + } + + @Test + void testIdempotentConsumerDropsTheDuplicate() throws Exception { + twice.add("direct://idempotent"); + MockEndpoint booked = getMockEndpoint("mock:booked"); + booked.expectedBodiesReceived("order 1", "order 2"); + + template.sendBodyAndHeader("direct:orders", "order 1", "orderId", "1"); + template.sendBodyAndHeader("direct:orders", "order 2", "orderId", "2"); + + assertMockEndpointsSatisfied(); + // each order reached the idempotent consumer twice + assertEquals(2, sending.get("direct://idempotent").get()); + assertEquals(2, after.get("direct://idempotent").get()); + } + + @Test + void testProducerTemplateSendsTwice() throws Exception { + twice.add("mock://target"); + getMockEndpoint("mock:target").expectedBodiesReceived("Hello", "Hello"); + + Exchange out = template.send("mock:target", ExchangePattern.InOnly, e -> e.getMessage().setBody("Hello")); + + assertMockEndpointsSatisfied(); + assertNull(out.getException()); + } + + @Test + void testSkipOfAnotherHookWins() throws Exception { + twice.add("mock://target"); + List<String> calls = new CopyOnWriteArrayList<>(); + // asked before the hook that sends twice: it skips the send, so that hook is not asked + context.getCamelContextExtension().addSendHook(new OrderedHook(-1, SendHook.SKIP, calls)); + getMockEndpoint("mock:target").expectedMessageCount(0); + getMockEndpoint("mock:after").expectedMessageCount(1); + + template.sendBody("direct:start", "Hello"); + + assertMockEndpointsSatisfied(); + assertFalse(after.containsKey("mock://target")); + assertEquals(List.of("before:mock://target", "after:mock://target", "before:mock://after", "after:mock://after"), + calls); + } + + @Test + void testOtherHooksAreCalledOnceAroundBothSends() throws Exception { + twice.add("mock://target"); + List<String> calls = new CopyOnWriteArrayList<>(); + // asked after the hook that sends twice, with a delay: the two sends happen, and it is called once after them + context.getCamelContextExtension().addSendHook(new OrderedHook(Ordered.LOWEST, 1, calls)); + getMockEndpoint("mock:target").expectedBodiesReceived("Hello", "Hello"); + + template.sendBody("direct:start", "Hello"); + + assertMockEndpointsSatisfied(); + assertEquals(List.of("before:mock://target", "after:mock://target", "before:mock://after", "after:mock://after"), + calls); + } + + private final class TwiceHook implements SendHook { + + @Override + public long beforeSend(Exchange exchange, Endpoint endpoint) { + return twice.contains(endpoint.getEndpointUri()) ? SEND_TWICE : PROCEED; + } + + @Override + public void afterSend(Exchange exchange, Endpoint endpoint) { + after.computeIfAbsent(endpoint.getEndpointUri(), k -> new AtomicInteger()).incrementAndGet(); + } + } + + private static final class OrderedHook implements SendHook, Ordered { + + private final int order; + private final long answer; + private final List<String> calls; + + private OrderedHook(int order, long answer, List<String> calls) { + this.order = order; + this.answer = answer; + this.calls = calls; + } + + @Override + public long beforeSend(Exchange exchange, Endpoint endpoint) { + if (!endpoint.getEndpointUri().startsWith("mock:")) { + return PROCEED; + } + calls.add("before:" + endpoint.getEndpointUri()); + return "mock://target".equals(endpoint.getEndpointUri()) ? answer : PROCEED; + } + + @Override + public void afterSend(Exchange exchange, Endpoint endpoint) { + if (endpoint.getEndpointUri().startsWith("mock:")) { + calls.add("after:" + endpoint.getEndpointUri()); + } + } + + @Override + public int getOrder() { + return order; + } + } +} diff --git a/core/camel-core/src/test/java/org/apache/camel/support/CompositeHookTest.java b/core/camel-core/src/test/java/org/apache/camel/support/CompositeHookTest.java index 885ab249fc9a..001d551360c1 100644 --- a/core/camel-core/src/test/java/org/apache/camel/support/CompositeHookTest.java +++ b/core/camel-core/src/test/java/org/apache/camel/support/CompositeHookTest.java @@ -263,6 +263,41 @@ class CompositeHookTest extends ContextTestSupport { assertEquals("b failed after the send", exchange.getException().getMessage()); } + @Test + void testSendTwiceAsksTheLaterHooksAndSendsTwice() { + ExtendedCamelContext ecc = context.getCamelContextExtension(); + ecc.addSendHook(new TestSendHook("twice", 1, SendHook.SEND_TWICE)); + ecc.addSendHook(new TestSendHook("c", 2, SendHook.PROCEED)); + + Exchange exchange = send(); + + // the hooks are asked once and called once after the two sends + assertEquals(List.of("before:twice", "before:c", "producer", "producer", "after:c", "after:twice"), calls); + assertNull(exchange.getException()); + } + + @Test + void testSkipWinsOverSendTwice() { + ExtendedCamelContext ecc = context.getCamelContextExtension(); + ecc.addSendHook(new TestSendHook("twice", 1, SendHook.SEND_TWICE)); + ecc.addSendHook(new TestSendHook("skip", 2, SendHook.SKIP)); + + Exchange exchange = send(); + + assertEquals(List.of("before:twice", "before:skip", "after:skip", "after:twice"), calls); + assertNull(exchange.getException()); + } + + @Test + void testCompositeCalledDirectlyReturnsSendTwice() { + SendHook composite = CompositeSendHook.of(List.of( + new TestSendHook("delay", 1, 30), + new TestSendHook("twice", 2, SendHook.SEND_TWICE))); + + assertEquals(SendHook.SEND_TWICE, composite.beforeSend(new DefaultExchange(context), + context.getEndpoint("mock:target"))); + } + @Test void testLongestSendDelayWins() { SendHook composite = CompositeSendHook.of(List.of( diff --git a/core/camel-core/src/test/java/org/apache/camel/support/SendHelperTest.java b/core/camel-core/src/test/java/org/apache/camel/support/SendHelperTest.java index dd2f99f599f2..65bdaed0073a 100644 --- a/core/camel-core/src/test/java/org/apache/camel/support/SendHelperTest.java +++ b/core/camel-core/src/test/java/org/apache/camel/support/SendHelperTest.java @@ -80,7 +80,8 @@ class SendHelperTest extends ContextTestSupport { DELAY, RESERVED, BEFORE_THROWS, - AFTER_THROWS + AFTER_THROWS, + TWICE } enum Kind { @@ -208,7 +209,7 @@ class SendHelperTest extends ContextTestSupport { assertEquals(1, completion.count.get()); assertTrue(completion.doneSync); assertInstanceOf(IllegalArgumentException.class, exchange.getException()); - assertEquals("SendHook.beforeSend returned -2: negative values other than SKIP are reserved", + assertEquals("SendHook.beforeSend returned -3: negative values other than SKIP and SEND_TWICE are reserved", exchange.getException().getMessage()); assertEquals(0, count(exchange, PRODUCER_COUNT)); assertEquals(1, count(exchange, AFTER_COUNT)); @@ -372,11 +373,11 @@ class SendHelperTest extends ContextTestSupport { void assertCompletedOnce() { assertEquals(1, completion.count.get(), this::toString); - boolean producerCalled = count(exchange, PRODUCER_COUNT) == 1; - assertTrue(count(exchange, PRODUCER_COUNT) <= 1, this::toString); boolean stoppedByHook = mode == Mode.STOP || mode == Mode.SKIP || mode == Mode.RESERVED || mode == Mode.BEFORE_THROWS; - assertEquals(!stoppedByHook, producerCalled, this::toString); + // SEND_TWICE calls the producer again after a first send that did not fail + int producerCalls = stoppedByHook ? 0 : mode == Mode.TWICE && kind != Kind.THROWS ? 2 : 1; + assertEquals(producerCalls, count(exchange, PRODUCER_COUNT), this::toString); // the after hook runs once for every before hook that returned assertEquals(mode == Mode.BEFORE_THROWS ? 0 : 1, count(exchange, AFTER_COUNT), this::toString); @@ -526,9 +527,11 @@ class SendHelperTest extends ContextTestSupport { Long fixed = exchange.getProperty(FIXED_DELAY, Long.class); return fixed != null ? fixed : 1 + ThreadLocalRandom.current().nextInt(3); case RESERVED: - return -2; + return -3; case BEFORE_THROWS: throw new IllegalStateException("Before failed"); + case TWICE: + return SEND_TWICE; default: return PROCEED; } diff --git a/core/camel-support/src/main/java/org/apache/camel/support/CompositeSendHook.java b/core/camel-support/src/main/java/org/apache/camel/support/CompositeSendHook.java index 33847a8832b6..dce6c9a694cb 100644 --- a/core/camel-support/src/main/java/org/apache/camel/support/CompositeSendHook.java +++ b/core/camel-support/src/main/java/org/apache/camel/support/CompositeSendHook.java @@ -34,13 +34,15 @@ import org.apache.camel.spi.SendHook; * <li>{@link #beforeSend(Exchange, Endpoint)} is called on the hooks in their order. A hook that throws an exception, * sets one on the exchange, returns {@link SendHook#SKIP} or a reserved negative value ends it: the hooks after it are * not asked. So the first {@link SendHook#SKIP} wins.</li> - * <li>When all the hooks proceed, the longest of their delays wins: the send waits for it, once.</li> + * <li>When all the hooks proceed, the longest of their delays wins: the send waits for it, once. When a hook returned + * {@link SendHook#SEND_TWICE}, the message is then sent twice.</li> * <li>{@link #afterSend(Exchange, Endpoint)} is called only on the hooks whose {@code beforeSend} returned, in the * reverse order, so the hook that was asked first is called last, as for nested hooks.</li> * </ul> * <p/> * Code that calls a composite itself does not know which hooks were asked: its {@code beforeSend} combines the hooks as - * above, and its {@code afterSend} calls every hook, in the reverse order. + * above (as it returns one value, {@link SendHook#SEND_TWICE} wins over a delay), and its {@code afterSend} calls every + * hook, in the reverse order. * <p/> * It is immutable: the {@link org.apache.camel.CamelContext} creates a new one when a hook is added or removed (see * {@link org.apache.camel.ExtendedCamelContext#addSendHook(SendHook)}). @@ -93,6 +95,7 @@ public final class CompositeSendHook implements SendHook { public long beforeSend(Exchange exchange, Endpoint endpoint) { final Exception existing = exchange.getException(); long answer = PROCEED; + boolean twice = false; for (SendHook hook : hooks) { long value = hook.beforeSend(exchange, endpoint); Exception exception = exchange.getException(); @@ -100,13 +103,17 @@ public final class CompositeSendHook implements SendHook { // the hook stopped the send: the return value is ignored return PROCEED; } - if (value < 0) { + if (value == SEND_TWICE) { + twice = true; + } else if (value < 0) { // SKIP, or a reserved value that fails the send return value; + } else { + answer = Math.max(answer, value); } - answer = Math.max(answer, value); } - return answer; + // one value only: sending twice wins over a delay + return twice ? SEND_TWICE : answer; } @Override diff --git a/core/camel-support/src/main/java/org/apache/camel/support/SendHelper.java b/core/camel-support/src/main/java/org/apache/camel/support/SendHelper.java index cefce803acb4..4efd3774c005 100644 --- a/core/camel-support/src/main/java/org/apache/camel/support/SendHelper.java +++ b/core/camel-support/src/main/java/org/apache/camel/support/SendHelper.java @@ -17,6 +17,8 @@ package org.apache.camel.support; import java.io.IOException; +import java.io.InputStream; +import java.io.Reader; import java.util.concurrent.CompletableFuture; import org.apache.camel.AsyncCallback; @@ -25,7 +27,9 @@ import org.apache.camel.AsyncProducer; import org.apache.camel.CamelContext; import org.apache.camel.Endpoint; import org.apache.camel.Exchange; +import org.apache.camel.Message; import org.apache.camel.ShutdownableService; +import org.apache.camel.StreamCache; import org.apache.camel.spi.CamelEvent; import org.apache.camel.spi.SendHook; import org.apache.camel.support.service.ServiceHelper; @@ -170,6 +174,10 @@ public final class SendHelper { * <p/> * With several hooks (a {@link CompositeSendHook}) the hooks are asked one by one, see {@link CompositeSendHook}: * the send keeps how many were asked, so only those are called after it, in the reverse order. + * <p/> + * When a hook returns {@link SendHook#SEND_TWICE}, the message is copied before the first send, and when the first + * send succeeded the copy is sent again: by this thread when the first send completed synchronously, else by the + * thread that completed it, so the exchange is on one thread at a time. The hooks are called after the second send. */ private static final class HookedSend implements AsyncCallback { @@ -183,6 +191,12 @@ public final class SendHelper { private final AsyncCallback callback; // how many hooks returned from beforeSend: they, and only they, are called after the send private int asked; + // a hook returned SEND_TWICE + private boolean twice; + // the exception the exchange had before the send, which is not a failure of the send + private Exception existing; + // the copy of the message for the second send, while the first one runs + private Message again; private HookedSend(SendHook hook, Exchange exchange, Endpoint endpoint, AsyncProcessor producer, AsyncCallback callback) { @@ -201,10 +215,10 @@ public final class SendHelper { boolean send() { // an exchange can be sent with an exception on it already, which does not stop the send - final Exception existing = exchange.getException(); + existing = exchange.getException(); final long delay; try { - delay = beforeSend(existing); + delay = beforeSend(); } catch (Exception e) { // a hook failed: the producer is not called, and there is no after for that hook (the hooks asked // before it are called after the send, as for any send they proceeded with) @@ -222,7 +236,8 @@ public final class SendHelper { if (delay < 0) { // the other negative values are reserved, so a hook that returns one fails the send exchange.setException(new IllegalArgumentException( - "SendHook.beforeSend returned " + delay + ": negative values other than SKIP are reserved")); + "SendHook.beforeSend returned " + delay + + ": negative values other than SKIP and SEND_TWICE are reserved")); return completeWithoutProducer(); } @@ -246,9 +261,19 @@ public final class SendHelper { } try { - return producer.process(exchange, this); + if (!twice) { + return producer.process(exchange, this); + } + again = copyForSecondSend(); + if (!producer.process(exchange, this::firstSendDone)) { + // the first send goes on asynchronously: the thread that completes it makes the second send + return false; + } + // the first send completed on this thread, which makes the second send + return secondSend(true); } catch (RuntimeException e) { // the caller handles a producer that throws as it does without a hook + again = null; exchange.setException(e); afterSend(); throw e; @@ -256,16 +281,83 @@ public final class SendHelper { } /** - * Asks the hooks before the send. + * Copies the message for the second send of {@link SendHook#SEND_TWICE}. A stream body is cached first, so both + * sends can read it. + */ + private Message copyForSecondSend() { + Message message = exchange.getMessage(); + Object body = message.getBody(); + if ((body instanceof InputStream || body instanceof Reader) && !(body instanceof StreamCache)) { + StreamCache cache = exchange.getContext().getTypeConverter().tryConvertTo(StreamCache.class, exchange, + body); + if (cache != null) { + message.setBody(cache); + } + } + return message.copy(); + } + + /** + * The first of two sends is done. When it completed synchronously, {@link #send()} makes the second send on the + * same thread after the producer returned; else this thread makes it. + */ + private void firstSendDone(boolean doneSync) { + if (doneSync) { + return; + } + try { + secondSend(false); + } catch (RuntimeException e) { + // the producer threw instead of calling the callback: the send fails + exchange.setException(e); + done(false); + } + } + + /** + * Makes the second send of {@link SendHook#SEND_TWICE}, with the copy of the message, unless the first send + * failed. + * + * @param sync whether the first send completed synchronously (and the caller of {@link #send()} waits for the + * return value) + * @return whether the send completed synchronously + */ + private boolean secondSend(boolean sync) { + final Message message = again; + again = null; + final Exception exception = exchange.getException(); + if (exception != null && exception != existing) { + // the first send failed: there is no second send + done(sync); + return sync; + } + // the copy is the in message, and the out message too if the first send made one + exchange.setIn(message); + exchange.setMessage(message); + // the first send read the cached stream of the body + MessageHelper.resetStreamCache(message); + if (sync) { + return producer.process(exchange, this); + } + // the caller was told the send goes on asynchronously + producer.process(exchange, doneSync -> done(false)); + return false; + } + + /** + * Asks the hooks before the send. A {@link SendHook#SEND_TWICE} is kept in {@link #twice}. * - * @param existing the exception the exchange had before the send - * @return what the hooks decided: {@link SendHook#PROCEED}, a delay, {@link SendHook#SKIP} or a - * reserved value (ignored when a hook set an exception on the exchange) + * @return what the hooks decided: {@link SendHook#PROCEED}, a delay, {@link SendHook#SKIP} or a reserved value + * (ignored when a hook set an exception on the exchange) */ - private long beforeSend(Exception existing) { + private long beforeSend() { if (hooks == null) { long answer = hook.beforeSend(exchange, endpoint); asked = 1; + if (answer == SendHook.SEND_TWICE) { + twice = true; + return SendHook.PROCEED; + } return answer; } long answer = SendHook.PROCEED; @@ -277,6 +369,11 @@ public final class SendHelper { // this hook stopped the send: the later hooks are not asked return SendHook.PROCEED; } + if (value == SendHook.SEND_TWICE) { + // the later hooks are asked too, and a SKIP of one of them still wins + twice = true; + continue; + } if (value < 0) { // the first SKIP wins (or a reserved value fails the send): the later hooks are not asked return value; diff --git a/design/chaos.adoc b/design/chaos.adoc index cc7bbc81b435..fd00c3da8d43 100644 --- a/design/chaos.adoc +++ b/design/chaos.adoc @@ -129,7 +129,7 @@ A rule has a target, a fault, how often it fires, and when it ends: ---- target: route=<id or pattern> | endpoint=<uri pattern> | component=<name> | step=<processor id pattern> at: send | poll | entry | step (optional, see Target and hooks) -fault: error | delay | drop | corrupt | lifecycle (stall) (parameters: see Rule syntax) +fault: error | delay | drop | corrupt | duplicate | lifecycle (stall) (parameters: see Rule syntax) rate: a probability (0.3), or every Nth match until: a number of matches, a duration, or removed by hand ---- @@ -180,6 +180,11 @@ an `InputStream` body that ends too soon. `charset` applies to a body that is te to an `InputStream` or stream cache body; on another body the message is left as it is and the report counts the rule as skipped for that message, so a fault that did nothing is visible. +|`duplicate` +|none +|Sends the message twice (CAMEL-26607), with `SendHook.SEND_TWICE`: the message is copied before the send, and the +copy is sent again when the first send succeeded. For the sends only, to test idempotent receivers. + |`lifecycle` |`action=suspend\|stop\|remove`, `duration=<time>` |Suspends, stops or removes the target route, then resumes it, starts it or adds its definition again after the @@ -204,7 +209,7 @@ Each target has these hooks, and each hook offers some of the faults: |`endpoint=`, `component=` |The sends to a matching endpoint (`send`) and the polls of its consumer (`poll`). -|`send`: `error`, `delay`, `drop`, `corrupt`. `poll`: `error`, `delay`, `drop`. +|`send`: `error`, `delay`, `drop`, `corrupt`, `duplicate`. `poll`: `error`, `delay`, `drop`. |`route=` |The messages that enter the route (`entry`), and the route lifecycle. diff --git a/docs/user-manual/modules/ROOT/pages/security-model.adoc b/docs/user-manual/modules/ROOT/pages/security-model.adoc index 322be0f6e27d..3b5a4a23b900 100644 --- a/docs/user-manual/modules/ROOT/pages/security-model.adoc +++ b/docs/user-manual/modules/ROOT/pages/security-model.adoc @@ -1835,13 +1835,13 @@ explicit-opt-in entry under _Out of scope_. | `camel.chaos.enabled=true` (`camel-chaos`) | unset (on with profile `dev`, never with profile `prod`) -| Chaos is switched on, and its rules fail, delay, drop and corrupt messages on purpose. The rules come only from the - developer: `camel.chaos.rules` (marked `insecure:dev`), Java, and the CLI connector, which the `camel cmd chaos-*` - commands and the chaos tools of `camel ask` and of the camel-jbang MCP server use. They are never added over the - HTTP server of the application: the `chaos` developer console and the development MCP tools of `camel run --mcp` - only list and report. The `exception=` of a rule names any public `Throwable` class on the classpath, whose - constructor runs when the fault fires. With profile `prod` chaos stays off, whatever `camel.chaos.enabled` and - `camel.chaos.rules` say. +| Chaos is switched on, and its rules fail, delay, drop, duplicate and corrupt messages on purpose. The rules come + only from the developer: `camel.chaos.rules` (marked `insecure:dev`), Java, and the CLI connector, which the + `camel cmd chaos-*` commands and the chaos tools of `camel ask` and of the camel-jbang MCP server use. They are + never added over the HTTP server of the application: the `chaos` developer console and the development MCP tools of + `camel run --mcp` only list and report. The `exception=` of a rule names any public `Throwable` class on the + classpath, whose constructor runs when the fault fires. With profile `prod` chaos stays off, whatever + `camel.chaos.enabled` and `camel.chaos.rules` say. |=== A machine-readable index of these variants, the entry-point trust levels, the
