juplo.de Git - demos/kafka/training/blob - src/main/java/de/juplo/kafka/RestProducer.java

   1 package de.juplo.kafka;
   2
   3 import lombok.extern.slf4j.Slf4j;
   4 import org.apache.kafka.clients.producer.RecordMetadata;
   5 import org.springframework.http.HttpStatus;
   6 import org.springframework.kafka.core.KafkaTemplate;
   7 import org.springframework.web.bind.annotation.*;
   8 import org.springframework.web.context.request.async.DeferredResult;
   9
  10 import javax.annotation.PreDestroy;
  11 import java.util.concurrent.ExecutionException;
  12
  13
  14 @Slf4j
  15 @RestController
  16 public class RestProducer
  17 {
  18   private final String id;
  19   private final KafkaTemplate<String, Object> kafkaTemplate;
  20
  21   private long produced = 0;
  22
  23   public RestProducer(
  24       ApplicationProperties properties,
  25       KafkaTemplate<String, Object> kafkaTemplate)
  26   {
  27     this.id = properties.getClientId();
  28     this.kafkaTemplate = kafkaTemplate;
  29   }
  30
  31   @PostMapping(path = "{key}")
  32   public DeferredResult<ProduceResult> message(
  33       @PathVariable String key,
  34       @RequestBody String value)
  35   {
  36     key = key.trim();
  37     return send(key, new ClientMessage(key, value));
  38   }
  39
  40   @PutMapping(path = "{key}")
  41   public DeferredResult<ProduceResult> message(@PathVariable String key)
  42   {
  43     key = key.trim();
  44     return send(key, new FooMessage(key, System.currentTimeMillis()));
  45   }
  46
  47   @PostMapping(path = "/")
  48   public DeferredResult<ProduceResult> greeting(
  49       @RequestBody String name)
  50   {
  51     name = name.trim();
  52     return send(name, new Greeting(name));
  53   }
  54
  55   private DeferredResult<ProduceResult> send(String key, Object value)
  56   {
  57     DeferredResult<ProduceResult> result = new DeferredResult<>();
  58
  59     final long time = System.currentTimeMillis();
  60
  61     kafkaTemplate.sendDefault(key, value).addCallback(
  62       (sendResult) ->
  63       {
  64         long now = System.currentTimeMillis();
  65
  66         // HANDLE SUCCESS
  67         RecordMetadata metadata = sendResult.getRecordMetadata();
  68         produced++;
  69         result.setResult(new ProduceSuccess(metadata.partition(), metadata.offset()));
  70         log.debug(
  71             "{} - Sent key={} message={} partition={}/{} timestamp={} latency={}ms",
  72             id,
  73             key,
  74             value,
  75             metadata.partition(),
  76             metadata.offset(),
  77             metadata.timestamp(),
  78             now - time
  79         );
  80       },
  81       (e) ->
  82       {
  83         long now = System.currentTimeMillis();
  84
  85         // HANDLE ERROR
  86         result.setErrorResult(new ProduceFailure(e));
  87         log.error(
  88             "{} - ERROR key={} timestamp=-1 latency={}ms: {}",
  89             id,
  90             key,
  91             now - time,
  92             e.toString()
  93         );
  94       });
  95
  96     long now = System.currentTimeMillis();
  97     log.trace(
  98         "{} - Queued key={} latency={}ms",
  99         id,
 100         key,
 101         now - time
 102     );
 103
 104     return result;
 105   }
 106
 107   @ExceptionHandler
 108   @ResponseStatus(HttpStatus.BAD_REQUEST)
 109   public ErrorResponse illegalStateException(IllegalStateException e)
 110   {
 111     return new ErrorResponse(e.getMessage(), HttpStatus.BAD_REQUEST.value());
 112   }
 113
 114   @PreDestroy
 115   public void destroy() throws ExecutionException, InterruptedException
 116   {
 117     log.info("{} - Destroy!", id);
 118     log.info("{}: Produced {} messages in total, exiting!", id, produced);
 119   }
 120 }