This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-25589-hazelcast-seda-rollback-delay in repository https://gitbox.apache.org/repos/asf/camel.git
commit db5bc83ca89a714b956fdea43a0f6e4e65d16844 Author: Claus Ibsen <[email protected]> AuthorDate: Thu Nov 12 21:20:56 2026 +0100 CAMEL-25589: camel-hazelcast - the transacted SEDA consumer waits onErrorDelay before it takes a rolled back message again When the transacted hazelcast-seda consumer rolled back the transaction of a failed exchange (or of one marked rollback only), it polled the queue again at once: onErrorDelay applied only to an exception from the poll or the transaction itself. A message that always fails was then taken and rolled back in a tight loop. The delay is now a waitOnErrorDelay() method, and the transacted path calls it after onExchangeDone() rolled the transaction back. The non-transacted path does not, as nothing goes back on the queue there. The onErrorDelay option description says it applies after a rollback too. Co-Authored-By: Claude Opus 5.5 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../camel/catalog/components/hazelcast-seda.json | 2 +- .../component/hazelcast/seda/hazelcast-seda.json | 2 +- .../hazelcast/seda/HazelcastSedaConfiguration.java | 3 +- .../hazelcast/seda/HazelcastSedaConsumer.java | 47 ++++++--- .../HazelcastSedaTransactedRollbackDelayTest.java | 109 +++++++++++++++++++++ .../dsl/HazelcastSedaEndpointBuilderFactory.java | 30 +++--- 6 files changed, 163 insertions(+), 30 deletions(-) diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/hazelcast-seda.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/hazelcast-seda.json index 3c655c5e30e3..26c9b6ab8a12 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/hazelcast-seda.json +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/hazelcast-seda.json @@ -46,7 +46,7 @@ "exchangePattern": { "index": 7, "kind": "parameter", "displayName": "Exchange Pattern", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "enum", "javaType": "org.apache.camel.ExchangePattern", "enum": [ "InOnly", "InOut" ], "deprecated": false, "autowired": false, "secret": false, "description": "Sets the exchange pattern when the consumer creates an exchange." }, "lazyStartProducer": { "index": 8, "kind": "parameter", "displayName": "Lazy Start Producer", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a produc [...] "concurrentConsumers": { "index": 9, "kind": "parameter", "displayName": "Concurrent Consumers", "group": "seda", "label": "seda", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 1, "configurationClass": "org.apache.camel.component.hazelcast.seda.HazelcastSedaConfiguration", "configurationField": "configuration", "description": "To use concurrent consumers polling from the SEDA queue." }, - "onErrorDelay": { "index": 10, "kind": "parameter", "displayName": "On Error Delay", "group": "seda", "label": "seda", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 1000, "configurationClass": "org.apache.camel.component.hazelcast.seda.HazelcastSedaConfiguration", "configurationField": "configuration", "description": "Milliseconds before consumer continues polling after an error has occurred." }, + "onErrorDelay": { "index": 10, "kind": "parameter", "displayName": "On Error Delay", "group": "seda", "label": "seda", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 1000, "configurationClass": "org.apache.camel.component.hazelcast.seda.HazelcastSedaConfiguration", "configurationField": "configuration", "description": "Milliseconds before the consumer continues polling after an error has occurred, or [...] "pollTimeout": { "index": 11, "kind": "parameter", "displayName": "Poll Timeout", "group": "seda", "label": "seda", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 1000, "configurationClass": "org.apache.camel.component.hazelcast.seda.HazelcastSedaConfiguration", "configurationField": "configuration", "description": "The timeout used when consuming from the SEDA queue. When a timeout occurs, the consum [...] "transacted": { "index": 12, "kind": "parameter", "displayName": "Transacted", "group": "seda", "label": "seda", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.hazelcast.seda.HazelcastSedaConfiguration", "configurationField": "configuration", "description": "If set to true then the consumer runs in transaction mode, where the messages in the [...] "transferExchange": { "index": 13, "kind": "parameter", "displayName": "Transfer Exchange", "group": "seda", "label": "seda", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "security": "insecure:serialization", "defaultValue": false, "configurationClass": "org.apache.camel.component.hazelcast.seda.HazelcastSedaConfiguration", "configurationField": "configuration", "description": "If set to true the whole Exchange [...] diff --git a/components/camel-hazelcast/src/generated/resources/META-INF/org/apache/camel/component/hazelcast/seda/hazelcast-seda.json b/components/camel-hazelcast/src/generated/resources/META-INF/org/apache/camel/component/hazelcast/seda/hazelcast-seda.json index 3c655c5e30e3..26c9b6ab8a12 100644 --- a/components/camel-hazelcast/src/generated/resources/META-INF/org/apache/camel/component/hazelcast/seda/hazelcast-seda.json +++ b/components/camel-hazelcast/src/generated/resources/META-INF/org/apache/camel/component/hazelcast/seda/hazelcast-seda.json @@ -46,7 +46,7 @@ "exchangePattern": { "index": 7, "kind": "parameter", "displayName": "Exchange Pattern", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "enum", "javaType": "org.apache.camel.ExchangePattern", "enum": [ "InOnly", "InOut" ], "deprecated": false, "autowired": false, "secret": false, "description": "Sets the exchange pattern when the consumer creates an exchange." }, "lazyStartProducer": { "index": 8, "kind": "parameter", "displayName": "Lazy Start Producer", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a produc [...] "concurrentConsumers": { "index": 9, "kind": "parameter", "displayName": "Concurrent Consumers", "group": "seda", "label": "seda", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 1, "configurationClass": "org.apache.camel.component.hazelcast.seda.HazelcastSedaConfiguration", "configurationField": "configuration", "description": "To use concurrent consumers polling from the SEDA queue." }, - "onErrorDelay": { "index": 10, "kind": "parameter", "displayName": "On Error Delay", "group": "seda", "label": "seda", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 1000, "configurationClass": "org.apache.camel.component.hazelcast.seda.HazelcastSedaConfiguration", "configurationField": "configuration", "description": "Milliseconds before consumer continues polling after an error has occurred." }, + "onErrorDelay": { "index": 10, "kind": "parameter", "displayName": "On Error Delay", "group": "seda", "label": "seda", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 1000, "configurationClass": "org.apache.camel.component.hazelcast.seda.HazelcastSedaConfiguration", "configurationField": "configuration", "description": "Milliseconds before the consumer continues polling after an error has occurred, or [...] "pollTimeout": { "index": 11, "kind": "parameter", "displayName": "Poll Timeout", "group": "seda", "label": "seda", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 1000, "configurationClass": "org.apache.camel.component.hazelcast.seda.HazelcastSedaConfiguration", "configurationField": "configuration", "description": "The timeout used when consuming from the SEDA queue. When a timeout occurs, the consum [...] "transacted": { "index": 12, "kind": "parameter", "displayName": "Transacted", "group": "seda", "label": "seda", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.hazelcast.seda.HazelcastSedaConfiguration", "configurationField": "configuration", "description": "If set to true then the consumer runs in transaction mode, where the messages in the [...] "transferExchange": { "index": 13, "kind": "parameter", "displayName": "Transfer Exchange", "group": "seda", "label": "seda", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "security": "insecure:serialization", "defaultValue": false, "configurationClass": "org.apache.camel.component.hazelcast.seda.HazelcastSedaConfiguration", "configurationField": "configuration", "description": "If set to true the whole Exchange [...] diff --git a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConfiguration.java b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConfiguration.java index 544285bb276e..a376900a80e7 100644 --- a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConfiguration.java +++ b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConfiguration.java @@ -94,7 +94,8 @@ public class HazelcastSedaConfiguration { } /** - * Milliseconds before consumer continues polling after an error has occurred. + * Milliseconds before the consumer continues polling after an error has occurred, or after a transacted consumer + * rolled back the transaction of a failed exchange (the message is then on the queue again). */ public void setOnErrorDelay(int onErrorDelay) { if (onErrorDelay < 0) { diff --git a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConsumer.java b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConsumer.java index 61f504fc936f..c575177e691f 100644 --- a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConsumer.java +++ b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConsumer.java @@ -133,7 +133,11 @@ public class HazelcastSedaConsumer extends DefaultConsumer implements Runnable { } catch (Exception e) { exchange.setException(e); } - onExchangeDone(exchange, txc); + if (onExchangeDone(exchange, txc)) { + // the message is back on the queue: wait before taking it again, or a message that + // always fails is taken and rolled back in a tight loop + waitOnErrorDelay(); + } } else { // process using the asynchronous routing engine processor.process(exchange, new AsyncCallback() { @@ -171,19 +175,7 @@ public class HazelcastSedaConsumer extends DefaultConsumer implements Runnable { } } getExceptionHandler().handleException("Error processing exchange", exchange, e); - // Use Camel's task API for error recovery delay instead of Thread.sleep() - // We use initialDelay for the actual delay, and maxIterations(1) to run once - Tasks.backgroundTask() - .withBudget(Budgets.iterationTimeBudget() - .withMaxIterations(1) - .withInitialDelay(Duration.ofMillis(endpoint.getConfiguration().getOnErrorDelay())) - .withInterval(Duration.ofMillis(1)) - .withUnlimitedDuration() - .build()) - .withScheduledExecutor(taskExecutor) - .withName("HazelcastSedaErrorRecoveryDelay") - .build() - .run(getEndpoint().getCamelContext(), () -> true); + waitOnErrorDelay(); } finally { if (transacted) { releaseExchange(exchange, false); @@ -213,7 +205,31 @@ public class HazelcastSedaConsumer extends DefaultConsumer implements Runnable { exchange.getIn().setBody(transferred.getIn().getBody()); } - private void onExchangeDone(Exchange exchange, TransactionContext txc) { + /** + * Waits onErrorDelay before the consumer polls again, after an error or a rolled back transaction. + */ + private void waitOnErrorDelay() { + // Use Camel's task API for error recovery delay instead of Thread.sleep() + // We use initialDelay for the actual delay, and maxIterations(1) to run once + Tasks.backgroundTask() + .withBudget(Budgets.iterationTimeBudget() + .withMaxIterations(1) + .withInitialDelay(Duration.ofMillis(endpoint.getConfiguration().getOnErrorDelay())) + .withInterval(Duration.ofMillis(1)) + .withUnlimitedDuration() + .build()) + .withScheduledExecutor(taskExecutor) + .withName("HazelcastSedaErrorRecoveryDelay") + .build() + .run(getEndpoint().getCamelContext(), () -> true); + } + + /** + * Ends the transaction of the exchange, if any, and handles its exception. + * + * @return true when the transaction was rolled back, so the message is on the queue again + */ + private boolean onExchangeDone(Exchange exchange, TransactionContext txc) { // a rollback only exchange (without an exception) did not complete successfully either boolean rollback = exchange.getException() != null || exchange.isRollbackOnly() || exchange.isRollbackOnlyLast(); @@ -229,6 +245,7 @@ public class HazelcastSedaConsumer extends DefaultConsumer implements Runnable { LOG.trace("Commit transaction: {}", txc.getTxnId()); txc.commitTransaction(); } + return rollback && txc != null; } } diff --git a/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastSedaTransactedRollbackDelayTest.java b/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastSedaTransactedRollbackDelayTest.java new file mode 100644 index 000000000000..596e1098964a --- /dev/null +++ b/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastSedaTransactedRollbackDelayTest.java @@ -0,0 +1,109 @@ +/* + * 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.hazelcast; + +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import com.hazelcast.collection.IQueue; +import com.hazelcast.core.HazelcastInstance; +import com.hazelcast.transaction.TransactionContext; +import com.hazelcast.transaction.TransactionalQueue; +import org.apache.camel.builder.RouteBuilder; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.mockito.Mock; +import org.mockito.Mockito; + +import static org.awaitility.Awaitility.await; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.atLeastOnce; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * A message whose exchange failed rolls back the Hazelcast transaction, so it is still on the queue: the transacted + * consumer must wait onErrorDelay before it polls again, and not take the same message again at once (CAMEL-25589). + */ +public class HazelcastSedaTransactedRollbackDelayTest extends HazelcastCamelTestSupport { + + private static final int ON_ERROR_DELAY = 3000; + + @Mock + private IQueue<Object> queue; + + @Mock + private TransactionalQueue<Object> tqueue; + + private final TransactionContext transaction = Mockito.mock(TransactionContext.class); + + private final AtomicInteger deliveries = new AtomicInteger(); + + @Override + protected void trainHazelcastInstance(HazelcastInstance hazelcastInstance) { + when(hazelcastInstance.newTransactionContext()).thenReturn(transaction); + when(hazelcastInstance.getQueue("foo")).thenReturn(queue); + when(transaction.getQueue("foo")).thenReturn(tqueue); + } + + @Override + protected void verifyHazelcastInstance(HazelcastInstance hazelcastInstance) { + verify(hazelcastInstance, atLeastOnce()).getQueue("foo"); + verify(hazelcastInstance, atLeastOnce()).newTransactionContext(); + } + + @Test + public void testRolledBackMessageIsTakenAgainAfterOnErrorDelay() throws Exception { + // the rolled back message is still on the queue, so every poll takes it again + when(tqueue.poll(anyLong(), any(TimeUnit.class))).thenReturn("kaboom"); + + context.getRouteController().startRoute("foo"); + + await().atMost(5, TimeUnit.SECONDS).until(() -> deliveries.get() >= 1); + + // without the delay, the message is taken and rolled back again in a tight loop + await().during(ON_ERROR_DELAY - 1000, TimeUnit.MILLISECONDS) + .atMost(ON_ERROR_DELAY + 1000, TimeUnit.MILLISECONDS) + .until(() -> deliveries.get() == 1); + + // and after the delay it is taken again + await().atMost(ON_ERROR_DELAY + 2000, TimeUnit.MILLISECONDS).until(() -> deliveries.get() >= 2); + + verify(transaction, atLeastOnce()).rollbackTransaction(); + verify(transaction, never()).commitTransaction(); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("hazelcast-seda:foo?transacted=true&onErrorDelay=" + ON_ERROR_DELAY).routeId("foo") + .autoStartup(false) + .process(e -> deliveries.incrementAndGet()) + .throwException(new IllegalStateException("Forced")); + } + }; + } + + @AfterEach + public final void stopContext() throws Exception { + context.stop(); + } +} diff --git a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/HazelcastSedaEndpointBuilderFactory.java b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/HazelcastSedaEndpointBuilderFactory.java index 49334cd28f2c..81d7bb4842a5 100644 --- a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/HazelcastSedaEndpointBuilderFactory.java +++ b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/HazelcastSedaEndpointBuilderFactory.java @@ -174,8 +174,9 @@ public interface HazelcastSedaEndpointBuilderFactory { return this; } /** - * Milliseconds before consumer continues polling after an error has - * occurred. + * Milliseconds before the consumer continues polling after an error has + * occurred, or after a transacted consumer rolled back the transaction + * of a failed exchange (the message is then on the queue again). * * The option is a: <code>int</code> type. * @@ -190,8 +191,9 @@ public interface HazelcastSedaEndpointBuilderFactory { return this; } /** - * Milliseconds before consumer continues polling after an error has - * occurred. + * Milliseconds before the consumer continues polling after an error has + * occurred, or after a transacted consumer rolled back the transaction + * of a failed exchange (the message is then on the queue again). * * The option will be converted to a <code>int</code> type. * @@ -595,8 +597,9 @@ public interface HazelcastSedaEndpointBuilderFactory { return this; } /** - * Milliseconds before consumer continues polling after an error has - * occurred. + * Milliseconds before the consumer continues polling after an error has + * occurred, or after a transacted consumer rolled back the transaction + * of a failed exchange (the message is then on the queue again). * * The option is a: <code>int</code> type. * @@ -611,8 +614,9 @@ public interface HazelcastSedaEndpointBuilderFactory { return this; } /** - * Milliseconds before consumer continues polling after an error has - * occurred. + * Milliseconds before the consumer continues polling after an error has + * occurred, or after a transacted consumer rolled back the transaction + * of a failed exchange (the message is then on the queue again). * * The option will be converted to a <code>int</code> type. * @@ -945,8 +949,9 @@ public interface HazelcastSedaEndpointBuilderFactory { return this; } /** - * Milliseconds before consumer continues polling after an error has - * occurred. + * Milliseconds before the consumer continues polling after an error has + * occurred, or after a transacted consumer rolled back the transaction + * of a failed exchange (the message is then on the queue again). * * The option is a: <code>int</code> type. * @@ -961,8 +966,9 @@ public interface HazelcastSedaEndpointBuilderFactory { return this; } /** - * Milliseconds before consumer continues polling after an error has - * occurred. + * Milliseconds before the consumer continues polling after an error has + * occurred, or after a transacted consumer rolled back the transaction + * of a failed exchange (the message is then on the queue again). * * The option will be converted to a <code>int</code> type. *
