Message store option is not available for priority-queue channel [INT-1870]

[spring-operator]

Andrey Belyaev opened INT-1870 and commented

The "message-store" attribute is not defined for priority-queue type in xsd schema. So we could not just say like this:

<int:channel id="queue">
    <int:priority-queue message-store="messageStore" />
</int:channel>

At the first glance it seems that message store implementation for priority queue could be exactly the same as for regular one.

---

This issue is a sub-task of #6168

Issue Links:

• 6016 Consider supporting FIFO semantics on the Priority Channel for tie breaks

• 7308 Implement MongoDB PriorityCapableChannelMessageStore

• 7258 Implement JDBC PriorityCapableChannelMessageStore

• 7309 Implement GemFire PriorityCapableChannelMessageStore

• 7295 Add List-Backed Redis Message Group Store for QueueChannel Use

1 votes, 2 watchers

[spring-operator]

Oleg Zhurakousky commented

I am pushing it to 2.1. I don't think its a bug, since storing messages with priority requires resorting of the Queue as each message arrives and if such queue is persistent the whole Queue would have to be persisted every time

[spring-operator]

Oleg Zhurakousky commented

Implemented, pending review https://github.com/SpringSource/spring-integration/pull/60

[spring-operator]

Oleg Zhurakousky commented

Moving it to RC1 as it requires a bit more work that would involve other components to make it usable.

[spring-operator]

Oleg Zhurakousky commented

I left a comment in the PR but i feel its worth duplicating it here as well so here it is:

I moved this issue to RC1 and will close this PR. Will open the new PR for RC1 Summary of the issue as it currently implemented:

• First the current approach exposing a loop hole by allowing PriorityChannel to be created with non-priority BlockingQueue.
• The current approach will load all Messages into Memory for comparison on the first call to receive() - not efficient and most likely will lead to crashes in the real environment with a lot of Messages. Here is the summary of the proposed approach: Today MessageGroupQueue is an implementation of the BlockingQueue with additional capabilities
• it provides persistence capabilities if MessageStore is specified
• it provides comparison capabilities if Comparator is specified (thus it inherits PriorityBlockingQueue features) This means that MGQ can exhibit the behavior of the LinkedBlockingQueue (used by QueueChannel) as well as PriorityBlockingQueue with the changes that are there now (e.g., comparator)

For consistency we can always rely on MGQ for both channels The enhancements needed are:

• Provide configurable blocking behavior for MGQ (currently if capacity is exceeded it throws an exception) so we can eliminate blocking code from PriorityChannel and simply delegate to 'super'
• Enhance MessageStore implementation to store Messages in ordered way so upon receive() calls we can simply poll for the right message instead of loading them all and sorting them

Also, the loading of Messages into Memory is the problem for both QueueChannel and PriorityChannel if used with MessageStore since MessageGroupQueue is going to be used if specifies the 'message-store'. Basically we need to make sure that all our MS implementation support FIFO semantics as well as accept the Comparator which might not be that simple or even doable.

[spring-operator]

Oleg Zhurakousky commented

Actually the current fix is valid. The only downside is that just like a regular QueueChannel Messages will be loaded into Memory upon the first receive call. So once #6168 is resolved this could be resolved as well. I'll issue a different PR for that since the previous one was for M2

[spring-operator]

Mark Fisher commented

That would work assuming the priority logic stays within the MessageGroupQueue in such a way that its getNext() (or whatever it may be called) returns the next prioritized Message. If we want to avoid priority logic in MGQ however and keep it encapsulated within PriorityChannel, there might not be any clean way to implement the one-at-a-time logic. So I guess we have to go with MGQ having support for a Comparator, most likely a setter on AbstractMessageGroupQueue.

Also, the constructor should be changed to take a MessageStore and Comparator (so that it can create a MGQ with that Comparator) instead of a java Queue instances (so the Priority-sorting behavior is maintained).

[spring-operator]

Oleg Zhurakousky commented

As I described earlierI am al for keeping it in MGQ.

Yes I agree as discussed that the constructor in the PC should change regardless of the approach

[spring-operator]

Mark Fisher commented

The main challenge will likely be how to translate a java Comparator instance into some logic that will obviously need to occur remotely (since the main goal is to not load Messages into memory). Any idea how you would accomplish that?

[spring-operator]

Oleg Zhurakousky commented

Moving it to 2.2.

[spring-operator]

Oleg Zhurakousky commented

We can do it only if relied on the default priority header. In this case 'message-store' and 'comparator' should be mutually exclusive.

[spring-operator]

Artem Bilan commented

PR: https://github.com/spring-projects/spring-integration/pull/926

[spring-operator]

Gary Russell commented

PR: https://github.com/spring-projects/spring-integration/pull/1084

Adds infrastructure and a RedisChannelPriorityMessageStore.

TODO: JDBC implementation (TBD whether to subclass JdbcChannelMessageStore or update it).

[spring-operator]

Artem Bilan commented

The <priority-queue message-store=""/> infrastructure and RedisChannelPriorityMessageStore is merged (#7295)