fix: prevent the notifyUpdate() during subscribe to die

A single notifyUpdate() is executed during the initial subscribe,
in case there are messages waiting in the queue.
Because the try-catch is outside the polling loop this can lead
to a dying polling loop at startup leaving the messages unprocessed.

Author: joshiste

spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/channel/PostgresSubscribableChannel.java

@@ -16,9 +16,6 @@
 
 package org.springframework.integration.jdbc.channel;
 
-import java.util.Optional;
-import java.util.concurrent.Executor;
-
 import org.springframework.core.log.LogAccessor;
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.integration.channel.AbstractSubscribableChannel;
@@ -32,6 +29,9 @@
 import org.springframework.transaction.support.TransactionTemplate;
 import org.springframework.util.Assert;
 
+import java.util.Optional;
+import java.util.concurrent.Executor;
+
 /**
  * An {@link AbstractSubscribableChannel} for receiving push notifications for
  * messages send to a group id of a {@link JdbcChannelMessageStore}. Receiving
@@ -160,14 +160,15 @@ protected boolean doSend(Message<?> message, long timeout) {
 	@Override
 	public void notifyUpdate() {
 		this.executor.execute(() -> {
-			try {
-				Optional<Message<?>> dispatchedMessage;
-				do {
-					dispatchedMessage = askForMessage();
-				} while (dispatchedMessage.isPresent());
-			}
-			catch (Exception ex) {
-				LOGGER.error(ex, "Exception during message dispatch");
+			while (true) {
+				try {
+					if (askForMessage().isEmpty()) {
+						break;
+					}
+				}
+				catch (Exception ex) {
+					LOGGER.error(ex, "Exception during message dispatch");
+				}
 			}
 		});
 	}

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriberTests.java

@@ -233,6 +233,31 @@ void testRetryOnErrorDuringDispatch(boolean transactionsEnabled) throws Interrup
 		assertThat(payloads).containsExactly("1");
 	}
 
+	@Test
+	void testPollLoopDiesOnNotifyInSubscribe() throws InterruptedException {
+		CountDownLatch latch = new CountDownLatch(2);
+		List<Object> payloads = new ArrayList<>();
+
+		messageStore.addMessageToGroup(groupId, new GenericMessage<>("error"));
+		messageStore.addMessageToGroup(groupId, new GenericMessage<>("1"));
+
+		postgresChannelMessageTableSubscriber.start();
+		postgresSubscribableChannel.subscribe(message -> {
+			try {
+				if ("error".equals(message.getPayload())) {
+					throw new RuntimeException("An error has occurred");
+				}
+				payloads.add(message.getPayload());
+			}
+			finally {
+				latch.countDown();
+			}
+		});
+
+		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(payloads).containsExactly("1");
+	}
+
 	@Configuration
 	@EnableIntegration
 	public static class Config {