GH-8582:Transactional support in PostgresSubscribableChannel #8587

igorlovich · 2023-03-28T07:34:29Z

As discussed in Transactional support in PostgresSubscribableChannel

pivotal-cla · 2023-03-28T07:34:34Z

@igorlovich Please sign the Contributor License Agreement!

Click here to manually synchronize the status of this Pull Request.

See the FAQ for frequently asked questions.

pivotal-cla · 2023-03-28T07:35:30Z

@igorlovich Thank you for signing the Contributor License Agreement!

igorlovich · 2023-03-28T14:56:43Z

The failing test has nothing to do with the MR

artembilan · 2023-03-28T14:23:29Z

...java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriber.java

+	private void notifyAll(Set<Subscription> subscriptions) {
+		subscriptions.forEach(it -> {
+			try {
+				this.executor.submit(it::notifyUpdate);


Why cannot we move this executor logic into a channel implementation?
Imaging we have several channels subscribed to this publisher, so it is really that channel responsibility to handle a notification its own way.

We can; my thinking was that since the Subscription interface is public, we can potentially have poorly behaving Subscriptions blocking the notification thread.

I don't think this is already our responsibility.
We can warn, though, in that Subscription Javadocs.

From here I wonder if that would be really better to have this feature implemented against this library: https://impossibl.github.io/pgjdbc-ng/

I have moved the executor into the channel.

artembilan · 2023-03-28T14:57:44Z

...java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriber.java

+	 *
+	 * @param pool The pool to shut down
+	 */
+	private void shutdownAndAwaitTermination(@Nullable ExecutorService pool) {


If we come back to a single task in this class as it was before, then we don't need this logic any more.

Yes, this logic would go into the channel

Probably no. The default SimpleAsyncTaskExecutor we use over there has no hook for shutdown.
It is really recommended to provide externally managed Executor, which in most cases is a org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor.

...java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriber.java

artembilan

This is very close to what we all can agree in the end.

artembilan · 2023-03-28T20:20:55Z

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

-		while ((message = this.jdbcChannelMessageStore.pollMessageFromGroup(this.groupId)) != null) {
-			this.dispatcher.dispatch(message);
-		}
+		this.executor.execute(() -> {


Can we look into the way to schedule a task per message?
Kinda perform an execute for single poll, if it was not null, then schedule the next one.

If there is a FIFO requirements, that's where this task executor can be customized externally.

By injecting a AsyncTaskExecutor and not an Executor we can get a CompletableFuture with the result. However getting the result (to determine whether to continue the loop) implies blocking, and is therefore equivalent to doing the loop as a single task, no matter what kind of an executor is passed in.

I can see 2 approaches (there may be more ) to achieve what you're proposing

Have an AtomicReference outside the loop that is set inside the task to the result of the poll. The exit condition for the while loop is the reference != null. Disadvantage of this approach is that in the case of a multi-threaded executor, depending on the order of tasks finishing, we can get useless iterations through the loop if the null from a later submitted, but earlier finishing task is overridden.

Ignore what is happening inside the task. The condition to exit from the while is this.jdbcChannelMessageStore.messageGroupSize(groupId) == 0 . There would need to be a sleep(X ms) to avoid hammering the message store. Disadvantage is that the call to notifyUpdate will take an X ms latency penalty.

OK. My suggestion is something like:

public void notifyUpdate() { this.executor.execute(this::pollAndSendMessage); } void pollAndSendMessage() { Message<?> message = this.jdbcChannelMessageStore.pollMessageFromGroup(this.groupId)); if (message != null) { // Process message this.executor.execute(this::pollAndSendMessage); } }

Do you see anything wrong with such an async stacking?

Does this not equally imply sequential message processing, irrespective of the Executor , since each task only submits the next one after it finishes?

Yeah... I see now. This is indeed the same what you have just explained with a CompletableFuture approach.
We could swap an new task submitting with the message processing. This way we can try to pull a new message in a new thread while we process the current one.
But is it really a goal for us? We would just eliminate any FIFO from this channel logic.
It might be as an option though, or can be implemented downstream as a bridge().channel(new ExecutorChannel()) to not care about FIFO.
But that's fully different story.

I think it is time for me to step back and agree with your current solution with a single-threaded process for the whole loop.

I'm going to pull your PR locally for the final review if you don't have anything else to add.

Thank you for your patience!

No worries it is important to get it right (at least with our current understanding of "right"). I will address your feedback on the documentation

artembilan · 2023-03-28T20:22:25Z

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

+	}
+
+	/**
+	 * Sets the transaction manager to use for message processing. Each message will be processed in a


Imperative language in method Javadocs, please: https://github.com/spring-projects/spring-framework/wiki/Code-Style#javadoc-formatting

artembilan · 2023-03-28T20:23:22Z

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

+	/**
+	 * Sets the transaction manager to use for message processing. Each message will be processed in a
+	 * separate transaction
+	 * @param transactionManager The transaction manager to use


@since 6.0.5

artembilan · 2023-03-28T20:24:42Z

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

 	private final JdbcChannelMessageStore jdbcChannelMessageStore;

+	private TransactionTemplate transactionTemplate;


Please, move this non-final property to respective section of props.

artembilan · 2023-03-28T20:27:06Z

src/reference/asciidoc/jdbc.adoc

@@ -632,6 +632,14 @@ Such connection pools do normally expect that issued connections are closed with
 For this need of an exclusive connection, it is also recommended that a JVM only runs a single `PostgresChannelMessageTableSubscriber` which can be used to register any number of subscriptions.
 ====

+==== Transaction support


I don't think we need a dedicated section id for these paragraphs. It is fully OK to have this info withing that top-level postgresql-push.

artembilan · 2023-03-28T20:28:01Z

src/reference/asciidoc/jdbc.adoc

@@ -632,6 +632,14 @@ Such connection pools do normally expect that issued connections are closed with
 For this need of an exclusive connection, it is also recommended that a JVM only runs a single `PostgresChannelMessageTableSubscriber` which can be used to register any number of subscriptions.
 ====

+==== Transaction support
+Specifying a `PlatformTransactionManager` on `PostgresSubscribableChannel` will notify subscribers in a transaction.


We also need to mention that this feature is available starting with version 6.0.5

artembilan · 2023-03-28T20:28:33Z

src/reference/asciidoc/jdbc.adoc

+Transactional support is not active by default.
+
+==== Retries
+A retry policy can be controlled by providing a `RetryTemplate` to `PostgresSubscribableChannel`. By default, no retries are performed.


One sentence per line, please: https://asciidoctor.org/docs/asciidoc-recommended-practices/#one-sentence-per-line

FYI: I prefer to not have these kind of comments in the PR.
If you agree with me, just push the change.
Otherwise feel free to discuss.
In the end this done looks like a spam in my e-mail box 😄

…hannel

artembilan · 2023-03-29T16:38:33Z

Merged as e39449b after some minor clean up.
And cherry-picked to 6.0.x.

@igorlovich ,

thank you very much for contribution; looking forward for more!

igorlovich force-pushed the GH-8582 branch 2 times, most recently from 6ad6722 to 1fd9694 Compare March 28, 2023 08:05

igorlovich marked this pull request as ready for review March 28, 2023 08:49

artembilan requested changes Mar 28, 2023

View reviewed changes

igorlovich force-pushed the GH-8582 branch from 1fd9694 to f020321 Compare March 28, 2023 16:29

artembilan requested changes Mar 28, 2023

View reviewed changes

spring-projectsGH-8582:Transactional support in PostgresSubscribableC…

81f3398

…hannel

igorlovich force-pushed the GH-8582 branch from f020321 to 81f3398 Compare March 29, 2023 15:38

artembilan closed this Mar 29, 2023

		private final JdbcChannelMessageStore jdbcChannelMessageStore;

		private TransactionTemplate transactionTemplate;

GH-8582:Transactional support in PostgresSubscribableChannel #8587

GH-8582:Transactional support in PostgresSubscribableChannel #8587

Uh oh!

Conversation

igorlovich commented Mar 28, 2023

Uh oh!

pivotal-cla commented Mar 28, 2023

Uh oh!

pivotal-cla commented Mar 28, 2023

Uh oh!

igorlovich commented Mar 28, 2023

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

artembilan left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

igorlovich Mar 29, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

artembilan commented Mar 29, 2023

Uh oh!

Uh oh!

igorlovich Mar 29, 2023 •

edited

Loading