More control on event publishing

MAIF · Nov 29, 2023 · 400611b · 400611b
1 parent 9e1ff0f
commit 400611b
Showing 1 changed file with 2 additions and 2 deletions.
diff --git a/...-core-reactor/src/main/java/fr/maif/reactor/eventsourcing/ReactorKafkaEventPublisher.java b/...-core-reactor/src/main/java/fr/maif/reactor/eventsourcing/ReactorKafkaEventPublisher.java
@@ -83,15 +83,15 @@ public <TxCtx> void start(EventStore<TxCtx, E, Meta, Context> eventStore, Concur
                             .from(eventStore.loadEventsUnpublished(tx, strategy))
                             .transform(publishToKafka(eventStore, Option.some(tx), groupFlow))
                             .doOnNext(logProgressSink::tryEmitNext)
-                            .concatMap(any -> Mono.fromCompletionStage(() -> {
+                            .collectList()
+                            .flatMap(any -> Mono.fromCompletionStage(() -> {
                                 LOGGER.info("Replaying events not published in DB is finished for {}", topic);
                                 return eventStore.commitOrRollback(Option.none(), tx);
                             }))
                             .doOnError(e -> {
                                 eventStore.commitOrRollback(Option.of(e), tx);
                                 LOGGER.error("Error replaying non published events to kafka for " + topic, e);
                             })
-                            .collectList()
                             .map(__ -> Tuple.empty());
                 })
                 .retryWhen(Retry.backoff(10, restartInterval)