diff --git a/src/main/java/it/tdlight/reactiveapi/KafkaConsumer.java b/src/main/java/it/tdlight/reactiveapi/KafkaConsumer.java index a96013b..47bcc41 100644 --- a/src/main/java/it/tdlight/reactiveapi/KafkaConsumer.java +++ b/src/main/java/it/tdlight/reactiveapi/KafkaConsumer.java @@ -129,9 +129,8 @@ public abstract class KafkaConsumer { return new Timestamped<>(1, record.value()); } }) - .transform(ReactorUtils::subscribeOnce); - //todo: check if they must be re-enabled - //.transform(this::retryIfCleanup) - //.transform(this::retryIfCommitFailed); + .transform(ReactorUtils::subscribeOnce) + .transform(this::retryIfCleanup) + .transform(this::retryIfCommitFailed); } }