shashank created CAMEL-25730:
--------------------------------
Summary: camel-web3j - the consumer never subscribes again after
its subscription ends with an error, so one failed request to the node stops
BLOCK_OBSERVABLE and TRANSACTION_OBSERVABLE while the route stays started
Key: CAMEL-25730
URL: https://issues.apache.org/jira/browse/CAMEL-25730
Project: Camel
Issue Type: Bug
Components: camel-web3j
Reporter: shashank
Assignee: shashank
{{Web3jConsumer.doStart}} (main 55ed20f14e90, lines 55-128) subscribes once to
a web3j {{Flowable}}, with {{processError}} as the error callback. An RxJava
{{Flowable}} that signals an error is over, and nothing subscribes again:
{{processError}} (lines 184-189) routes one exchange with the exception, and
from then on the consumer receives nothing while the route stays started.
In web3j 5.0.0 several of these flowables make a request to the node for each
item, inside a {{flatMap}}:
* {{blockFlowable}} = {{ethBlockHashFlowable(...).flatMap(hash ->
web3j.ethGetBlockByHash(hash, full).flowable())}}
({{JsonRpc2_0Rx.lambda$blockFlowable$6}});
* {{transactionFlowable}} = {{blockFlowable(true, ...).flatMapIterable(...)}};
* {{pendingTransactionFlowable}} =
{{ethPendingTransactionHashFlowable(...).flatMap(hash ->
ethGetTransactionByHash(hash).flowable())}}.
{{Request.flowable()}} is a {{RemoteCall}} over {{Request.send()}}, which
throws the {{IOException}} of a failed request. So with
{{operation=BLOCK_OBSERVABLE}}, {{TRANSACTION_OBSERVABLE}} or
{{PENDING_TRANSACTION_OBSERVABLE}}, a single failed request (a connection
reset, a node restart, a timeout of the HTTP client) ends the consumer for
good. The block hash filter itself survives such failures (web3j's {{Filter}}
catches the {{IOException}} of its polls), so the stop depends on which request
fails. For every operation, a node that cannot be reached when the route starts
has the same effect.
The consumer is a long-running listener; a node that is briefly unavailable is
normal. After the one routed error the blocks and transactions are lost without
further notice, and only a route restart brings the consumer back.
h3. Reproduction
New {{Web3jConsumerResubscribeTest}} (mocked {{Web3j}}, no node):
* {{blockSubscriptionEndedByAFailedRequestIsSubscribedAgain}}:
{{blockFlowable}} emits a block, then fails with {{IOException("Connection
reset")}}; a second subscription emits a block. On main the second block is
never received ({{mock://result Received message count. Expected: <2> but was:
<1>}}).
* {{failuresWhileSubscribingAgainAreRoutedOnce}}: three failed subscriptions,
then one that emits a block ({{Expected: <1> but was: <0>}}).
* {{logFilterFromTheLatestBlockIsSubscribedAgain}}: {{ETH_LOG_OBSERVABLE}}
whose filter cannot be installed at start ({{Expected: <1> but was: <0>}}).
* {{subscriptionMadeWhileTheConsumerStopsIsDisposed}}: the route is stopped
while the consumer subscribes again (a slow request to the node); the new
subscription must be disposed.
The committed tests set a 10 ms retry delay through a package-private setter
that main does not have. Run against unmodified main, a copy of the tests
without that setter call and with a 25 s wait fails the same three ({{Expected:
<2> but was: <1>}}, {{Expected: <1> but was: <0>}}, {{Expected: <1> but was:
<0>}}); with the fix the same copy passes with the default 15 s delay. Two more
tests pass on main and with the fix; they guard what must not be subscribed
again: a replay of a block range, and a log filter from a given block number.
The stop test needs the fix (it waits for the second subscription); removing
only the dispose after a subscription made while stopping makes it fail.
h3. Proposed fix
After a live subscription ended with an error, the consumer subscribes again
every 15 seconds until it succeeds, on a scheduled executor created in
{{doStart}} and shut down in {{doStop}}. The first error is routed as before
and logged once at WARN without a stack trace; the attempts and their failures
are logged at DEBUG until an item is received again, which is logged once at
INFO. A subscription that is made while the consumer stops (the executor
shutdown does not wait for a running attempt) is disposed.
{{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. A new
subscription starts at the latest block, so the blocks between the error and
the new subscription are not routed; this is stated in the upgrade note (4.24
guide).
With the fix the camel-web3j tests pass: 138 run, 0 failures, 0 errors, 67
skipped (the integration tests that need a node).
Why the resubscription does not route anything twice: the block hash, pending
transaction and log flowables use web3j's {{Filter}}, which catches the
{{IOException}} of its polls and logs it, so they end with an error only when
the filter cannot be installed (or its first {{eth_getFilterLogs}} fails),
before anything was routed; a new filter only returns what comes after it is
installed. {{blockFlowable}} and the transaction flowables end on a failed
per-item request, and the new subscription starts with a new block filter.
Not covered: no option for the delay (a constant, with a package-private setter
for the tests); the blocks of the gap are not fetched; a log filter from
{{safe}} or {{finalized}} is not subscribed again (only {{latest}} and
{{pending}} are treated as live); when the node forgot a filter and web3j's own
reinstall of it fails, web3j stops polling without an error, which the consumer
cannot see. Checked against web3j 5.0.0 only (CAMEL-23567, the upgrade to
5.0.1+, is open).
Affected: main (tested); camel-4.23.x, camel-4.22.x and camel-4.18.x have the
same consumer (read, not run).
Duplicate check (2026-10-17): JIRA text "web3j consumer", "web3j subscription
error", "web3j resubscribe", "web3j reconnect": only CAMEL-24748 (metadata);
GitHub pull requests "web3j": dependency bumps, header renames and metadata
fixes only.
_Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
