1 package de.juplo.kafka;
3 import lombok.extern.slf4j.Slf4j;
4 import org.apache.kafka.clients.consumer.Consumer;
5 import org.springframework.beans.factory.annotation.Autowired;
6 import org.springframework.boot.ApplicationArguments;
7 import org.springframework.boot.ApplicationRunner;
8 import org.springframework.boot.SpringApplication;
9 import org.springframework.boot.autoconfigure.SpringBootApplication;
10 import org.springframework.context.ConfigurableApplicationContext;
11 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
12 import org.springframework.util.concurrent.ListenableFuture;
14 import javax.annotation.PreDestroy;
15 import java.util.concurrent.ExecutionException;
18 @SpringBootApplication
20 public class Application implements ApplicationRunner
23 ThreadPoolTaskExecutor taskExecutor;
25 Consumer<?, ?> kafkaConsumer;
27 SimpleConsumer simpleConsumer;
29 ConfigurableApplicationContext context;
31 ListenableFuture<Integer> consumerJob;
34 public void run(ApplicationArguments args) throws Exception
36 log.info("Starting SimpleConsumer");
37 consumerJob = taskExecutor.submitListenable(simpleConsumer);
38 consumerJob.addCallback(
41 log.info("SimpleConsumer exited normally, exit-status: {}", exitStatus);
42 SpringApplication.exit(context, () -> exitStatus);
46 log.error("SimpleConsumer exited abnormally!", t);
47 SpringApplication.exit(context, () -> 2);
52 public void shutdown() throws ExecutionException, InterruptedException
54 log.info("Signaling SimpleConsumer to quit its work");
55 kafkaConsumer.wakeup();
56 log.info("Waiting for SimpleConsumer to finish its work");
58 log.info("SimpleConsumer finished its work");
62 public static void main(String[] args)
64 SpringApplication.run(Application.class, args);