Fix PostgresSubscribableChannel.notifyUpdate()

When transaction is configured for the `PostgresSubscribableChannel.notifyUpdate()`
and it is rolled back, the next poll in that loop will return the same message.
Again and again if transaction is always rolled back.
This leads to the condition when we never leave this loop even if
we fully unsubscribed from this channel.

The issue has need spotted after introducing `SKIP LOCKED`
for `PostgresChannelMessageStoreQueryProvider` which leads to
the locked record in DB in the mentioned above transaction.

* Introduce `PostgresSubscribableChannel.hasHandlers` flag
to check in the `notifyUpdate()` before performing poll query in DB.

**Cherry-pick to `6.1.x` & `6.0.x`**
This commit is contained in:
Artem Bilan
2023-09-18 08:18:40 -04:00
committed by GitHub
parent df67c59f8e
commit 78367f2b73
2 changed files with 25 additions and 14 deletions

View File

@@ -67,6 +67,8 @@ public class PostgresSubscribableChannel extends AbstractSubscribableChannel
private Executor executor;
private volatile boolean hasHandlers;
/**
* Create a subscribable channel for a Postgres database.
* @param jdbcChannelMessageStore The message store to use for the relevant region.
@@ -128,6 +130,7 @@ public class PostgresSubscribableChannel extends AbstractSubscribableChannel
boolean subscribed = super.subscribe(handler);
if (this.dispatcher.getHandlerCount() == 1) {
this.messageTableSubscriber.subscribe(this);
this.hasHandlers = true;
notifyUpdate();
}
return subscribed;
@@ -138,6 +141,7 @@ public class PostgresSubscribableChannel extends AbstractSubscribableChannel
boolean unsubscribed = super.unsubscribe(handle);
if (this.dispatcher.getHandlerCount() == 0) {
this.messageTableSubscriber.unsubscribe(this);
this.hasHandlers = false;
}
return unsubscribed;
}
@@ -159,18 +163,7 @@ public class PostgresSubscribableChannel extends AbstractSubscribableChannel
try {
Optional<Message<?>> dispatchedMessage;
do {
if (this.transactionTemplate != null) {
dispatchedMessage =
this.retryTemplate.execute(context ->
this.transactionTemplate.execute(status ->
pollMessage()
.map(this::dispatch)));
}
else {
dispatchedMessage =
pollMessage()
.map(message -> this.retryTemplate.execute(context -> dispatch(message)));
}
dispatchedMessage = askForMessage();
} while (dispatchedMessage.isPresent());
}
catch (Exception ex) {
@@ -179,6 +172,20 @@ public class PostgresSubscribableChannel extends AbstractSubscribableChannel
});
}
private Optional<Message<?>> askForMessage() {
if (this.hasHandlers) {
if (this.transactionTemplate != null) {
return this.retryTemplate.execute(context ->
this.transactionTemplate.execute(status -> pollMessage().map(this::dispatch)));
}
else {
return pollMessage()
.map(message -> this.retryTemplate.execute(context -> dispatch(message)));
}
}
return Optional.empty();
}
private Optional<Message<?>> pollMessage() {
return Optional.ofNullable(this.jdbcChannelMessageStore.pollMessageFromGroup(this.groupId));
}

View File

@@ -45,6 +45,7 @@ import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.jdbc.datasource.init.DataSourceInitializer;
import org.springframework.jdbc.datasource.init.ResourceDatabasePopulator;
import org.springframework.jdbc.datasource.init.ScriptUtils;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.test.annotation.DirtiesContext;
@@ -172,14 +173,16 @@ public class PostgresChannelMessageTableSubscriberTests implements PostgresConta
postgresSubscribableChannel.setTransactionManager(transactionManager);
postgresChannelMessageTableSubscriber.start();
postgresSubscribableChannel.subscribe(message -> {
MessageHandler messageHandler =
message -> {
try {
throw new RuntimeException("An error has occurred");
}
finally {
latch.countDown();
}
});
};
postgresSubscribableChannel.subscribe(messageHandler);
messageStore.addMessageToGroup(groupId, new GenericMessage<>("1"));
messageStore.addMessageToGroup(groupId, new GenericMessage<>("2"));
@@ -188,6 +191,7 @@ public class PostgresChannelMessageTableSubscriberTests implements PostgresConta
// Stop subscriber to unlock records from TX for the next verification
postgresChannelMessageTableSubscriber.stop();
postgresSubscribableChannel.unsubscribe(messageHandler);
assertThat(messageStore.messageGroupSize(groupId)).isEqualTo(2);
assertThat(messageStore.pollMessageFromGroup(groupId).getPayload()).isEqualTo("1");