Properly restarting ContainerGroupSequencer after processing all groups

[curban111]

Hi everyone,

We are using Kafka in a project where we are migrating from a monolithic Oracle database application to a Self-Contained System (SCS) architecture. The goal is to stream data and changes from the Oracle database to several SCSs. To capture the changes, we are using the Debezium Oracle connector. This connector populates a Kafka topic for each Oracle table with events. These events are then translated into a new data model.

Both the Oracle tables and the new data model have dependencies that must be respected when processing events. To handle these dependencies, we created groups using the ContainerGroupSequencer. This allows us to sequentially process data with interdependencies.

However, we encountered an issue: once the last group is processed, the sequencer stops because we have enabled the stopOnLastGroup property. To restart the sequencer, we tried using ContainerEvents. Despite our efforts, the state where all listeners are no longer running is never reached. Here is our approach:

@EventListener
public void restartIfNeeded(ListenerContainerIdleEvent event) throws InterruptedException {
    if (registry.getAllListenerContainers().stream().noneMatch(MessageListenerContainer::isRunning)) {
        registry.getAllListenerContainers().forEach(Lifecycle::start);
    }
}

How can we properly restart the ContainerGroupSequencer after processing all groups?