Release message group with groupTimeout in multiple instances context

[Ngoctrong102]

Hi,

I using Aggregator with RedisMessageStore, RedisLockRegistry and partial timeout release.

As I know, Aggregator will create timer for groupTimeout when it get a new message on pod, so there will be one timer on each pod.
Thus, I think perhaps message group will be released more than once.
I wonder if RedisLockRegistry is used to prevent this case. And when the second thread get lock, will it try to release a deleted group?

Thanks.

[artembilan]

The logic there is like this:

1. We schedule a group for expiration after timeout:

			if (startTime != null) {
				Object groupId = messageGroup.getGroupId();
				long timestamp = messageGroup.getTimestamp();
				long lastModified = messageGroup.getLastModified();
				ScheduledFuture<?> scheduledFuture =
						getTaskScheduler()
								.schedule(() -> {
									try {
										processForceRelease(groupId, timestamp, lastModified);
									}
									catch (MessageDeliveryException ex) {
										logger.warn(ex, () ->
												"The MessageGroup [" + groupId +
														"] is rescheduled by the reason of: ");
										scheduleGroupToForceComplete(groupId);
									}
								}, startTime.toInstant());

				this.logger.debug(() -> "Schedule MessageGroup [ " + messageGroup + "] to 'forceComplete'.");
				this.expireGroupScheduledFutures.put(UUIDConverter.getUUID(groupId), scheduledFuture);
			}

2. When that happen we do:

	private void processForceRelease(Object groupId, long timestamp, long lastModified) {
		MessageGroup messageGroup = this.messageStore.getMessageGroup(groupId);
		if (messageGroup.getTimestamp() == timestamp && messageGroup.getLastModified() == lastModified) {
			this.forceReleaseProcessor.processMessageGroup(messageGroup);
		}
	}

So, the group is released only if it has not been changed in between.
Since you say that all your instances got a message, and even if the group is scheduled for expiration, only one of them is going to release this group. Even if all those messages are stored into DB.
And that will be the one which got the last message. Since this is only a place where those timestamps in the store are the same.

After that we have additional logic in the forceComplete() which does some additional duplicate prevention:

			/*
				 * If the group argument is not already complete,
				 * re-fetch it because it might have changed while we were waiting on
				 * its lock. If the last modified timestamp changed, defer the completion
				 * because the selection condition may have changed such that the group
				 * would no longer be eligible. If the timestamp changed, it's a completely new
				 * group and should not be reaped on this cycle.
				 *
				 * If the group argument is already complete, do not re-fetch.
				 * Note: not all message stores provide a direct reference to its internal
				 * group so the initial 'isComplete()` will only return true for those stores if
				 * the group was already complete at the time of its selection as a candidate.
				 *
				 * If the group is marked complete, only consider it
				 * for reaping if it's empty (and both timestamps are unaltered).
				 */
				if (!group.isComplete()) {
					groupNow = this.messageStore.getMessageGroup(correlationKey);
				}

If you observer some other behavior, then, please, share with us some simple project where we ca reproduce the problem.
Otherwise I treat this as Works as Designed.

Thanks

[Ngoctrong102]

It was good until now, thank for your supporting.