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();
    }

}


How to send Messages after Spring Integration application restarted?
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).

