Выполните метод после того, как все сообщения помещены в kafka в весеннем потоке интеграции облака после разделения

Question

Как я могу выполнить метод после того, как все сообщения будут помещены в Kafka после разделения.

Я пытался использовать handle () после канала, но в итоге не отправлял 2-е сообщение в Kafka.

 IntegrationFlowBuilder flowBuilder = IntegrationFlows
        .from(() -> jdbcTemplate....);
 flowBuilder.split();
 flowBuilder.channel(messageChannel);

Я хотел бы выполнить метод после записи всех разделенных сообщений в Kafka.

Answer 1

Поскольку вы разделены и только после этого отправляете в Кафку, нет способа поймать их всех, кроме aggregate после.Вы можете сделать это с помощью publishSubscribeChannel, чтобы отправить Kafka и агрегатору одно и то же сообщение.Когда агрегатор освобождает группу, вы можете выполнить вызов метода.