This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 4b504fd443b4 CAMEL-25730: camel-web3j - subscribe again when the
subscription of the consumer ends with an error (#28037)
4b504fd443b4 is described below
commit 4b504fd443b41b8f6096854cb44effd16688d4c0
Author: allthingssecurity <[email protected]>
AuthorDate: Sun Oct 18 21:00:32 2026 +0530
CAMEL-25730: camel-web3j - subscribe again when the subscription of the
consumer ends with an error (#28037)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/component/web3j/Web3jConsumer.java | 122 ++++++++++++-
.../web3j/Web3jConsumerResubscribeTest.java | 196 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 11 ++
3 files changed, 323 insertions(+), 6 deletions(-)
diff --git
a/components/camel-web3j/src/main/java/org/apache/camel/component/web3j/Web3jConsumer.java
b/components/camel-web3j/src/main/java/org/apache/camel/component/web3j/Web3jConsumer.java
index 1716ab9e8efb..ebff5d510a80 100644
---
a/components/camel-web3j/src/main/java/org/apache/camel/component/web3j/Web3jConsumer.java
+++
b/components/camel-web3j/src/main/java/org/apache/camel/component/web3j/Web3jConsumer.java
@@ -16,6 +16,10 @@
*/
package org.apache.camel.component.web3j;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
import io.reactivex.disposables.Disposable;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
@@ -35,10 +39,16 @@ import static
org.apache.camel.component.web3j.Web3jHelper.toDefaultBlockParamet
*/
public class Web3jConsumer extends DefaultConsumer {
private static final Logger LOG =
LoggerFactory.getLogger(Web3jConsumer.class);
+ // the delay before subscribing again after a subscription ended with an
error
+ private static final long DEFAULT_RESUBSCRIBE_DELAY = 15000;
private final Web3jConfiguration configuration;
private Web3j web3j;
- private Disposable subscription;
+ private volatile Disposable subscription;
private Web3jEndpoint endpoint;
+ private long resubscribeDelay = DEFAULT_RESUBSCRIBE_DELAY;
+ private volatile ScheduledExecutorService resubscribeExecutor;
+ // true from a subscription error until the next item, so the failures
while subscribing again are routed once
+ private volatile boolean failing;
public Web3jConsumer(Web3jEndpoint endpoint, Processor processor,
Web3jConfiguration configuration) {
super(endpoint, processor);
@@ -51,13 +61,31 @@ public class Web3jConsumer extends DefaultConsumer {
return (Web3jEndpoint) super.getEndpoint();
}
+ /**
+ * Sets the delay in millis before subscribing again after a subscription
ended with an error. Only for tests.
+ */
+ void setResubscribeDelay(long resubscribeDelay) {
+ this.resubscribeDelay = resubscribeDelay;
+ }
+
@Override
protected void doStart() throws Exception {
super.doStart();
this.web3j = getEndpoint().getWeb3j();
+ failing = false;
+ resubscribeExecutor =
getEndpoint().getCamelContext().getExecutorServiceManager()
+ .newSingleThreadScheduledExecutor(this,
"Web3jConsumerResubscribe");
- LOG.info("Subscribing to: {}", endpoint.getNodeAddress());
+ subscribe();
+ }
+
+ private void subscribe() {
+ if (failing) {
+ LOG.debug("Subscribing again to: {}", endpoint.getNodeAddress());
+ } else {
+ LOG.info("Subscribing to: {}", endpoint.getNodeAddress());
+ }
switch (configuration.getOperation()) {
case Web3jConstants.ETH_LOG_OBSERVABLE:
EthFilter ethFilter =
Web3jEndpoint.buildEthFilter(toDefaultBlockParameter(configuration.getFromBlock()),
@@ -125,10 +153,25 @@ public class Web3jConsumer extends DefaultConsumer {
throw new IllegalArgumentException("Unsupported operation " +
configuration.getOperation());
}
- LOG.info("Subscribed: {}", this.configuration);
+ if (failing) {
+ LOG.debug("Subscribed again: {}", this.configuration);
+ } else {
+ LOG.info("Subscribed: {}", this.configuration);
+ }
+ }
+
+ /**
+ * An item was received, so the subscription works: an error after this
one is routed and logged again.
+ */
+ private void received() {
+ if (failing) {
+ failing = false;
+ LOG.info("Receiving again from: {}", endpoint.getNodeAddress());
+ }
}
private void ethBlockHashObservable(String x) {
+ received();
LOG.debug("processEthBlock {}", x);
Exchange exchange = this.getEndpoint().createExchange();
exchange.getIn().setBody(x);
@@ -136,6 +179,7 @@ public class Web3jConsumer extends DefaultConsumer {
}
private void ethPendingTransactionHashObservable(String x) {
+ received();
LOG.debug("processEthBlock {}", x);
Exchange exchange = this.getEndpoint().createExchange();
exchange.getIn().setBody(x);
@@ -143,6 +187,7 @@ public class Web3jConsumer extends DefaultConsumer {
}
private void blockObservable(EthBlock x) {
+ received();
EthBlock.Block block = x.getBlock();
LOG.debug("processEthBlock {}", block);
Exchange exchange = this.getEndpoint().createExchange();
@@ -151,6 +196,7 @@ public class Web3jConsumer extends DefaultConsumer {
}
private void processTransaction(Transaction x) {
+ received();
LOG.debug("processTransaction {}", x);
Exchange exchange = this.getEndpoint().createExchange();
exchange.getIn().setBody(x);
@@ -158,6 +204,7 @@ public class Web3jConsumer extends DefaultConsumer {
}
private void ethLogObservable(Log x) {
+ received();
LOG.debug("processLogObservable {}", x);
Exchange exchange = this.getEndpoint().createExchange();
exchange.getIn().setBody(x);
@@ -183,13 +230,76 @@ public class Web3jConsumer extends DefaultConsumer {
private void processError(Throwable throwable, String operation) {
LOG.debug("processError for operation: {} {}", operation,
throwable.getMessage(), throwable);
- Exchange exchange = this.getEndpoint().createExchange();
- exchange.setException(throwable);
- processEvent(exchange);
+ boolean again = resubscribes(operation) && isRunAllowed();
+ if (!failing) {
+ failing = true;
+ if (again) {
+ LOG.warn("The {} subscription to {} ended with an error,
subscribing again every {} ms: {}", operation,
+ endpoint.getNodeAddress(), resubscribeDelay,
throwable.getMessage());
+ }
+ Exchange exchange = this.getEndpoint().createExchange();
+ exchange.setException(throwable);
+ processEvent(exchange);
+ } else {
+ LOG.debug("Subscribing again to {} for operation {} failed: {}",
endpoint.getNodeAddress(), operation,
+ throwable.getMessage());
+ }
+ ScheduledExecutorService executor = resubscribeExecutor;
+ if (again && executor != null) {
+ try {
+ executor.schedule(() -> resubscribe(operation),
resubscribeDelay, TimeUnit.MILLISECONDS);
+ } catch (RejectedExecutionException e) {
+ // the consumer is stopping
+ LOG.debug("Not subscribing again to {} as the consumer is
stopping", endpoint.getNodeAddress());
+ }
+ }
+ }
+
+ /**
+ * Whether a subscription that ended with an error is subscribed again:
the live operations, which start again at
+ * the latest block. Not the replay of a block range, nor a log filter
from a given block, which would route the
+ * blocks or logs from the start of the range again.
+ */
+ private boolean resubscribes(String operation) {
+ if (Web3jConstants.REPLAY_BLOCKS_OBSERVABLE.equals(operation)
+ ||
Web3jConstants.REPLAY_TRANSACTIONS_OBSERVABLE.equals(operation)) {
+ return false;
+ }
+ if (Web3jConstants.ETH_LOG_OBSERVABLE.equals(operation)) {
+ String fromBlock = configuration.getFromBlock();
+ return fromBlock == null || "latest".equalsIgnoreCase(fromBlock)
|| "pending".equalsIgnoreCase(fromBlock);
+ }
+ return true;
+ }
+
+ private void resubscribe(String operation) {
+ if (!isRunAllowed()) {
+ return;
+ }
+ Disposable old = subscription;
+ if (old != null && !old.isDisposed()) {
+ old.dispose();
+ }
+ try {
+ subscribe();
+ } catch (Exception e) {
+ processError(e, operation);
+ return;
+ }
+ // doStop may have run while subscribing (it does not wait for this
task), and disposed the previous
+ // subscription only
+ Disposable current = subscription;
+ if (!isRunAllowed() && current != null && !current.isDisposed()) {
+ current.dispose();
+ }
}
@Override
protected void doStop() throws Exception {
+ if (resubscribeExecutor != null) {
+
getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(resubscribeExecutor);
+ resubscribeExecutor = null;
+ }
if (subscription != null && !subscription.isDisposed()) {
subscription.dispose();
}
diff --git
a/components/camel-web3j/src/test/java/org/apache/camel/component/web3j/Web3jConsumerResubscribeTest.java
b/components/camel-web3j/src/test/java/org/apache/camel/component/web3j/Web3jConsumerResubscribeTest.java
new file mode 100644
index 000000000000..294a71d8f0d8
--- /dev/null
+++
b/components/camel-web3j/src/test/java/org/apache/camel/component/web3j/Web3jConsumerResubscribeTest.java
@@ -0,0 +1,196 @@
+/*
+ * 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.web3j;
+
+import java.io.IOException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import io.reactivex.Flowable;
+import org.apache.camel.EndpointInject;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.spi.Registry;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+import org.web3j.protocol.Web3j;
+import org.web3j.protocol.core.methods.response.EthBlock;
+import org.web3j.protocol.core.methods.response.Log;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.Mockito.after;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * A live subscription that ends with an error (such as a failed
eth_getBlockByHash request inside web3j's
+ * blockFlowable) is subscribed again, instead of leaving a started consumer
that receives nothing.
+ */
+class Web3jConsumerResubscribeTest extends CamelTestSupport {
+
+ private final Web3j web3j = mock(Web3j.class);
+
+ @EndpointInject("mock:result")
+ private MockEndpoint result;
+
+ @EndpointInject("mock:error")
+ private MockEndpoint error;
+
+ Web3jConsumerResubscribeTest() {
+ // the mocked client is stubbed before the context starts
+ testConfiguration().withUseAdviceWith(true);
+ }
+
+ @Override
+ protected void bindToRegistry(Registry registry) {
+ registry.bind("mockWeb3j", web3j);
+ }
+
+ private static EthBlock block(String number) {
+ EthBlock.Block block = new EthBlock.Block();
+ block.setNumber(number);
+ EthBlock ethBlock = new EthBlock();
+ ethBlock.setResult(block);
+ return ethBlock;
+ }
+
+ private void startRoute(String operation) throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ onException(Exception.class).handled(true).to("mock:error");
+
from("web3j://http://127.0.0.1:8545?web3j=#mockWeb3j&operation="; + operation)
+ .routeId("web3j").autoStartup(false)
+ .to("mock:result");
+ }
+ });
+ context.start();
+ ((Web3jConsumer)
context.getRoute("web3j").getConsumer()).setResubscribeDelay(10);
+ context.getRouteController().startRoute("web3j");
+ }
+
+ @Test
+ void blockSubscriptionEndedByAFailedRequestIsSubscribedAgain() throws
Exception {
+ when(web3j.blockFlowable(anyBoolean())).thenReturn(
+ Flowable.concat(Flowable.just(block("0x1")),
Flowable.error(new IOException("Connection reset"))),
+ Flowable.concat(Flowable.just(block("0x2")),
Flowable.never()));
+ result.expectedMessageCount(2);
+ error.expectedMessageCount(1);
+
+ startRoute(Web3jConstants.BLOCK_OBSERVABLE);
+
+ MockEndpoint.assertIsSatisfied(context, 5, TimeUnit.SECONDS);
+ assertEquals("0x2",
result.getReceivedExchanges().get(1).getIn().getBody(EthBlock.Block.class).getNumberRaw());
+ }
+
+ @Test
+ void failuresWhileSubscribingAgainAreRoutedOnce() throws Exception {
+ // the node is unreachable for three attempts, then it answers again
+ IOException down = new IOException("Connection refused");
+ when(web3j.blockFlowable(anyBoolean())).thenReturn(
+ Flowable.error(down), Flowable.error(down),
Flowable.error(down),
+ Flowable.concat(Flowable.just(block("0x1")),
Flowable.never()));
+ result.expectedMessageCount(1);
+ error.expectedMessageCount(1);
+
+ startRoute(Web3jConstants.BLOCK_OBSERVABLE);
+
+ MockEndpoint.assertIsSatisfied(context, 5, TimeUnit.SECONDS);
+ verify(web3j, times(4)).blockFlowable(anyBoolean());
+ }
+
+ @Test
+ void replayOfARangeIsNotSubscribedAgain() throws Exception {
+ // a replay of a block range would route the blocks of the range again
+ when(web3j.replayPastBlocksFlowable(any(), any(), anyBoolean()))
+ .thenReturn(Flowable.concat(Flowable.just(block("0x1")),
Flowable.error(new IOException("Connection reset"))));
+ result.expectedMessageCount(1);
+ error.expectedMessageCount(1);
+
+ startRoute(Web3jConstants.REPLAY_BLOCKS_OBSERVABLE);
+
+ MockEndpoint.assertIsSatisfied(context, 5, TimeUnit.SECONDS);
+ verify(web3j, after(200).times(1)).replayPastBlocksFlowable(any(),
any(), anyBoolean());
+ }
+
+ @Test
+ void logFilterFromAGivenBlockIsNotSubscribedAgain() throws Exception {
+ // a new filter from the same block would route the logs from that
block again
+ when(web3j.ethLogFlowable(any())).thenReturn(Flowable.error(new
IOException("Connection refused")));
+ error.expectedMessageCount(1);
+
+ startRoute(Web3jConstants.ETH_LOG_OBSERVABLE + "&fromBlock=16");
+
+ MockEndpoint.assertIsSatisfied(context, 5, TimeUnit.SECONDS);
+ verify(web3j, after(200).times(1)).ethLogFlowable(any());
+ }
+
+ @Test
+ void logFilterFromTheLatestBlockIsSubscribedAgain() throws Exception {
+ Log log = new Log();
+ log.setData("0x01");
+ when(web3j.ethLogFlowable(any())).thenReturn(
+ Flowable.error(new IOException("Connection refused")),
+ Flowable.concat(Flowable.just(log), Flowable.never()));
+ result.expectedMessageCount(1);
+ error.expectedMessageCount(1);
+
+ startRoute(Web3jConstants.ETH_LOG_OBSERVABLE);
+
+ MockEndpoint.assertIsSatisfied(context, 5, TimeUnit.SECONDS);
+ verify(web3j, times(2)).ethLogFlowable(any());
+ }
+
+ @Test
+ void subscriptionMadeWhileTheConsumerStopsIsDisposed() throws Exception {
+ // the second subscription is made while the route stops: the request
to the node is slow and is not
+ // interrupted by the shutdown of the executor, so it ends after
doStop disposed the previous subscription
+ CountDownLatch subscribing = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ CountDownLatch cancelled = new CountDownLatch(1);
+ Flowable<EthBlock> slow = Flowable.defer(() -> {
+ subscribing.countDown();
+ boolean released = false;
+ while (!released) {
+ try {
+ released = release.await(10, TimeUnit.SECONDS);
+ if (!released) {
+ throw new IllegalStateException("not released");
+ }
+ } catch (InterruptedException e) {
+ // ignore the interrupt of the executor shutdown, as a
blocking request may do
+ }
+ }
+ return Flowable.<EthBlock> never();
+ }).doOnCancel(cancelled::countDown);
+ when(web3j.blockFlowable(anyBoolean())).thenReturn(Flowable.error(new
IOException("Connection reset")), slow);
+ error.expectedMessageCount(1);
+
+ startRoute(Web3jConstants.BLOCK_OBSERVABLE);
+ assertTrue(subscribing.await(5, TimeUnit.SECONDS));
+ context.getRouteController().stopRoute("web3j");
+ release.countDown();
+
+ assertTrue(cancelled.await(5, TimeUnit.SECONDS), "the subscription
made while stopping must be disposed");
+ MockEndpoint.assertIsSatisfied(context);
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 861d15d4719a..5d31b92772b7 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -990,6 +990,17 @@ new `compile` option of the endpoint (default `false`).
The module is now loaded
every producer, so stopping and starting a route does not read the module
again: remove the route and add it
again, or restart the application, to load a module that has changed.
+=== camel-web3j - the consumer subscribes again after an error
+
+When the subscription of a consumer ends with an error (for example a request
to the node that fails inside
+`BLOCK_OBSERVABLE` or `TRANSACTION_OBSERVABLE`, or a node that cannot be
reached when the route starts), the consumer
+now subscribes again every 15 seconds until it succeeds. The error is routed
as before, and logged at `WARN`; the
+failed attempts that follow are logged at `DEBUG`. Previously the consumer
received nothing more while the route
+stayed started. A new subscription starts at the latest block, so the blocks
between the error and the new
+subscription are not routed. `REPLAY_BLOCKS_OBSERVABLE`,
`REPLAY_TRANSACTIONS_OBSERVABLE` and an
+`ETH_LOG_OBSERVABLE` with `fromBlock` set to a block number are not subscribed
again, as that would route the same
+blocks or logs again.
+
=== camel-vertx-http - a header with several values is sent as several headers
A message header whose value is a `java.util.Collection` or an object array
(as the HTTP consumers map a request
