Aggregator - Orphaned Groups Don't Time Out After Restart

[garyrussell]

In what version(s) of Spring Integration are you seeing this issue?

All

Describe the bug

When using an external message group store, orphaned groups are not timed out.
https://stackoverflow.com/questions/62748832

Expected behavior

When there is a group timeout, and a persistent group store, the aggregator should schedule a timeout for all existing groups that were created by this aggregator. We don't currently have a mechanism to determine this; it could be a problem when the same store is used for different aggregators - we shouldn't schedule a timeout for groups we don't "own".

@SpringBootApplication
public class So62748832Application {

	public static void main(String[] args) throws InterruptedException {
		ConfigurableApplicationContext ctx = SpringApplication.run(So62748832Application.class, args);
		Thread.sleep(10000);
		ctx.close();
	}

	@Bean
	public IntegrationFlow flow(RedisConnectionFactory cf) {
		return f -> f.aggregate(agg -> agg.correlationExpression("'foo'")
				.messageStore(store(cf))
				.sendPartialResultOnExpiry(true)
				.groupTimeout(15000))
				.handle(System.out::println);
	}

	@Bean
	MessageGroupStore store(RedisConnectionFactory connectionFactory) {
		return new RedisMessageStore(connectionFactory);
	}

//	@Bean
	@DependsOn("flow")
	public ApplicationRunner runner(@Qualifier("flow.input") MessageChannel channel) {
		return args -> channel.send(new GenericMessage<>("foo"));
	}

}

Run it once with the runner bean uncommented. Run again with it commented - no timeout.

[artembilan]

Sounds like something similar what we have in DelayHandler with its reschedulePersistedMessages()...
But do we have some identificator in those groups which aggregator has created them?..

[garyrussell]

Right.

it could be a problem when the same store is used for different aggregators - we shouldn't schedule a timeout for groups we don't "own".

We either need add a qualifier to the groupId for each different aggregator (same across multiple instances of the same aggregator), or add the qualifier to the group metadata so we can find the groups that belong to this aggregator.

[toananthan]

Hi Gary, Hi Artem,

here I want to provide few more details, find below jdbcMessageStore configuration for 2 aggregators, I'm setting unique region for each aggregator.

To avoid conflict of message group as I'm using single message store for both aggregators, I have defined correlation key of each aggregator prefixed with region name that is being set for jdbcMessageStore.

String correlationKeyChk = "CHKAggregator" + applicationNumber;
String correlationKeySav = "SAVAggregator" + applicationNumber;

Q1) is it possible that framework itself prefix such string for each aggregator in the application?

@Configuration
public class MessageGroupStoreConfiguration {

    @Autowired
    private DataSource dataSource;

    @Bean(name="chkAggregatorMessageStore")
    public JdbcMessageStore jdbcMessageGroupStoreChk(){
        JdbcMessageStore jdbcMessageStore = new JdbcMessageStore(dataSource);
        jdbcMessageStore.setRegion("ChkAggregator");
        return jdbcMessageStore;
    }

    @Bean(name="savAggregatorMessageStore")
    public JdbcMessageStore jdbcMessageGroupStoreSav(){
        JdbcMessageStore jdbcMessageStore = new JdbcMessageStore(dataSource);
        jdbcMessageStore.setRegion("SavAggregator");
        return jdbcMessageStore;
    }
}

/*
    @Bean
    MessageGroupStoreReaper getMessageGroupStoreReaperSav(){
        MessageGroupStoreReaper messageGroupStoreReaper = new MessageGroupStoreReaper(savAggregatorMessageStore);
        messageGroupStoreReaper.setTimeout(150000L);
        messageGroupStoreReaper.setExpireOnDestroy(true);
        return messageGroupStoreReaper;
    }*/

My aggregator bean, I have another similar bean for another aggregator poddChkAggregatorService.

    @Bean("poddSavAggregatorService.handler")
    @EndpointId("poddSavAggregatorService")
    @ServiceActivator(inputChannel = "savAggregatorChannel.input")
    FactoryBean<MessageHandler> aggregatorFactoryBean( ) {
        AggregatorFactoryBean aggregator = new AggregatorFactoryBean();
        aggregator.setProcessorBean(poddSavEventAggregator);//your Aggregator object
        aggregator.setMethodName("aggregate");//your aggregator method
        aggregator.setCorrelationStrategy(getSavCorrelationStrategy());
        aggregator.setReleaseStrategy(getSavReleaseStrategy());
        aggregator.setMessageStore(savAggregatorMessageStore);
        aggregator.setSendPartialResultOnExpiry(false);
        aggregator.setExpireGroupsUponCompletion(true);
        aggregator.setExpireGroupsUponTimeout(true);
        aggregator.setMinimumTimeoutForEmptyGroups(TimeUnit.SECONDS.toMillis(180));
        aggregator.setGroupTimeoutExpression(new ValueExpression<>(TimeUnit.SECONDS.toMillis(poddSavGrouptimeout)));
        //aggregator.setGroupTimeoutExpression(new SpelExpressionParser().parseExpression("60000"));// @MessagingGateway defaultReplyTimeout=2000
        aggregator.setDiscardChannelName("expiryMsgSavAuditChannel.input");
        aggregator.setOutputChannelName("coreFlow.input");
        //aggregator.setTaskScheduler(getMessageGroupStoreReaperSav());//set reaper
        //aggregator.setLockRegistry();
        return aggregator;
    }

Q2. how to set a TaskSheduler?, do you have sample application for that? application deployed on multiple pools/instances, should I configure the scheduler to run on each instance or just one instance (if the the instance is down scheduler will stop running)?

Thanks
Anantha

[toananthan]

Message store snapshot:

