From 116e082d563c20fc1ada5e7ac684fd1ce5aa47aa Mon Sep 17 00:00:00 2001 From: Andrea Cavalli Date: Fri, 30 Sep 2022 15:25:06 +0200 Subject: [PATCH] Fix resubscription failures --- src/main/java/it/tdlight/reactiveapi/KafkaConsumer.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/main/java/it/tdlight/reactiveapi/KafkaConsumer.java b/src/main/java/it/tdlight/reactiveapi/KafkaConsumer.java index e86b3e9..e1bdebb 100644 --- a/src/main/java/it/tdlight/reactiveapi/KafkaConsumer.java +++ b/src/main/java/it/tdlight/reactiveapi/KafkaConsumer.java @@ -120,8 +120,8 @@ public abstract class KafkaConsumer { return new Timestamped<>(1, record.value()); } }) - .transform(ReactorUtils::subscribeOnce) .transform(this::retryIfCleanup) - .transform(this::retryIfCommitFailed); + .transform(this::retryIfCommitFailed) + .transform(ReactorUtils::subscribeOnce); } }