[jira] [Resolved] (CAMEL-25545) camel-pgevent - the consumer reconnects only after the first lost connection; after the next one it stays disconnected without an error while the route keeps running

["Claus Ibsen (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/CAMEL-25545?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Claus Ibsen resolved CAMEL-25545.
---------------------------------
    Resolution: Fixed

Fixed by https://github.com/apache/camel/pull/27693 (merged to main as 35053b9b02e5, 4.24.0).

_Claude Code on behalf of Claus Ibsen_

> camel-pgevent - the consumer reconnects only after the first lost connection; after the next one it stays disconnected without an error while the route keeps running
> ---------------------------------------------------------------------------------------------------------------------------------------------------------------------
>
>                 Key: CAMEL-25545
>                 URL: https://issues.apache.org/jira/browse/CAMEL-25545
>             Project: Camel
>          Issue Type: Bug
>          Components: camel-pgevent
>            Reporter: shashank
>            Assignee: shashank
>            Priority: Major
>             Fix For: 4.24.0
>
>
> {{PgEventConsumer.doInit}} (line 72 at main 9432ba16d026) creates one {{BackgroundTask}} for reconnecting, and {{PgEventListener.reconnect()}} (line 103), which the driver calls through {{closed()}} when the connection is lost, runs that same task every time:
> {code:java}
> if (!reconnectTask.isRunning()) {
>     reconnectTask.run(endpoint.getCamelContext(), () -> {
>         ...
>         initConnection();   // new connection, LISTEN, notification listener
>         ...
>         return true;
>     });
> }
> {code}
> A {{BackgroundTask}} completes only once. Its latch ({{new CountDownLatch(1)}}) is counted down by the first successful attempt; after that {{run()}} schedules an attempt with the initial delay of one second, {{waitForTaskCompletion}} returns at once because the latch is already 0 and cancels the attempt, and {{runTaskWrapper}} would do nothing anyway ("the task is done and every further run is a no-op"). The other users of the task API (camel-master, camel-sjms, camel-kafka) build a new task for each use.
> So the consumer reconnects after the first connection loss, and after the second one {{reconnect()}} returns without connecting: the consumer no longer listens to the channel, no notification is routed any more, nothing is logged after the "Connection to PostgreSQL lost unexpected. Re-connecting..." warning, and the route still shows as started. Only a restart of the route (which runs {{doInit}} again) repairs it. A database that restarts or fails over twice during the lifetime of the route is enough.
> h3. Reproduction
> New {{PgEventConsumerReconnectTest}} (no database: a mocked {{DataSource}} hands out mocked {{PGConnection}}s through the {{datasource}} option; the test calls {{closed()}} on the listener as the driver does, and the call returns when the reconnect is done). On main, twice:
> {noformat}
> PgEventConsumerReconnectTest.reconnectsAfterEveryConnectionLoss: The consumer must reconnect after the second connection loss ==> expected: <3> but was: <2>
> {noformat}
> h3. Proposed fix
> {{reconnect()}} creates a new {{BackgroundTask}} (same pool, name and budget as before) for each reconnect, and keeps it in the (now volatile) field for the "already running" check. With the fix the test also verifies that the third connection executed {{LISTEN testchannel}} and got the notification listener. No upgrade guide note: nothing changes for a consumer that has not lost its connection twice. camel-pgevent with the fix: 1 unit test (the new one; the module had none), 0 failures; the integration tests need a PostgreSQL container (Docker) and were not run. Test dependency added: mockito-junit-jupiter.
> Not changed here: a consumer that is stopped while a reconnect is waiting leaves the thread that called {{closed()}} waiting on the task (the pool is shut down, the task is not cancelled); that needs its own look.
> Found by reading how the task API is used across components, with a small TLA+ model of the reconnect (connection lost, attempt, return of {{run()}}, the latch of the task): "when reconnect() has returned the consumer listens to the channel" holds for one connection loss and is violated for two (6 states); with a new task per reconnect it holds for two and three losses. Confirmed with the real consumer.
> Affected: main, camel-4.22.x, camel-4.18.x, camel-4.14.x (same reconnect code and the same single-use latch in {{BackgroundTask}}, GitHub contents API).
> Duplicate check (2026-10-11): JIRA text "pgevent" since 2021 (10: CAMEL-22146 added the reconnect, CAMEL-22949 and CAMEL-23646 changed the task callback, none about a second connection loss); GitHub open pull requests for camel-pgevent: none. No open pull request changes {{PgEventConsumer}}.
> _Filed with Claude Code on behalf of allthingssecurity._



--
This message was sent by Atlassian Jira
(v8.20.10#820010)