This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 35053b9b02e5 CAMEL-25545: camel-pgevent - Reconnect every time the
connection is lost (#27693)
35053b9b02e5 is described below
commit 35053b9b02e55a1b07452c5388d040a0c54e0c73
Author: allthingssecurity <[email protected]>
AuthorDate: Tue Oct 13 01:27:46 2026 +0530
CAMEL-25545: camel-pgevent - Reconnect every time the connection is lost
(#27693)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
components/camel-pgevent/pom.xml | 17 ++
.../camel/component/pgevent/PgEventConsumer.java | 101 ++++++++--
.../pgevent/PgEventConsumerReconnectTest.java | 207 +++++++++++++++++++++
3 files changed, 307 insertions(+), 18 deletions(-)
diff --git a/components/camel-pgevent/pom.xml b/components/camel-pgevent/pom.xml
index d6031e097a32..04682feb0603 100644
--- a/components/camel-pgevent/pom.xml
+++ b/components/camel-pgevent/pom.xml
@@ -54,6 +54,23 @@
<artifactId>camel-test-junit6</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-junit-jupiter</artifactId>
+ <version>${mockito-version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.assertj</groupId>
+ <artifactId>assertj-core</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <version>${awaitility-version}</version>
+ <scope>test</scope>
+ </dependency>
<!-- test infra -->
<dependency>
diff --git
a/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventConsumer.java
b/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventConsumer.java
index 421a47171d8e..61ae21a798fd 100644
---
a/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventConsumer.java
+++
b/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventConsumer.java
@@ -43,9 +43,11 @@ public class PgEventConsumer extends DefaultConsumer {
private final PgEventListener listener = new PgEventListener();
private final PgEventEndpoint endpoint;
- private PGConnection dbConnection;
+ // set from the thread that connects (start, or the reconnect pool) and
read from the thread that stops
+ private volatile PGConnection dbConnection;
private ScheduledExecutorService reconnectPool;
- private BackgroundTask reconnectTask;
+ // the task of the current (or last) reconnect, set from the thread that
reports the lost connection
+ private volatile BackgroundTask reconnectTask;
private ExecutorService workerPool;
private boolean shutdownWorkerPool;
@@ -69,7 +71,11 @@ public class PgEventConsumer extends DefaultConsumer {
// used for re-connecting to the database
reconnectPool =
getEndpoint().getCamelContext().getExecutorServiceManager()
.newSingleThreadScheduledExecutor(this, "PgEventReconnect");
- reconnectTask = Tasks.backgroundTask()
+ reconnectTask = null;
+ }
+
+ private BackgroundTask createReconnectTask() {
+ return Tasks.backgroundTask()
.withScheduledExecutor(reconnectPool)
.withBudget(Budgets.iterationTimeBudget()
.withInterval(Duration.ofMillis(endpoint.getReconnectDelay()))
@@ -89,6 +95,12 @@ public class PgEventConsumer extends DefaultConsumer {
@Override
protected void doStop() throws Exception {
super.doStop();
+ // a pending reconnect has the thread that reported the lost
connection (a thread of the driver) waiting in
+ // run(). No attempt runs once the pool is shut down, so nothing else
would release that thread.
+ BackgroundTask task = reconnectTask;
+ if (task != null) {
+ task.cancel(true);
+ }
listener.closeConnection();
getEndpoint().getCamelContext().getExecutorServiceManager().shutdown(reconnectPool);
if (shutdownWorkerPool && workerPool != null) {
@@ -101,16 +113,34 @@ public class PgEventConsumer extends DefaultConsumer {
public class PgEventListener implements PGNotificationListener {
public void reconnect() {
+ if (!isRunAllowed()) {
+ // stopping: the reconnect pool is about to be shut down and
would never run the task
+ return;
+ }
// only submit the task if not already running
- if (!reconnectTask.isRunning()) {
- reconnectTask.run(endpoint.getCamelContext(), () -> {
+ BackgroundTask current = reconnectTask;
+ if (current == null || !current.isRunning()) {
+ // a background task completes only once (running it again
does nothing), so every reconnect needs
+ // its own task
+ final BackgroundTask task = createReconnectTask();
+ reconnectTask = task;
+ if (!isRunAllowed()) {
+ // the stop began after the check above: it may have
looked for a task to cancel before this one
+ // was set
+ return;
+ }
+ task.run(endpoint.getCamelContext(), () -> {
if (isRunAllowed()) {
- LOG.debug("Connecting attempt #{}",
reconnectTask.iteration());
+ LOG.debug("Connecting attempt #{}", task.iteration());
try {
initConnection();
+ if (!isRunAllowed()) {
+ // the consumer was stopped while this attempt
was connecting
+ closeConnection();
+ }
} catch (Exception e) {
String message
- = "Failed to connect attempt #" +
reconnectTask.iteration() + " due to: " + e.getMessage();
+ = "Failed to connect attempt #" +
task.iteration() + " due to: " + e.getMessage();
getExceptionHandler().handleException(message, e);
// make the task runner aware of the exception
(will retry)
throw new TaskRunFailureException(message, e);
@@ -123,36 +153,71 @@ public class PgEventConsumer extends DefaultConsumer {
}
public void initConnection() throws Exception {
- dbConnection = endpoint.initJdbc();
String channel = endpoint.getChannel();
if (!channel.matches("[a-zA-Z_][a-zA-Z0-9_]*")) {
throw new IllegalArgumentException("Invalid channel name");
}
- String sql = String.format("LISTEN %s", channel);
- try (PreparedStatement statement =
dbConnection.prepareStatement(sql)) { // NOSONAR
- statement.execute();
+ // the connection that is replaced (a lost one, when reconnecting)
is closed first, so that a pooled
+ // data source gets it back
+ PGConnection previous = dbConnection;
+ dbConnection = null;
+ closeQuietly(previous, channel);
+
+ PGConnection connection = endpoint.initJdbc();
+ try {
+ String sql = String.format("LISTEN %s", channel);
+ try (PreparedStatement statement =
connection.prepareStatement(sql)) { // NOSONAR
+ statement.execute();
+ }
+ connection.addNotificationListener(channel, channel, listener);
+ } catch (Exception e) {
+ // not used, and the next attempt gets a new one
+ closeQuietly(connection, channel);
+ throw e;
}
- dbConnection.addNotificationListener(channel, channel, listener);
+ dbConnection = connection;
}
public void closeConnection() throws Exception {
- if (dbConnection != null) {
+ PGConnection connection = dbConnection;
+ dbConnection = null;
+ if (connection != null) {
+ String channel = endpoint.getChannel();
try {
- String channel = endpoint.getChannel();
if (!channel.matches("[a-zA-Z_][a-zA-Z0-9_]*")) {
throw new IllegalArgumentException("Invalid channel
name");
}
- dbConnection.removeNotificationListener(channel);
+ connection.removeNotificationListener(channel);
String sql = String.format("UNLISTEN %s", channel);
- try (PreparedStatement statement =
dbConnection.prepareStatement(sql)) { // NOSONAR
+ try (PreparedStatement statement =
connection.prepareStatement(sql)) { // NOSONAR
statement.execute();
}
- dbConnection.close();
} catch (Exception e) {
// ignore
+ } finally {
+ // also when UNLISTEN failed, as it does on a connection
that is lost
+ closeQuietly(connection, channel);
}
}
- dbConnection = null;
+ }
+
+ /**
+ * Closes a connection that is no longer used, without its listener
being told about it.
+ */
+ private void closeQuietly(PGConnection connection, String channel) {
+ if (connection == null) {
+ return;
+ }
+ try {
+ connection.removeNotificationListener(channel);
+ } catch (Exception e) {
+ // ignore
+ }
+ try {
+ connection.close();
+ } catch (Exception e) {
+ // ignore
+ }
}
@Override
diff --git
a/components/camel-pgevent/src/test/java/org/apache/camel/component/pgevent/PgEventConsumerReconnectTest.java
b/components/camel-pgevent/src/test/java/org/apache/camel/component/pgevent/PgEventConsumerReconnectTest.java
new file mode 100644
index 000000000000..a0e44bde013e
--- /dev/null
+++
b/components/camel-pgevent/src/test/java/org/apache/camel/component/pgevent/PgEventConsumerReconnectTest.java
@@ -0,0 +1,207 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.pgevent;
+
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import javax.sql.DataSource;
+
+import com.impossibl.postgres.api.jdbc.PGConnection;
+import org.apache.camel.BindToRegistry;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * The consumer must connect again and listen to the channel again every time
the connection to the database is lost,
+ * not only the first time. It gives back the connections it no longer uses,
and a stop releases the thread that waits
+ * for a reconnect.
+ */
+class PgEventConsumerReconnectTest extends CamelTestSupport {
+
+ // the connections the data source has handed out
+ private final List<PGConnection> connections = new
CopyOnWriteArrayList<>();
+
+ @BindToRegistry("pgDataSource")
+ private final DataSource dataSource = mock(DataSource.class);
+
+ private PgEventConsumer createConsumer() throws Exception {
+ PgEventEndpoint endpoint = context.getEndpoint(
+
"pgevent:///testdb/testchannel?datasource=#pgDataSource&reconnectDelay=100",
PgEventEndpoint.class);
+ return (PgEventConsumer) endpoint.createConsumer(exchange -> {
+ });
+ }
+
+ private PGConnection newConnection() throws SQLException {
+ PGConnection connection = mock(PGConnection.class);
+
when(connection.prepareStatement(anyString())).thenReturn(mock(PreparedStatement.class));
+ connections.add(connection);
+ return connection;
+ }
+
+ @Test
+ void reconnectsAfterEveryConnectionLoss() throws Exception {
+ when(dataSource.getConnection()).thenAnswer(invocation ->
newConnection());
+
+ PgEventConsumer consumer = createConsumer();
+ consumer.start();
+ try {
+ assertThat(connections).hasSize(1);
+
+ // the driver reports a lost connection; the call returns when the
consumer has connected again
+ consumer.getPgEventListener().closed();
+ assertThat(connections).as("connections after the first connection
loss").hasSize(2);
+
+ consumer.getPgEventListener().closed();
+ assertThat(connections).as("connections after the second
connection loss").hasSize(3);
+
+ // and it listens to the channel on the new connection
+ PGConnection last = connections.get(2);
+ verify(last).prepareStatement("LISTEN testchannel");
+ verify(last).addNotificationListener("testchannel", "testchannel",
consumer.getPgEventListener());
+ } finally {
+ consumer.stop();
+ }
+ }
+
+ @Test
+ void closesTheLostConnectionWhenItReconnects() throws Exception {
+ when(dataSource.getConnection()).thenAnswer(invocation ->
newConnection());
+
+ PgEventConsumer consumer = createConsumer();
+ consumer.start();
+ try {
+ consumer.getPgEventListener().closed();
+ assertThat(connections).hasSize(2);
+
+ // the lost connection is given back (to the pool, with a pooled
data source), the new one is in use
+ verify(connections.get(0)).close();
+ verify(connections.get(1), never()).close();
+ } finally {
+ consumer.stop();
+ }
+ verify(connections.get(1)).close();
+ }
+
+ @Test
+ void closesAConnectionWhoseListenFails() throws Exception {
+ when(dataSource.getConnection()).thenAnswer(invocation -> {
+ PGConnection connection = newConnection();
+ if (connections.size() == 2) {
+ // the first attempt to reconnect gets a connection, and
LISTEN fails on it
+ when(connection.prepareStatement(anyString())).thenThrow(new
SQLException("LISTEN failed"));
+ }
+ return connection;
+ });
+
+ PgEventConsumer consumer = createConsumer();
+ consumer.start();
+ try {
+ // returns when the second attempt has connected
+ consumer.getPgEventListener().closed();
+ assertThat(connections).hasSize(3);
+
+ PGConnection failed = connections.get(1);
+ verify(failed).close();
+ verify(failed, never()).addNotificationListener("testchannel",
"testchannel", consumer.getPgEventListener());
+ verify(connections.get(2), never()).close();
+ } finally {
+ consumer.stop();
+ }
+ }
+
+ @Test
+ void stopReleasesTheThreadThatWaitsForAReconnect() throws Exception {
+ CountDownLatch attempted = new CountDownLatch(1);
+ when(dataSource.getConnection()).thenAnswer(invocation -> {
+ if (connections.isEmpty()) {
+ return newConnection();
+ }
+ // the database stays away, so the reconnect stays pending
+ attempted.countDown();
+ throw new SQLException("Connection refused");
+ });
+
+ PgEventConsumer consumer = createConsumer();
+ consumer.start();
+ // as the thread of the driver that reports the lost connection: it
waits until the consumer has reconnected
+ Thread driver = new Thread(() ->
consumer.getPgEventListener().closed(), "driver-io");
+ driver.setDaemon(true);
+ try {
+ driver.start();
+ assertThat(attempted.await(20, TimeUnit.SECONDS)).as("a reconnect
was attempted").isTrue();
+ assertThat(driver.isAlive()).as("the reconnect is
pending").isTrue();
+ } finally {
+ consumer.stop();
+ }
+
+ driver.join(TimeUnit.SECONDS.toMillis(20));
+ assertThat(driver.isAlive()).as("the thread that reported the lost
connection still waits").isFalse();
+ }
+
+ @Test
+ void connectionLostWhileStoppingDoesNotStartAReconnect() throws Exception {
+ PgEventConsumer[] consumer = new PgEventConsumer[1];
+ // as the thread of the driver that reports the lost connection
+ Thread driver = new Thread(() ->
consumer[0].getPgEventListener().closed(), "driver-io");
+ driver.setDaemon(true);
+ CountDownLatch reported = new CountDownLatch(1);
+ PreparedStatement unlisten = mock(PreparedStatement.class);
+ when(unlisten.execute()).thenAnswer(invocation -> {
+ // the connection is lost while the consumer stops: the stop goes
on when the driver thread has returned,
+ // or waits for a reconnect
+ driver.start();
+ await().atMost(20, TimeUnit.SECONDS).until(() -> driver.getState()
== Thread.State.TERMINATED
+ || driver.getState() == Thread.State.WAITING ||
driver.getState() == Thread.State.TIMED_WAITING);
+ reported.countDown();
+ throw new SQLException("Connection lost");
+ });
+ when(dataSource.getConnection()).thenAnswer(invocation -> {
+ PGConnection connection = newConnection();
+ when(connection.prepareStatement("UNLISTEN
testchannel")).thenReturn(unlisten);
+ return connection;
+ });
+
+ consumer[0] = createConsumer();
+ consumer[0].start();
+ try {
+ consumer[0].stop();
+ assertThat(reported.getCount()).as("the lost connection was
reported during the stop").isZero();
+
+ // the pool that would run a reconnect is shut down, so a thread
that waits for one waits forever
+ driver.join(TimeUnit.SECONDS.toMillis(20));
+ assertThat(driver.isAlive()).as("the thread that reported the lost
connection still waits").isFalse();
+ assertThat(connections).as("no reconnect was
attempted").hasSize(1);
+ verify(connections.get(0)).close();
+ } finally {
+ // never leave the thread behind
+ driver.interrupt();
+ }
+ }
+}
