Retries

2022-09-22 03:01:59 +02:00 · 2022-09-22 03:01:59 +02:00 · dfc393c953
parent 41be43d711
commit dfc393c953
1 changed files with 3 additions and 4 deletions
--- 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<K> {
 						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);
 	}
 }