"Unnecessary" Tasks when using PostgresSubscribableChannel

[martinitus]

In our application we use the PostgresSubscribableChannel for a small queue that is backed by our PostgreSQL database. It works really well, especially with the new changes w.r.t. transaction safety and rollback in case the message handling caused an exception.

What we have noticed, is that in the case of many concurrent writes into the queue, the number of queued tasks in the executor keeps going up. This is obvious, as each write causes a NOTIFY on the database side which is translated to a new task on the Spring side.

However: The current implementation of the tasks triggered by NOTIFY contain a loop to process queued up messages as long as new messages can be read from the database. So what happens in the case of many concurrent writes to the queue, is that some task A processes multiple messages, while other tasks remain in the queue (e.g. the executor only allows for N concurrent tasks). The list of queued tasks keeps growing until eventually there is more capacity in the executor (e.g. task A has finished processing all messages).

The queued up tasks will then all run their query to check for items in the database queue, but won't find any, as they already have been processed by - let's say - task A.

Admittedly, it's not a huge deal, and using concurrent executors would also prevent the tasks to queue up, but I wonder whether there is some other mechanism that we could use to prevent having the tasks build up.
One thing that came to mind was simply have a 1:1 relationship between tasks and dequeued items, i.e. remove the do ... while loop that processes multiple items.

Another hack we came up with was customizing the executor to simply drop tasks if there are already N tasks running (as messages will be processed by these running tasks and the NOTIFY was sort of unnecessary in the first place):

 PostgresChannelMessageTableSubscriber(...).also {
      var numRunningNotifications: Int = 0
      it.setDispatcherExecutor(Executor { r ->
          executor.execute(Runnable { ->
              if (numRunningNotifications < 5) {
                  numRunningNotifications = numRunningNotifications + 1
                  try {
                      log.info("running notify")
                      r.run()
                  } finally {
                      numRunningNotifications = numRunningNotifications - 1
                  }
              } else {
                  log.info("skipping notify")
              }
          })
})

[artembilan]

I think something like ThreadPoolTaskExecutor with a decent pool size and DiscardPolicy (see setRejectedExecutionHandler()) is way to go if you'd like to have a concurrency for messages processing, but don't want to have void tasks to be piled up.
It technically would be similar to what you have so far manually.

Perhaps that's what we can do by default in the framework any way: accept some concurrency and create respective ThreadPoolTaskExecutor instead of SimpleAsyncTaskExecutor by default.

Just let me know if that works for your and we will proceed with issue and the fix!

[martinitus]

Thank you!
I just had a look into the ThreadPoolTaskExecutor and DiscardPolicy - seems exactly what I need to eliminate our "custom hack" :-)

I mainly opened the discussion to point out the current behavior and to double check that I didn't miss any configuration option to prevent it.

Maybe it helps some other developer who stumbles across.

I think the current setup is quite flexible as it is, and users sort of have to dig deeper into the internals anyway to understand whats going on.
That said, I think some concurrency argument could make sense. If there is need, one could still override the executor by setting explicitly.