[
https://issues.apache.org/jira/browse/CAMEL-25654?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Work on CAMEL-25654 started by shashank.
----------------------------------------
> camel-pg-replication-slot - consumer never connects again after the
> PostgreSQL server closed the connection (restart, shutdown), so the route
> fails every poll for as long as it runs
> -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-25654
> URL: https://issues.apache.org/jira/browse/CAMEL-25654
> Project: Camel
> Issue Type: Bug
> Components: camel-pg-replicationslot
> Reporter: shashank
> Assignee: shashank
> Priority: Major
>
> The consumer is meant to connect again when its connection to PostgreSQL is
> lost: {{PgReplicationSlotConsumer.poll()}} (lines at main 1de5ec5135f8)
> catches the {{SQLException}} of the stream read (117-125) and calls
> {{connect()}} - but only when {{e.getCause() instanceof SocketException}}.
> Everything else is rethrown and the connection is kept. {{getStream()}} (92)
> is called before that {{try}}, so nothing it throws leads to a reconnect.
> h3. 1. A server restart or shutdown is never followed by a reconnect
> When the server is restarted or shut down, it closes the connection of the
> walsender. The PostgreSQL JDBC driver (42.7.13, read with javap) behaves like
> this:
> * {{QueryExecutorImpl.readFromCopy}} reports the failed read as
> {{PSQLException}} "Database connection failed when reading from copy", SQL
> state {{08006}}, with the {{IOException}} of the socket as cause - an
> {{EOFException}} when the server closed the connection, not a
> {{SocketException}}. Before throwing it releases the copy operation, so
> {{V3PGReplicationStream.isClosed()}} ({{closeFlag || !copyDual.isActive()}})
> is true from then on. The consumer rethrows and keeps the connection.
> * The next poll's {{getStream()}} sees the closed stream and runs the slot
> query ({{isSlotActive}}) on the dead connection:
> {{QueryExecutorImpl.execute}} fails with "An I/O error occurred while sending
> to the backend." ({{08006}}) and aborts the connection. This is outside the
> {{try}}, so it is just rethrown.
> * Every later poll fails in {{getStream()}} with "This connection has been
> closed." ({{08003}}, {{PgConnection.checkClosed}}).
> The route stays started, the scheduled poll consumer logs a WARN for every
> poll (every 500 ms with the default {{delay}}), and nothing is consumed
> again, also long after the server is back. Only a restart of the route helps.
> A server that ends the replication (the read returns null and the stream is
> closed) and then closes the connection takes the same {{getStream()}} path.
> h3. 2. A reconnect that fails once is never tried again
> When the cause is a {{SocketException}} (connection reset), {{connect()}}
> (252-264) closes the old connection first and then opens a new one. If the
> server is still down - which is normal right after a connection reset -
> {{newDbConnection()}} throws, and the field keeps the closed connection and
> its closed stream. Every later poll fails in {{getStream()}} with "This
> connection has been closed." ({{08003}}, no {{SocketException}}), so again no
> further reconnect.
> How common: any restart of the PostgreSQL server (minor version upgrade,
> configuration change that needs a restart, failover, maintenance,
> {{pg_terminate_backend}} of the walsender) with default options. A consumer
> of a replication slot typically runs for a long time, so it will see such
> restarts. It is not silent (a WARN with stack trace per poll), but it never
> recovers.
> h3. Reproduction
> New {{PgReplicationSlotConsumerReconnectTest}}: the JDBC objects are Mockito
> mocks that fail the way the driver fails (a read after the server went away:
> {{08006}} with the socket's exception as cause, and the stream is closed; a
> query on such a connection: {{08006}} "An I/O error occurred while sending to
> the backend." and the connection is closed; a closed connection: {{08003}}
> "This connection has been closed."; the slot sends again what was not
> flushed). The route uses {{startScheduler=false}} and the test calls
> {{poll()}} itself, so nothing depends on timing. On main:
> * server stopped (read fails with an {{EOFException}} cause), polled twice,
> started again, a new message published, polled up to five times: {{the
> consumer did not consume again after the PostgreSQL server was back; received
> [first], poll failures [Database connection failed when reading from copy, An
> I/O error occurred while sending to the backend., This connection has been
> closed., ...]}};
> * the same with a {{SocketException}} cause, so the first reconnect runs
> while the server is still down: {{received [first], poll failures [Connection
> to localhost:5432 refused., This connection has been closed., ...]}};
> * the server ends the replication and closes the connection: {{received
> [first], poll failures [An I/O error occurred while sending to the backend.,
> This connection has been closed., ...]}};
> * the first delivery of a message fails while the server goes away:
> {{received [first], poll failures [An I/O error occurred while sending to the
> backend., This connection has been closed., ...]}}.
> h3. Proposed fix
> * A connection is lost when the SQL state is of class {{08}} (connection
> exception) or {{57P}} (operator intervention, e.g. {{57P01}} admin shutdown;
> this only notices the loss one poll earlier), when the cause is a
> {{SocketException}} (as before), or when the connection is closed. Then the
> consumer closes the connection (ignoring errors), forgets it, its stream and
> the payload that was read but not processed successfully, and rethrows.
> * {{poll()}} connects first when there is no connection, and {{getStream()}}
> is inside the same {{try}}. A connect that fails after the connection was
> opened closes it again, so the next poll starts from the beginning. The poll
> delay and the backoff options of the scheduled poll consumer apply as before.
> * The INFO line about the lost connection is logged only when a connection
> was dropped, not for every failed connect while the server is down.
> * Other errors keep the connection, as before (a test pins that: an error
> with SQL state {{42704}} does not open a second connection).
> Delivery: the fix relies on the slot resuming at its {{confirmed_flush_lsn}}.
> The stream is started without a start position and the consumer sets the
> flushed LSN only after a successful processing, so the server sends a dropped
> (unflushed) payload again on the new connection. Delivery stays
> at-least-once, as the component documentation says ("the same message might
> come more than once"); dropping the payload avoids routing it once from the
> old connection and once more from the server. A test pins it: a message whose
> first delivery failed while the connection was lost is delivered exactly once
> after the reconnect.
> No option, default or output changes, so no upgrade guide entry. While the
> server is down there is one connect attempt per poll, with the WARN line of
> the failed poll that main logs already.
> camel-pg-replication-slot tests with the fix: 7 run, 0 failures, 0 skipped.
> Not run: a real PostgreSQL server (the module's integration test needs a
> container); which message a restart produces first (ErrorResponse {{57P01}},
> end of the replication, or end of the socket) was not observed.
> How it was found: a sweep of the reconnect code of the components, then a
> small TLA+ model of the consumer's connection field (server up / down,
> connection ok / broken / closed / none, cause of the failed read). For main
> it reaches a lost connection that no poll replaces, for both causes, and a
> run that never reconnects; with the fix the properties hold for two outages
> (126 states). The model is a sketch; the tests are the evidence.
> Affected: main (1de5ec5135f8; tested). camel-4.23.x, camel-4.22.x,
> camel-4.18.x and camel-4.14.x have the same {{SocketException}} check in
> {{poll()}} and the same {{connect()}} (read, not run).
> Duplicate check (2026-10-14): JIRA component camel-pg-replicationslot (only
> CAMEL-15614, automate the integration test), text "replication slot
> reconnect" (nothing); GitHub pull requests "PgReplicationSlotConsumer",
> "pg-replication-slot" (nothing related).
> _Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
