How to send Messages after Spring Integration application restarted?

I have a small Spring Integration application, I'm storing messages and messaging groups in the database. Currently, I have a case when some messages/groups are waiting to be sent after group timeout, but the application restarted. And when the application started I still have messages in DB and they won't be sent. I need some configuration to send expired message group from DB or resume timer. I tried to use reaper, but it does not work as expected. My code is:

@Configuration
public class ConsumingChannelConfig {
@Bean
public DirectChannel consumingChannel() {
return new DirectChannel();
}
@Bean
public KafkaMessageDrivenChannelAdapter<String, String> kafkaMessageDrivenChannelAdapter() {
KafkaMessageDrivenChannelAdapter<String, String> kafkaMessageDrivenChannelAdapter =
new KafkaMessageDrivenChannelAdapter<>(kafkaListenerContainer());
kafkaMessageDrivenChannelAdapter.setOutputChannel(consumingChannel());
MessagingMessageConverter messageConverter = new MessagingMessageConverter();
messageConverter.setGenerateMessageId(true);
kafkaMessageDrivenChannelAdapter.setRecordMessageConverter(messageConverter);
return kafkaMessageDrivenChannelAdapter;
}
@Bean
public DataSource getDataSource() {
return ...;
}
@Bean
public JdbcMessageStore jdbcMessageStore() {
return new JdbcMessageStore(getDataSource());
}
@ServiceActivator(inputChannel = "consumingChannel")
@Bean
public MessageHandler aggregator() {
long timeout = 10000L;
AggregatingMessageHandler aggregator =
new AggregatingMessageHandler(new DefaultAggregatingMessageGroupProcessor(),
jdbcMessageStore());
aggregator.setOutputChannel((message, l) -> {
System.out.println("MESSAGE: " + message);
return true;
});
aggregator.setGroupTimeoutExpression(new ValueExpression<>(timeout));
//        aggregator.setTaskScheduler(this.taskScheduler);
aggregator.setCorrelationStrategy(new MyCorrelationStrategy());
aggregator.setSendPartialResultOnExpiry(true);
aggregator.setExpireGroupsUponCompletion(true);
aggregator.setExpireGroupsUponTimeout(true);
aggregator.setDiscardChannel((message, timeout1) -> {
System.out.println("DISCARD: " + message + ", timeout: " + timeout1);
return true;
});
aggregator.setReleaseStrategy(new ReleaseStrategy() {
@Override
public boolean canRelease(MessageGroup group) {
return System.currentTimeMillis() - group.getTimestamp() >= timeout;
}
});
return aggregator;
}
@Bean
public MessageGroupStoreReaper reaper() {
MessageGroupStoreReaper reaper = new MessageGroupStoreReaper(jdbcMessageStore());
reaper.setPhase(1);
reaper.setTimeout(2000L);
reaper.setAutoStartup(true);
//        reaper.setExpireOnDestroy(true);
return reaper;
}
@Bean
public ConcurrentMessageListenerContainer<String, String> kafkaListenerContainer() {
ContainerProperties containerProps = new ContainerProperties("spring-integration-topic");
return new ConcurrentMessageListenerContainer<>(
consumerFactory(), containerProps);
}
@Bean
public ConsumerFactory<String, String> consumerFactory() {
return new DefaultKafkaConsumerFactory<>(consumerConfigs());
}
@Bean
public Map<String, Object> consumerConfigs() {
Map<String, Object> properties = new HashMap<>();
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
properties.put(ConsumerConfig.GROUP_ID_CONFIG, "spring-integration");
// automatically reset the offset to the earliest offset
properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
//        DefaultKafkaHeaderMapper mapper = new DefaultKafkaHeaderMapper();
return properties;
}
}

UPD: My Solution

@EnableScheduling
@SpringBootApplication
public class SpringIntegrationExampleApplication {
public static void main(String[] args) {
SpringApplication.run(SpringIntegrationExampleApplication.class, args);
}
@Autowired
private MessageGroupStoreReaper reaper;
@Scheduled(initialDelay = 2000, fixedDelay = Long.MAX_VALUE)
public void start() {
reaper.run();
}
}

The MessageGroupStoreReaper doesn't work by itself, it has to be called from a @Scheduled method: https://docs.spring.io/spring-integration/docs/current/reference/html/message-routing.html#reaper

However there is a nice option for you from an aggregator perspective:

/**
* Perform a {@link MessageGroupStore#expireMessageGroups(long)} with the provided {@link #expireTimeout}.
* Can be called externally at any time.
* Internally it is called from the scheduled task with the configured {@link #expireDuration}.
* @since 5.4
*/
public void purgeOrphanedGroups() {

You just need to set that expireTimeout > 0:

/**
* Configure a timeout in milliseconds for purging old orphaned groups from the store.
* Used on startup and when an {@link #expireDuration} is provided, the task for running
* {@link #purgeOrphanedGroups()} is scheduled with that period.
* The {@link #forceReleaseProcessor} is used to process those expired groups according
* the "force complete" options. A group can be orphaned if a persistent message group
* store is used and no new messages arrive for that group after a restart.
* @param expireTimeout the number of milliseconds to determine old orphaned groups in the store to purge.
* @since 5.4
* @see #purgeOrphanedGroups()
*/
public void setExpireTimeout(long expireTimeout) {

See also docs on the matter: https://docs.spring.io/spring-integration/docs/current/reference/html/message-routing.html#aggregator-xml

> Starting with version 5.4, the aggregator (and resequencer) can be configured to expire orphaned groups (groups in a persistent message store that might not otherwise be released).