allthingssecurity opened a new pull request, #27857: URL: https://github.com/apache/camel/pull/27857
# Description [CAMEL-25654](https://issues.apache.org/jira/browse/CAMEL-25654) The pg-replication-slot consumer is meant to connect again when its connection to PostgreSQL is lost, but on main `PgReplicationSlotConsumer.poll()` did so only when the failed read had a `SocketException` as cause (lines 117-125 on main). What went wrong (pgjdbc 42.7.13, read with javap: `QueryExecutorImpl.readFromCopy`, `V3PGReplicationStream.isClosed`, `PgConnection.checkClosed`): 1. When the server is restarted or shut down it closes the connection. The driver reports the failed read as "Database connection failed when reading from copy" (SQL state 08006) with the socket's `IOException` as cause, an `EOFException` when the server closed the connection. Before it throws, the driver ends the copy operation, so the stream reports itself closed from then on (`isClosed()` is `closeFlag || !copyDual.isActive()`). The consumer rethrew and kept the connection. The next poll's `getStream()`, which was outside the `try`, saw the closed stream and queried the slot (`isSlotActive`) on the dead connection: "An I/O error occurred while sending to the backend." (08006), after which the driver closes the connection. Every poll after that failed with "This connection has been closed." (08003). The route stayed started, logged a WARN per poll, and never consumed again, also after the server was back. A server that ends the replication (the read returns null and the stream i s closed) and then closes the connection takes the same `getStream()` path. 2. With a `SocketException` cause, `connect()` closed the old connection before opening the new one. When the server was still down, the closed connection stayed in the field, and every later poll failed in `getStream()` with "This connection has been closed." (08003), so no further reconnect was tried. Changes, all in `PgReplicationSlotConsumer`: - A connection is lost when the SQL state is of class `08` (connection exception) or `57P` (operator intervention, e.g. admin shutdown), when the cause is a `SocketException` (as before), or when the connection is closed. The consumer then closes and forgets the connection, its stream and the payload that was read but not processed successfully, and rethrows. For `57P` this only notices the loss one poll earlier: the server ends the session, and the next call would fail with 08006 anyway. - `poll()` connects first when there is no connection, and `getStream()` is now inside the same `try`, so a failure of the slot query or of the stream start is handled like a failed read. - A connect that fails after the connection was opened (for example while checking or creating the slot) closes that connection again, so the next poll starts from the beginning, including the slot creation of `autoCreateSlot`. - The INFO line "Connection to PostgreSQL server has been lost ..." is logged only when a connection was actually dropped; a poll that fails to connect while the server is down logs it at DEBUG (the WARN of the failed poll is logged as on main). - Other errors keep the connection, as before. Delivery: the fix relies on the replication 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 message was processed successfully, so a message that was read but not flushed is sent again by the server on the new connection. Delivery stays at-least-once, as the component documentation says ("the same message might come more than once"). Dropping the unflushed payload on a lost connection avoids routing it once from the old connection's buffer and then once more from the server. What users can notice: after a server restart the consumer consumes again once the server is back. While the server is down there is one connect attempt per poll; each failed poll logs the WARN that main logs already, and the poll delay and the backoff options of the scheduled poll consumer apply. No option, default or message format changes, so there is no upgrade guide entry. Tests: - New `PgReplicationSlotConsumerReconnectTest` (5 tests). The JDBC objects are Mockito mocks that fail the way the driver does: a read after the server went away throws 08006 with the socket's exception as cause and closes the stream (a failed status update too); when the server ended the replication the read returns null and the stream is closed; a query on a connection whose server went away throws 08006 "An I/O error occurred while sending to the backend." and closes the connection; a closed connection throws 08003 "This connection has been closed." and a closed stream 08003 "This replication stream has been closed."; the slot sends again what was not flushed. The route uses `startScheduler=false` and the test calls `poll()` itself, so there is no timing. - The server stops (`EOFException` cause), two polls fail, the server starts again and a message is published: it is consumed within five polls. - The same with a `SocketException` cause, so that the first reconnect runs while the server is down. - The server ends the replication and closes the connection, so the next `getStream()` fails with an I/O error: the consumer connects again and consumes. - The first delivery of a message fails; while it is processed the server goes away and the status update fails. After the server is back the message is delivered exactly once (`[first, second]`). - An error that is not a connection loss (SQL state 42704) does not open a second connection. This one passes on main too; it guards against the fix reconnecting on every error. - Without the change (main's `PgReplicationSlotConsumer`, same test class) the first four fail, for example: `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., ...]`; with the `SocketException` cause `poll failures [Connection to localhost:5432 refused., This connection has been closed., ...]`. - Each part of the change is needed: without the reset of the payload on a lost connection the exactly-once test fails (`expected: <[first, second]> but was: <[first, second, second]>`); with `getStream()` moved back out of the `try`, the ended-replication test and the exactly-once test fail. - With the change, camel-pg-replication-slot tests pass: 7 run, 0 failures, 0 skipped. What is not changed: - `doStart` still fails when the server cannot be reached at start. - The status-update task of a message that is being processed keeps the stream it was created for, as before. - The `57P` branch has no test of its own; it only notices the loss one poll earlier than the 08006 that follows. What was not run: a real PostgreSQL server (the module's integration test needs a container), so which message a server restart produces first (an ErrorResponse 57P01, the end of the replication, or the end of the socket) was not observed; each of them ends in an 08 state on the next call, which the change handles. The driver behaviour the mocks follow is read from the driver's bytecode. Notes: - How it was found: a sweep of the reconnect code of the components, then a small TLA+ model of the consumer's connection field. For main it reaches a lost connection that no poll replaces (for both causes); with the change the consumer is connected again in the end. The model is only a sketch; the tests are the evidence. # Target - [x] I checked that the commit is targeting the correct branch (Camel 4 uses the `main` branch) # Tracking - [ ] If this is a large change, bug fix, or code improvement, I checked there is a [JIRA issue](https://issues.apache.org/jira/browse/CAMEL) filed for the change (usually before you start working on it). (The ticket is drafted and not filed yet; CAMEL-25654 is a placeholder until it is.) # Apache Camel coding standards and style - [x] I checked that each commit in the pull request has a meaningful subject line and body. - [ ] I have run `mvn clean install -DskipTests` locally from root folder and I have committed all auto-generated changes. (I built and tested camel-pg-replication-slot, including the formatter and import-sort plugins. I did not run the full root build.) # AI-assisted contributions - [x] If this PR includes AI-generated code, commits have proper co-authorship attribution (e.g., `Co-authored-by` trailers) and the PR description identifies the AI tool used. This PR was prepared with Claude Code (Claude Opus 5.5). The commit carries a `Co-Authored-By` trailer. _Claude Code on behalf of allthingssecurity_ 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
