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

   1 package de.juplo.kafka;
   2
   3 import lombok.RequiredArgsConstructor;
   4 import lombok.extern.slf4j.Slf4j;
   5 import org.apache.kafka.clients.producer.KafkaProducer;
   6 import org.apache.kafka.clients.producer.ProducerRecord;
   7 import org.springframework.http.HttpStatus;
   8 import org.springframework.web.bind.annotation.*;
   9 import org.springframework.web.context.request.async.DeferredResult;
  10
  11 import javax.annotation.PreDestroy;
  12 import java.util.concurrent.ExecutionException;
  13
  14
  15 @Slf4j
  16 @RequestMapping
  17 @ResponseBody
  18 @RequiredArgsConstructor
  19 public class RestProducer
  20 {
  21   private final String id;
  22   private final String topic;
  23   private final Integer partition;
  24   private final KafkaProducer<String, String> producer;
  25
  26   private long produced = 0;
  27
  28   @PostMapping(path = "{key}")
  29   public DeferredResult<ProduceResult> send(
  30       @PathVariable String key,
  31       @RequestHeader(name = "X-id", required = false) Long correlationId,
  32       @RequestBody String value)
  33   {
  34     DeferredResult<ProduceResult> result = new DeferredResult<>();
  35
  36     final long time = System.currentTimeMillis();
  37
  38     final ProducerRecord<String, String> record = new ProducerRecord<>(
  39         topic,  // Topic
  40         key,    // Key
  41         value   // Value
  42     );
  43
  44     producer.send(record, (metadata, e) ->
  45     {
  46       long now = System.currentTimeMillis();
  47       if (e == null)
  48       {
  49         // HANDLE SUCCESS
  50         produced++;
  51         result.setResult(new ProduceSuccess(metadata.partition(), metadata.offset()));
  52         log.debug(
  53             "{} - Sent key={} message={} partition={}/{} timestamp={} latency={}ms",
  54             id,
  55             record.key(),
  56             record.value(),
  57             metadata.partition(),
  58             metadata.offset(),
  59             metadata.timestamp(),
  60             now - time
  61         );
  62       }
  63       else
  64       {
  65         // HANDLE ERROR
  66         result.setErrorResult(new ProduceFailure(e));
  67         log.error(
  68             "{} - ERROR key={} timestamp={} latency={}ms: {}",
  69             id,
  70             record.key(),
  71             metadata == null ? -1 : metadata.timestamp(),
  72             now - time,
  73             e.toString()
  74         );
  75       }
  76     });
  77
  78     long now = System.currentTimeMillis();
  79     log.trace(
  80         "{} - Queued message with key={} latency={}ms",
  81         id,
  82         record.key(),
  83         now - time
  84     );
  85
  86     return result;
  87   }
  88
  89   @ExceptionHandler
  90   @ResponseStatus(HttpStatus.BAD_REQUEST)
  91   public ErrorResponse illegalStateException(IllegalStateException e)
  92   {
  93     return new ErrorResponse(e.getMessage(), HttpStatus.BAD_REQUEST.value());
  94   }
  95 }