MESSAGE_ID	REGION	CREATED_DATE	MESSAGE_BYTES
6b4b6fb0-5a8a-b24a-9e57-7860c23ea6ec	ChkAggregator	12-JUL-20 08.46.54.187000000 PM	(BLOB)
2b4b6fb0-5a8a-b24a-9e57-7860c23ea6eb	SavAggregator	12-JUL-20 07.46.54.187000000 PM	(BLOB)
5383922a-2ab3-5d81-b8cf-0282084b8c68	DEFAULT	03-JUL-20 10.25.28.711000000 AM	(BLOB)
a51e4c1e-1e3c-3731-82fe-50bacaf34fea	DEFAULT	03-JUL-20 10.39.16.711000000 AM	(BLOB)

GROUP_KEY	REGION	MARKED	COMPLETE	LAST_RELEASED_SEQUENCE	CREATED_DATE	UPDATED_DATE
6dee1a00-6f91-3a2c-a9a6-bb2242aa4156	ChkAggregator	0	0	0	12-JUL-20 08.46.54.181000000 PM	12-JUL-20 08.46.54.181000000 PM
9dee1a00-6f91-3a2c-a9a6-bb2242aa4151	SavAggregator	0	0	0	12-JUL-20 07.46.54.181000000 PM	12-JUL-20 07.46.54.181000000 PM
b326b506-2b2f-3e69-8468-10717534cb09	DEFAULT	0	0	0	03-JUL-20 10.25.28.705000000 AM	03-JUL-20 10.25.28.705000000 AM
a01a0380-ca3c-3142-8c26-a231f0e49a09	DEFAULT	0	0	0	03-JUL-20 10.39.16.704000000 AM	03-JUL-20 10.39.16.704000000 AM

GROUP_KEY	MESSAGE_ID	REGION
6dee1a00-6f91-3a2c-a9a6-bb2242aa4156	6b4b6fb0-5a8a-b24a-9e57-7860c23ea6ec	ChkAggregator
9dee1a00-6f91-3a2c-a9a6-bb2242aa4151	2b4b6fb0-5a8a-b24a-9e57-7860c23ea6eb	SavAggregator
a01a0380-ca3c-3142-8c26-a231f0e49a09	a51e4c1e-1e3c-3731-82fe-50bacaf34fea	DEFAULT
b326b506-2b2f-3e69-8468-10717534cb09	5383922a-2ab3-5d81-b8cf-0282084b8c68	DEFAULT

[artembilan]

So, we probably can't do something automatic as out-of-the-box since an aggregator can't determine which group it has created before.
Therefore an existing feature in the AbstractCorrelatingMessageHandler via:

public final void setMessageStore(MessageGroupStore store) {
		this.messageStore = store;
		UniqueExpiryCallback expiryCallback =
				(messageGroupStore, group) -> this.forceReleaseProcessor.processMessageGroup(group);
		store.registerMessageGroupExpiryCallback(expiryCallback);
	}

(pay attention to the expiryCallback) and combination with an external MessageGroupStoreReaper supplied with the same MessageGroupStore which is ran at startup is the way to go. This way any orphaned groups are going to be expired.

Or if you really talk about a "rescheduling", then we probably need to introduce a new MessageGroupCallback which is going to be added only once at the reschedulePersistedMessageGroups() call and removed immediately. But again: the MessageGroupStore has to be unique per an aggregator to avoid collision with "not-your-groups".

WDYT?

[garyrussell]

Hmmm - since this is about expiring "orphaned" groups - no new message arrivals after startup - and given that a configured MessageGroupStoreReaper will solve this problem, I wonder if we can simply call MGS.expireMessageGroups() during startup and from time-to-time to clean up any groups that are in that state? e.g. Add 'cleanOrphanedGroups` timeout and interval.

[artembilan]

Hm. Do we need to introduce a new API since we just can document it properly and fully rely on the existing API of the MessageGroupStoreReaper? It has that run() to be called whenever it is necessary. Even if ii just a catch of an AsyncConsumerStartedEvent via a ApplicationEventListeningMessageProducer and call that run() in the service activator subscriber. The "cleanOrphanedGroups" is really achieved with already existing MessageGroupStoreReaper API and its configuration via a scheduled task. Why should we worry about an aggregator API in this case?

[garyrussell]

The reaper has a timeout configured, we can't use the groupTimeoutExpression because it might refer to a message.

But I guess it would be just a convenience; we could just document that the reaper is a better solution for persistent stores unless you need to use variable group timeouts based on messages, in which case you should still configure a reaper to clean up orphaned groups.

Bear in mind that when you use a reaper, you still need a separate MGS/region for each aggregator so the expirations will be processed by the right aggregator.

[artembilan]

Well, in that case we can recommend to call a MessageGroupStore.expireMessageGroups(timeout) directly if there is a requirement to deal with an arbitrary timeout. So, we don't remove extra groups which might be pretty "young" in this store.

[garyrussell]

e.g. Add 'cleanOrphanedGroups` timeout and interval.

That's what I meant by these two new properties - how often to purge and what timeout to use.

[artembilan]

OK. Since an aggregator is doing already some scheduling let's introduce one more for end-user convenience!
Technically this is still a reaper functionality with a scheduled task, but it is going already to be done in the framework component - less boilerplate code!

Did I get it right, Gary?

[toananthan]

In what version(s) of Spring Integration are you seeing this issue?

I did POC with below versions, I see issue in this version:
Spring Boot: 2.0.9
Spring Integration: 5.0.13

next week I will be working with below versions:
Spring Boot:2.1.14
Spring Integration: 5.1.10

after 3 weeks I will be working with below versions (going to be Prod version):
Spring Boot: 2.2.8
Spring Integration:5.2.7

[garyrussell]

Did I get it right, Gary?

Exactly - an embedded reaper.