X-Git-Url: http://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2FApplication.java;fp=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2FApplication.java;h=a9b466da3b0ba3eb8f45aadaf92419a89c7f9833;hb=28861eab2d4da8a0594a115de989ffeb90b05cd4;hp=718676b95e10c3590aeb7f185e0ee30da7bbcf21;hpb=1e858c1835d47259583cf3cce18921c550682e2b;p=demos%2Fkafka%2Ftraining diff --git a/src/main/java/de/juplo/kafka/Application.java b/src/main/java/de/juplo/kafka/Application.java index 718676b..a9b466d 100644 --- a/src/main/java/de/juplo/kafka/Application.java +++ b/src/main/java/de/juplo/kafka/Application.java @@ -7,17 +7,19 @@ import org.springframework.boot.ApplicationArguments; import org.springframework.boot.ApplicationRunner; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; -import org.springframework.scheduling.annotation.EnableAsync; import javax.annotation.PreDestroy; import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executor; +import java.util.concurrent.Future; @SpringBootApplication -@EnableAsync @Slf4j public class Application implements ApplicationRunner { + @Autowired + Executor executor; @Autowired Consumer kafkaConsumer; @Autowired @@ -27,25 +29,14 @@ public class Application implements ApplicationRunner public void run(ApplicationArguments args) throws Exception { log.info("Starting SimpleConsumer"); - simpleConsumer.run(); + executor.execute(simpleConsumer); } @PreDestroy - public void stop() throws ExecutionException, InterruptedException + public void shutdown() throws ExecutionException, InterruptedException { log.info("Signaling SimpleConsumer to quit its work"); kafkaConsumer.wakeup(); - - while (simpleConsumer.isRunning()) - { - log.info("Waiting for SimpleConsumer to finish its work"); - try - { - Thread.sleep(1000); - } - catch (InterruptedException e) {} - } - log.info("SimpleConsumer finished its work"); }