From 65ab9dbcb8b311c94e0f73119335e9e275ca031d Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Sat, 13 Aug 2022 18:37:19 +0200 Subject: [PATCH] refactor: `RestProducer` wird explizit erzeugt MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit * `ApplicationConfiguration` eingeführt. * Um eine bessere Kontrolle über die verwendeten Parameter zu erhalten, wird `RestProducer` jetzt explizit in `ApplicationConfiguration` erzeugt. * `KafkaProducer` wird in `ApplicationConfiguration` erzeugt und in `RestProducer` hereingereicht. --- src/main/java/de/juplo/kafka/Application.java | 7 --- .../juplo/kafka/ApplicationConfiguration.java | 46 +++++++++++++++++++ .../java/de/juplo/kafka/RestProducer.java | 37 ++------------- .../java/de/juplo/kafka/ApplicationTests.java | 1 - 4 files changed, 50 insertions(+), 41 deletions(-) create mode 100644 src/main/java/de/juplo/kafka/ApplicationConfiguration.java diff --git a/src/main/java/de/juplo/kafka/Application.java b/src/main/java/de/juplo/kafka/Application.java index 9f3e3ed..0069257 100644 --- a/src/main/java/de/juplo/kafka/Application.java +++ b/src/main/java/de/juplo/kafka/Application.java @@ -1,17 +1,10 @@ package de.juplo.kafka; -import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; -import org.springframework.boot.context.properties.EnableConfigurationProperties; -import org.springframework.context.annotation.Bean; -import org.springframework.util.Assert; - -import java.util.concurrent.Executors; @SpringBootApplication -@EnableConfigurationProperties(ApplicationProperties.class) public class Application { public static void main(String[] args) diff --git a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java new file mode 100644 index 0000000..0642aa4 --- /dev/null +++ b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java @@ -0,0 +1,46 @@ +package de.juplo.kafka; + +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.common.serialization.StringSerializer; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +import java.util.Properties; + + +@Configuration +@EnableConfigurationProperties(ApplicationProperties.class) +public class ApplicationConfiguration +{ + @Bean + public RestProducer restProducer( + ApplicationProperties properties, + KafkaProducer kafkaProducer) + { + return + new RestProducer( + properties.getClientId(), + properties.getTopic(), + properties.getPartition(), + kafkaProducer); + } + + @Bean(destroyMethod = "close") + public KafkaProducer kafkaProducer(ApplicationProperties properties) + { + Properties props = new Properties(); + props.put("bootstrap.servers", properties.getBootstrapServer()); + props.put("client.id", properties.getClientId()); + props.put("acks", properties.getAcks()); + props.put("batch.size", properties.getBatchSize()); + props.put("delivery.timeout.ms", 20000); // 20 Sekunden + props.put("request.timeout.ms", 10000); // 10 Sekunden + props.put("linger.ms", properties.getLingerMs()); + props.put("compression.type", properties.getCompressionType()); + props.put("key.serializer", StringSerializer.class.getName()); + props.put("value.serializer", StringSerializer.class.getName()); + + return new KafkaProducer<>(props); + } +} diff --git a/src/main/java/de/juplo/kafka/RestProducer.java b/src/main/java/de/juplo/kafka/RestProducer.java index b5bb2d9..27c9911 100644 --- a/src/main/java/de/juplo/kafka/RestProducer.java +++ b/src/main/java/de/juplo/kafka/RestProducer.java @@ -1,20 +1,21 @@ package de.juplo.kafka; +import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; -import org.apache.kafka.common.serialization.StringSerializer; import org.springframework.http.HttpStatus; import org.springframework.web.bind.annotation.*; import org.springframework.web.context.request.async.DeferredResult; import javax.annotation.PreDestroy; -import java.util.Properties; import java.util.concurrent.ExecutionException; @Slf4j -@RestController +@RequestMapping +@ResponseBody +@RequiredArgsConstructor public class RestProducer { private final String id; @@ -24,27 +25,6 @@ public class RestProducer private long produced = 0; - public RestProducer(ApplicationProperties properties) - { - this.id = properties.getClientId(); - this.topic = properties.getTopic(); - this.partition = properties.getPartition(); - - Properties props = new Properties(); - props.put("bootstrap.servers", properties.getBootstrapServer()); - props.put("client.id", properties.getClientId()); - props.put("acks", properties.getAcks()); - props.put("batch.size", properties.getBatchSize()); - props.put("delivery.timeout.ms", 20000); // 20 Sekunden - props.put("request.timeout.ms", 10000); // 10 Sekunden - props.put("linger.ms", properties.getLingerMs()); - props.put("compression.type", properties.getCompressionType()); - props.put("key.serializer", StringSerializer.class.getName()); - props.put("value.serializer", StringSerializer.class.getName()); - - this.producer = new KafkaProducer<>(props); - } - @PostMapping(path = "{key}") public DeferredResult send( @PathVariable String key, @@ -113,13 +93,4 @@ public class RestProducer { return new ErrorResponse(e.getMessage(), HttpStatus.BAD_REQUEST.value()); } - - @PreDestroy - public void destroy() throws ExecutionException, InterruptedException - { - log.info("{} - Destroy!", id); - log.info("{} - Closing the KafkaProducer", id); - producer.close(); - log.info("{}: Produced {} messages in total, exiting!", id, produced); - } } diff --git a/src/test/java/de/juplo/kafka/ApplicationTests.java b/src/test/java/de/juplo/kafka/ApplicationTests.java index cf70c81..646a335 100644 --- a/src/test/java/de/juplo/kafka/ApplicationTests.java +++ b/src/test/java/de/juplo/kafka/ApplicationTests.java @@ -20,7 +20,6 @@ import static de.juplo.kafka.ApplicationTests.PARTITIONS; import static de.juplo.kafka.ApplicationTests.TOPIC; import static org.awaitility.Awaitility.*; import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post; -import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.put; import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; -- 2.20.1