From: Kai Moritz <kai@juplo.de>
Date: Sun, 10 Apr 2022 20:15:34 +0000 (+0200)
Subject: Merge branch 'endless-stream-consumer' into rebalance-listener
X-Git-Tag: deserialization-synchroner-test~16
X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;h=92882a04df8b54ae6ac000c14a9793f7411318f9;p=demos%2Fkafka%2Ftraining

Merge branch 'endless-stream-consumer' into rebalance-listener
---

92882a04df8b54ae6ac000c14a9793f7411318f9
diff --cc src/main/java/de/juplo/kafka/ApplicationHealthIndicator.java
index 0000000,0000000..ab9782c
new file mode 100644
--- /dev/null
+++ b/src/main/java/de/juplo/kafka/ApplicationHealthIndicator.java
@@@ -1,0 -1,0 +1,32 @@@
++package de.juplo.kafka;
++
++import lombok.RequiredArgsConstructor;
++import org.springframework.boot.actuate.health.Health;
++import org.springframework.boot.actuate.health.HealthIndicator;
++import org.springframework.stereotype.Component;
++
++
++@Component
++@RequiredArgsConstructor
++public class ApplicationHealthIndicator implements HealthIndicator
++{
++  private final EndlessConsumer consumer;
++
++
++  @Override
++  public Health health()
++  {
++    try
++    {
++      return consumer
++          .exitStatus()
++          .map(Health::down)
++          .orElse(Health.outOfService())
++          .build();
++    }
++    catch (IllegalStateException e)
++    {
++      return Health.up().build();
++    }
++  }
++}
diff --cc src/main/java/de/juplo/kafka/DriverController.java
index a504842,a02fd2c..1525f5a
--- a/src/main/java/de/juplo/kafka/DriverController.java
+++ b/src/main/java/de/juplo/kafka/DriverController.java
@@@ -1,11 -1,12 +1,14 @@@
  package de.juplo.kafka;
  
  import lombok.RequiredArgsConstructor;
+ import org.springframework.http.HttpStatus;
+ import org.springframework.web.bind.annotation.ExceptionHandler;
 +import org.springframework.web.bind.annotation.GetMapping;
  import org.springframework.web.bind.annotation.PostMapping;
+ import org.springframework.web.bind.annotation.ResponseStatus;
  import org.springframework.web.bind.annotation.RestController;
  
 +import java.util.Map;
  import java.util.concurrent.ExecutionException;
  
  
@@@ -28,10 -29,10 +31,16 @@@ public class DriverControlle
      consumer.stop();
    }
  
- 
 +  @GetMapping("seen")
 +  public Map<Integer, Map<String, Integer>> seen()
 +  {
 +    return consumer.getSeen();
 +  }
++
+   @ExceptionHandler
+   @ResponseStatus(HttpStatus.BAD_REQUEST)
+   public ErrorResponse illegalStateException(IllegalStateException e)
+   {
+     return new ErrorResponse(e.getMessage(), HttpStatus.BAD_REQUEST.value());
+   }
  }