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 import java.util.List;
  14
  15
  16 @Slf4j
  17 @SpringBootApplication
  18 public class Application implements ApplicationRunner
  19 {
  20   public final static String ARG_NUM = "num";
  21
  22   @Autowired
  23   KafkaTemplate<String, String> kafkaTemplate;
  24
  25
  26   void send(String key, String value)
  27   {
  28     ListenableFuture<SendResult<String, String>> listenableFuture =
  29         kafkaTemplate.sendDefault(key, value);
  30
  31     listenableFuture.addCallback(
  32         result -> log.debug(
  33             "Sent {}={} to partition={}, offset={}",
  34             result.getProducerRecord().key(),
  35             result.getProducerRecord().value(),
  36             result.getRecordMetadata().partition(),
  37             result.getRecordMetadata().offset()),
  38         e -> log.error("ERROR sendig message", e));
  39   }
  40
  41   @Override
  42   public void run(ApplicationArguments args)
  43   {
  44     int num = 10;
  45
  46     if (args.containsOption(ARG_NUM))
  47     {
  48       List<String> numArgs = args.getOptionValues(ARG_NUM);
  49       if (numArgs.size() > 1)
  50       {
  51         log.error(
  52             "Only one occurence of argument {} is allowed, but found: {}",
  53             ARG_NUM,
  54             numArgs.size());
  55         return;
  56       }
  57
  58       try
  59       {
  60         num = Integer.parseInt(numArgs.get(0));
  61       }
  62       catch (NumberFormatException e)
  63       {
  64         log.error("{} is not a number: {}", numArgs.get(0), e.getMessage());
  65       }
  66     }
  67
  68     for (int i = 0; i < num; i++)
  69     {
  70       send(Long.toString(i%10), Long.toString(i));
  71       try
  72       {
  73         Thread.sleep(500);
  74       }
  75       catch (InterruptedException e)
  76       {
  77         log.info("Interrupted after sending {} messages", i);
  78         return;
  79       }
  80     }
  81   }
  82
  83
  84   public static void main(String[] args)
  85   {
  86     SpringApplication.run(Application.class, args);
  87   }
  88 }