Give CorrelatingMessageHandler a way to retry messages after a time-out expires. [INT-1068]

[spring-operator]

Jeroen van Erp opened INT-1068 and commented

Currently the CorrelatingMessageHandler only seems to have a way to force messages to pass the Barrier if the time-out has expired, or discard them. It would be nice to have a way to retry them after a given amount of time. In our usecase we have messages coming in, which are delayed because another message with the same correlation id is being processed. These messages are then kept back to await the other message to finish the processing chain. In this way we prevent messages from going through our system in the wrong order. However it may take a long time before another message with the same correlation id arrives at our processor, so all held back messages stay in the message store a long time and could/should have been processed earlier.

In essence we're using the CorrelatingMessageHandler as a Barrier or Latch, where a message with a specific correlation id can pass, unless another message with the same correlation id is in the chain.

Instead of having only an option of forcing or discarding messages, we should also have a way of gracefully retrying and delaying the messages if they couldn't be retried yet.

---

Affects: 2.0 M3

Issue Links:

• Add CorrelatingMessageBarrier [INT-1174] #5182 Add CorrelatingMessageBarrier
  ("depends on")

[spring-operator]

Dave Syer commented

Things have changed quite a bit in the CMH, so it might be that this feature is already addressable. I'd like to understand it a bit better before I could be sure though. It is no longer mandatory to have a timeout on the message groups (but you can set one up with a MessageGroupStoreReaper or just by calling MessageGroupStore.expireMessageGroups(correlationId)) for instance.

When you say "a specific correlation id can pass, unless another message with the same correlation id is in the chain," that sounds like the default behaviour. So what do you want to happen if the other message is being held up? The default behaviour is just to wait for it and release the group when it arrives (as long as the group can be identified through a sequence). You can get the waiting messages released, or discarded when the message group store expires the group according to the old "sendPartialResultOnTimeout" setting (which I am thinking should be renamed "sendPartialResultOnExpiry").

So what is missing? You use the word "retry" a couple of times, but I'm not sure I understand what needs to be retried and when.

[spring-operator]

Jeroen van Erp commented

Let's see if I can explain it...

T0: Message1 with correlation-id 1 reaches CMH and can be processed (no other messages with that correlation-id present in chain
T1: Message2 with correlation-id 1 reaches CMH and cannot be processed as Message1 is currently being processed
T2: Message1 finished processing, and correlation-id 1 becomes unblocked.
TX: Message3 with correlation-id 1 reaches CMH, and now Message2 and Message3 will process sequentially

Between T2 and TX a lot of time can pass, and it is not certain that TX will ever occur. It is certain however that T2 occurs, so we know that eventually correlation-id 1 becomes unblocked. So we need a mechanism that can retry Message2 after a delay period to check whether it can already pass the Barrier. The current timeout behaviour isn't good enough, as it either forces Message2 through (which is not allowed as Message1 might still be processed) or it discards Message2 (not wanted, because Message2 contains valuable information).

Does this make it a bit more clear? If not, please say so :-)

[spring-operator]

Dave Syer commented

That is very useful thank you. My reading is that you need three features that CMH doesn't support:

1. release a single message from a group
2. re-handle the group "normally" without a new message to trigger it
3. never discard messages even if a sequence is not defined

It think it is tough to support #1 with the existing ReleaseStrategy interface (since it only can make a decision for the whole group, not individual messages). It might be possible but only with some pretty opaque manipulation of the message group store.

I did have a prototype at one point where ReleaseStrategy had a method like this:

Collection<Message<?>> release(MessageGroup group);

I think maybe that is more what you need. Then you could simply write a release strategy that returned at most one message.

Which then leaves #2 and #3 (normal handling when no message has arrived and never discarding). CMH supports "abnormal" handling with no new message through an expiry callback it registers with the message store, but again that is not what is needed here because you never want to discard. Neither requirement seems to fit well in the existing design.

My conclusion is that this is an interesting use case, but it should probably be implemented as a separate MessageHandler, since CMH is already quite useful and doesn't do what you need. According to Iwein the use case is a Message Barrier in EIP terms - and now that MessageBarrier is not a reserved term in Spring Integration (it was removed from 2.0) maybe that's what we should call it?

If it turns out you can implement this with small changes to CMH (e.g. with custom strategies) I can easily change my opinion. I am worried that on the contrary it is not feasible in the current scope and it will lead to more features being added to the message store. (Therefore we don't have time to do it in 2.0.M4 which is due tomorrow.)

[spring-operator]

Dave Syer commented

Actually, having re-read the requirement and stared at the CMH some more, it seems that we already support "one at a time per correlation key". In fact in some ways it is the only behaviour that is supported with a DirectChannel downstream from the CMH, at least per application. Subsequent messages will be held up at the monitor lock in CMH.handleMessageInternal(), and they can only be processed one at a time because they are only received one at a time. So all you need to support this barrier (in a single node application) is a ReleaseStrategy that always releases, and a stream of messages that do not have a sequence. In fact you can already see this in action in CorrelatingMessageHandlerIntegrationTests: just use a PassThroughMessageGroupProcessor and a MessageCountReleaseStrategy and you are in business.

If your use case is different (e.g. downstream is not a DirectChannel, or the application is distributed), then you must have a different definition of a message "being processed", or "being in the chain" which is not encapsulated anywhere in Spring Integration yet anyway. What do you think?

[spring-operator]

Jeroen van Erp commented

Hi Dave,

Your first point was indeed implemented as such, that wasn't the hard part.
The main focus point of this issue was number 2 and 3.
What I did in the 2.0.0M3 version of CMH was add the following to the forceComplete method:

if (forcePartialResultOnTimeout) {
...
} else if (delayPartialResultOnTimeout) {
    if (logger.isInfoEnabled()) {
        logger.info("Delaying message for key [" + key + "]");
    }
    // Create a new DelayedKey with a new timeout, so as to prevent the current Correlation Key from ending
    // up on the head of the queue again immediately. Need to return true because we finished what we wanted to do.
    keysInBuffer.offer(new DelayedKey(key, timeout));
    return true;
}

And by setting the forcePartialResultOnTimeout to 'false' and the delayPartialResultOnTimeout to 'true', this proved to be working.

[spring-operator]

Dave Syer commented

M4 is very different to M3. I suggest we revisit this when you've had a chance to try it.

[spring-operator]

Iwein Fuld commented

I've written CorrelatingMessageBarrier to fulfill this requirement and it is working for the particular client where we found the use case. It only needs to make it's way into SI.

[spring-operator]

Dave Syer commented

In that case this duplicates #5182 (so resolving)