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

   1 package de.juplo.kafka;
   2
   3 import lombok.extern.slf4j.Slf4j;
   4 import org.springframework.beans.factory.annotation.Autowired;
   5 import org.springframework.boot.ApplicationArguments;
   6 import org.springframework.boot.ApplicationRunner;
   7 import org.springframework.boot.SpringApplication;
   8 import org.springframework.boot.autoconfigure.SpringBootApplication;
   9 import org.springframework.kafka.core.KafkaTemplate;
  10 import org.springframework.kafka.support.SendResult;
  11 import org.springframework.util.concurrent.ListenableFuture;
  12
  13
  14 // tag::supersimple[]
  15 @Slf4j
  16 @SpringBootApplication
  17 public class Application implements ApplicationRunner
  18 {
  19   @Autowired
  20   KafkaTemplate<String, String> kafkaTemplate;
  21
  22   @Override
  23   public void run(ApplicationArguments args)
  24   {
  25     for (int i = 0; i < 100; i++)
  26     {
  27       // end::supersimple[]
  28       // tag::callback[]
  29       ListenableFuture<SendResult<String, String>> listenableFuture =
  30       // tag::supersimple[]
  31           kafkaTemplate.send("test", Long.toString(i%10), Long.toString(i));
  32       // end::supersimple[]
  33
  34       listenableFuture.addCallback(
  35           result -> log.info(
  36               "Sent {}={} to partition={}, offset={}",
  37               result.getProducerRecord().key(),
  38               result.getProducerRecord().value(),
  39               result.getRecordMetadata().partition(),
  40               result.getRecordMetadata().offset()),
  41           e -> log.error("ERROR sendig message", e));
  42       // end::callback[]
  43       // tag::supersimple[]
  44     }
  45   }
  46
  47   public static void main(String[] args)
  48   {
  49     SpringApplication.run(Application.class, args);
  50   }
  51 }
  52 // end::supersimple[]