From c42441da317b370fe7680b9f4a43a4c963bf1bfd Mon Sep 17 00:00:00 2001 From: Johannes Edmeier Date: Mon, 1 Apr 2024 19:05:56 +0200 Subject: [PATCH] GH-9061: renew connection in PostgresChannelMessageTableSubscriber Fixes: #9061 `PostgresChannelMessageTableSubscriber` never renews the connection. This causes problems on DB failover. With this change the connection is renewed when notifications are not received for a certain time. (cherry picked from commit 642278dad1bea9d424166a2e3150991f58104525) # Conflicts: # spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriber.java # spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriberTests.java --- ...PostgresChannelMessageTableSubscriber.java | 44 +++++++++++----- ...resChannelMessageTableSubscriberTests.java | 51 ++++++++++++++++--- 2 files changed, 77 insertions(+), 18 deletions(-) diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriber.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriber.java index c68f166008..fb622ebf62 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriber.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriber.java @@ -1,5 +1,5 @@ /* - * Copyright 2022-2023 the original author or authors. + * Copyright 2022-2024 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -18,6 +18,7 @@ package org.springframework.integration.jdbc.channel; import java.sql.SQLException; import java.sql.Statement; +import java.time.Duration; import java.util.Map; import java.util.Set; import java.util.concurrent.CompletableFuture; @@ -60,6 +61,7 @@ import org.springframework.util.Assert; * @author Rafael Winterhalter * @author Artem Bilan * @author Igor Lovich + * @author Johannes Edmeier * * @since 6.0 */ @@ -83,6 +85,8 @@ public final class PostgresChannelMessageTableSubscriber implements SmartLifecyc @Nullable private volatile PgConnection connection; + private Duration notificationTimeout = Duration.ofSeconds(60); + /** * Create a new subscriber using the {@link JdbcChannelMessageStore#DEFAULT_TABLE_PREFIX}. * @param connectionSupplier The connection supplier for the targeted Postgres database. @@ -113,6 +117,19 @@ public final class PostgresChannelMessageTableSubscriber implements SmartLifecyc this.executor = executor; } + /** + * Set the timeout for the notification polling. + * If for the specified duration no notificiation are received the underlying connection is closed and re-established. + * Setting a value of {@code Duration.ZERO} will disable the timeout and wait forever. + * This might cause problems in DB failover scenarios. + * @param notificationTimeout the timeout for the notification polling. + * @since 6.1.8 + */ + public void setNotificationTimeout(Duration notificationTimeout) { + Assert.notNull(notificationTimeout, "'notificationTimeout' must not be null."); + this.notificationTimeout = notificationTimeout; + } + /** * Add a new subscription to this subscriber. * @param subscription The subscription to register. @@ -176,22 +193,25 @@ public final class PostgresChannelMessageTableSubscriber implements SmartLifecyc while (isActive()) { startingLatch.countDown(); - PGNotification[] notifications = conn.getNotifications(0); + PGNotification[] notifications = conn.getNotifications((int) this.notificationTimeout.toMillis()); // Unfortunately, there is no good way of interrupting a notification // poll but by closing its connection. if (!isActive()) { return; } - if (notifications != null) { - for (PGNotification notification : notifications) { - String parameter = notification.getParameter(); - Set subscriptions = this.subscriptionsMap.get(parameter); - if (subscriptions == null) { - continue; - } - for (Subscription subscription : subscriptions) { - subscription.notifyUpdate(); - } + if (notifications == null || notifications.length == 0) { + //We did not receive any notifications within the timeout period. + //We will close the connection and re-establish it. + break; + } + for (PGNotification notification : notifications) { + String parameter = notification.getParameter(); + Set subscriptions = this.subscriptionsMap.get(parameter); + if (subscriptions == null) { + continue; + } + for (Subscription subscription : subscriptions) { + subscription.notifyUpdate(); } } } diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriberTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriberTests.java index fb4348a167..1dda46b363 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriberTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriberTests.java @@ -17,6 +17,8 @@ package org.springframework.integration.jdbc.channel; import java.sql.DriverManager; +import java.sql.SQLException; +import java.time.Duration; import java.util.ArrayList; import java.util.List; import java.util.concurrent.CountDownLatch; @@ -61,6 +63,7 @@ import static org.assertj.core.api.Assertions.assertThat; * @author Rafael Winterhalter * @author Artem Bilan * @author Igor Lovich + * @author Johannes Edmeier * * @since 6.0 */ @@ -116,15 +119,14 @@ public class PostgresChannelMessageTableSubscriberTests implements PostgresConta private String groupId; + private ConnectionSupplier connectionSupplier; + @BeforeEach void setUp(TestInfo testInfo) { // Not initiated as a bean to allow for registrations prior and post the life cycle - this.postgresChannelMessageTableSubscriber = - new PostgresChannelMessageTableSubscriber(() -> - DriverManager.getConnection(POSTGRES_CONTAINER.getJdbcUrl(), - POSTGRES_CONTAINER.getUsername(), - POSTGRES_CONTAINER.getPassword()) - .unwrap(PgConnection.class)); + this.connectionSupplier = new ConnectionSupplier(); + this.postgresChannelMessageTableSubscriber = new PostgresChannelMessageTableSubscriber(connectionSupplier); + this.postgresChannelMessageTableSubscriber.setNotificationTimeout(Duration.ofSeconds(5)); this.taskExecutor = new ThreadPoolTaskExecutor(); @@ -277,6 +279,26 @@ public class PostgresChannelMessageTableSubscriberTests implements PostgresConta assertThat(payloads).containsExactly("1"); } + @Test + public void testRenewConnection() throws Exception { + CountDownLatch latch = new CountDownLatch(2); + List payloads = new ArrayList<>(); + CountDownLatch connectionLatch = new CountDownLatch(2); + connectionSupplier.onGetConnection = connectionLatch::countDown; + postgresChannelMessageTableSubscriber.start(); + postgresSubscribableChannel.subscribe(message -> { + payloads.add(message.getPayload()); + latch.countDown(); + }); + + assertThat(connectionLatch.await(10, TimeUnit.SECONDS)).isTrue(); + + messageStore.addMessageToGroup(groupId, new GenericMessage<>("1")); + messageStore.addMessageToGroup(groupId, new GenericMessage<>("2")); + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(payloads).containsExactlyInAnyOrder("1", "2"); + } + @Configuration @EnableIntegration public static class Config { @@ -316,4 +338,21 @@ public class PostgresChannelMessageTableSubscriberTests implements PostgresConta } + private static class ConnectionSupplier implements PgConnectionSupplier { + + Runnable onGetConnection; + + @Override + public PgConnection get() throws SQLException { + var conn = DriverManager.getConnection(POSTGRES_CONTAINER.getJdbcUrl(), + POSTGRES_CONTAINER.getUsername(), + POSTGRES_CONTAINER.getPassword()) + .unwrap(PgConnection.class); + if (this.onGetConnection != null) { + this.onGetConnection.run(); + } + return conn; + } + + } }