Add locking strategy to AbstractMessageGroupStore [INT-1117]

[spring-operator]

Dave Syer opened INT-1117 and commented

Global locks might be an interesting feature in general, but the most pressing need is for message store use cases. It only makes sense if the store uses back-end persistence, so it seems like there is an obvious tie in to the store implementation itself,

---

Issue Links:

• 6182 Consider improving message polling in JdbcMessageStore

• 7321 Implement RedisLockRegistry

• 6730 Improve MessageGroupStoreReaper Algorithms

• 5182 Add CorrelatingMessageBarrier

• 6131 Consider refactoring the scope of synchronized blocks in AbstractCorrelatingMessageHandler

  ("supersedes")

1 votes, 4 watchers

[spring-operator]

Dave Syer commented

I know this is my issue, but I can't think of a good use case that can't be addressed with transaction isolation. Over to Mark to remind us...

[spring-operator]

Mark Fisher commented

IIRC one of the discussions that led to this issue was related to ensuring FIFO ordering when using a MessageStore-backed MessageChannel. Another concern would be for Aggregators in a distributed environment, where checking for completion and processing the aggregated result itself should only happen on one node at a time. If both of these can be handled in other ways (within the particular implementations of MessageStore), then perhaps we can close this issue.

[spring-operator]

Oleg Zhurakousky commented

I think it would be interesting to get more details on how we can get a hold of RegionQueue described in https://wiki.springsource.com/display/GMFI/Gemfire+API+TODOs (if MessageStore is Gemfire). But even if underlying implementation is not Gemfire IMHO we should still be thinking in terms of global Queue impelmentation to ensure FIFO semantics.

[spring-operator]

Dave Syer commented

@Oleg: MessageGroupQueue is a global queue implementation provided the store is global. #5237 and the demo I set up on githib for Brian proved that this works out of the box with some configuration of the transactions using JDBC. Gemfire is another story, but should work the same way provided they have transaction isolation (and if not it probably should go to Spring Gemfire, not wait for an implementation here).

@Mark: I'm not sure I ever understood the FIFO use case in any detail (it seems futile to try and impose an order globally on a concurrent system). The discussion in #5237 led me to add some new features to JDBC, but it wasn't really FIFO: more like "fair distribution of business keys to single consumer".

[spring-operator]

Brian Dussault commented

The following link is how Oracle ESB provides a cluster wide locking mechanism for resequencing: http://download.oracle.com/docs/cd/E12524_01/doc.1013/e12638/sequencer.htm

Another area for "locking" might be for things such as polling adapters. For example: It may not be desirable to have multiple jdbc adapters polling at the same time but rather a single jdbc adapter instance would execute the polling query. In this context, the cluster wide "lock" means single threaded on the jdbc adapter. If the node assigned to run the jdbc polling fails, another cluster node should take over.

[spring-operator]

Dave Syer commented

We already support global resequencing and locking in polling adapters (which we proved with that demo in #5237). As long as the database supports transaction isolation it should work with no more code from Spring Integration.

There still might be a need for explicit locks, but I don't want to make more work here when I think it should be a Spring Data feature anyway.

[spring-operator]

Mark Fisher commented

Dave, I'm reassigning to you in case you have changed your mind. Otherwise, I'd say we should perhaps resolve as "deferred" and/or remove the target version for this so that we don't lose track. I don't want to add it to 2.0.0 unless we have a very clear idea of the purpose and benefits.

[spring-operator]

Anderson Balduino commented

Dave, we will use an aggregator based on Infinispan. This issue is addressing the possibility of multiple concurrent access aggregator in a distributed environment?

[spring-operator]

Dave Syer commented

Correct. And the current plan is to do nothing, since the only implementation of a global store we provide out of the box in Spring Integration has built in locks at the server level (JDBC with transaction isolation=READ_COMMITTED or better). I imagine you would be writing your own message store for Infinispan, and therefore would need to implement the lock in the MessageGroupStore mutator methods. Ideally you would be able to inspect the current transaction definition to lock only if the poller wants you to, but that's probably a nice to have.

[spring-operator]

Mark Fisher commented

Will revisit this in the 2.1 M1 timeframe.

[spring-operator]

Mark Fisher commented

Perhaps AbstractMessageGroupStore should provide the abstract methods for locking as well. That way each implementation would use its own persistence mechanism (or simple Lock objects in the case of SimpleMessageStore).

[spring-operator]

Oleg Zhurakousky commented

That is essentially what I was experimenting with yesterday

[spring-operator]

Mark Fisher commented

SimpleMessageStore can rely on in-process locks (java.util.concurrent), but the distribute-able implementations should rely on strategies available within their respective environments (e.g. a region lock in GemFire).

[spring-operator]

Oleg Zhurakousky commented

This is rather serious issue and requires investigation across all available MessageStore implementations. Moving it to 2.2.m1

[spring-operator]

Artem Bilan commented

When you have time, please, look into my commit: https://github.com/artembilan/spring-integration/commit/d5af04c225bf33c782e90ae2570c5fcab9aa7b78 It's about locks on FOR UPDATE in the JdbcMessageStore and I would call it "Competing Consumers in Cluster"

[spring-operator]

Oleg Zhurakousky commented

Technically this one has been resolved already by introducing LockRegistry which produces java.util.concurrent.locks.Lock What we probably need is raise separate JIRAs to provide implementations based on different mechanisms (e.g., JDBC, Gemfire, Redis etc)