1 package de.juplo.kafka;
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;
13 import java.util.List;
17 @SpringBootApplication
18 public class Application implements ApplicationRunner
20 public final static String ARG_NUM = "num";
23 KafkaTemplate<String, String> kafkaTemplate;
26 void send(String key, String value)
28 ListenableFuture<SendResult<String, String>> listenableFuture =
29 kafkaTemplate.sendDefault(key, value);
31 listenableFuture.addCallback(
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));
42 public void run(ApplicationArguments args)
46 if (args.containsOption(ARG_NUM))
48 List<String> numArgs = args.getOptionValues(ARG_NUM);
49 if (numArgs.size() > 1)
52 "Only one occurence of argument {} is allowed, but found: {}",
60 num = Integer.parseInt(numArgs.get(0));
62 catch (NumberFormatException e)
64 log.error("{} is not a number: {}", numArgs.get(0), e.getMessage());
68 for (int i = 0; i < num; i++)
70 send(Long.toString(i%10), Long.toString(i));
75 catch (InterruptedException e)
77 log.info("Interrupted after sending {} messages", i);
84 public static void main(String[] args)
86 SpringApplication.run(Application.class, args);