Use Retry

statistics-service-provider/src/main/java/de/sample/schulung/statistics/kafka/CustomerEventListener.java

@@ -2,28 +2,41 @@
 
 import de.sample.schulung.statistics.domain.Customer;
 import de.sample.schulung.statistics.domain.CustomersService;
+import jakarta.validation.Valid;
 import jakarta.validation.ValidationException;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.annotation.RetryableTopic;
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.messaging.handler.annotation.Header;
 import org.springframework.messaging.handler.annotation.Payload;
+import org.springframework.retry.annotation.Backoff;
 import org.springframework.stereotype.Component;
+import org.springframework.validation.annotation.Validated;
 
+@Validated
 @Component
 @Slf4j
 @RequiredArgsConstructor
 public class CustomerEventListener {
 
   private final CustomersService customersService;
 
+  // Versuch, danach 1 Versuch nach 1s, danach Versuch nach 2s, danach Versuch nach 4s
+  @RetryableTopic(
+    attempts = "4",
+    backoff = @Backoff(
+      delay = 1000L,
+      multiplier = 2
+    )
+  )
   @KafkaListener(
     topics = KafkaConstants.CUSTOMER_EVENTS_TOPIC
   )
   public void consume(
-    @Payload CustomerEventRecord record,
+    @Valid @Payload CustomerEventRecord record,
     @Header(KafkaHeaders.RECEIVED_PARTITION) String partition,
     @Header(KafkaHeaders.OFFSET) int offset,
     Acknowledgment  acknowledgment