shashank created CAMEL-25654:
--------------------------------
Summary: 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
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)
