[
https://issues.apache.org/jira/browse/CAMEL-25730?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18128459#comment-18128459
]
Claus Ibsen commented on CAMEL-25730:
-------------------------------------
Merged to main in https://github.com/apache/camel/pull/28037 (commit
4b504fd443b4), fix version 4.24.0.
_Claude Code on behalf of Claus Ibsen_
> 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
> Priority: Major
> Fix For: 4.24.0
>
>
> {{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)
