From f348b232d39c0b03f21a4d1e083fc6f2b44468f0 Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Mon, 5 Sep 2022 17:48:28 +0200 Subject: [PATCH] WIP --- pom.xml | 9 +--- .../juplo/kafka/ApplicationConfiguration.java | 36 ++++++--------- .../de/juplo/kafka/ApplicationProperties.java | 16 ------- src/main/java/de/juplo/kafka/RestGateway.java | 4 +- src/main/resources/application.yml | 46 +++++++++++++------ 5 files changed, 49 insertions(+), 62 deletions(-) diff --git a/pom.xml b/pom.xml index de665ad..2fa762f 100644 --- a/pom.xml +++ b/pom.xml @@ -36,8 +36,8 @@ spring-boot-starter-validation - org.apache.kafka - kafka-clients + org.springframework.kafka + spring-kafka org.projectlombok @@ -48,11 +48,6 @@ spring-boot-starter-test test - - org.springframework.kafka - spring-kafka - test - org.springframework.kafka spring-kafka-test diff --git a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java index d9c7661..1d64221 100644 --- a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java +++ b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java @@ -1,47 +1,39 @@ package de.juplo.kafka; import org.apache.kafka.clients.producer.KafkaProducer; -import org.apache.kafka.common.serialization.IntegerSerializer; -import org.apache.kafka.common.serialization.StringSerializer; +import org.apache.kafka.clients.producer.Producer; +import org.springframework.boot.autoconfigure.kafka.KafkaProperties; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.kafka.annotation.EnableKafka; +import org.springframework.kafka.core.ProducerFactory; import java.util.Properties; @Configuration -@EnableConfigurationProperties(ApplicationProperties.class) +@EnableConfigurationProperties({ KafkaProperties.class, ApplicationProperties.class }) +@EnableKafka public class ApplicationConfiguration { @Bean public RestGateway restGateway( - ApplicationProperties properties, - KafkaProducer kafkaProducer) + ApplicationProperties applicationProperties, + KafkaProperties kafkaProperties, + Producer kafkaProducer) { return new RestGateway( - properties.getClientId(), - properties.getTopic(), - properties.getPartition(), + kafkaProperties.getClientId(), + applicationProperties.getTopic(), + applicationProperties.getPartition(), kafkaProducer); } @Bean(destroyMethod = "close") - public KafkaProducer kafkaProducer(ApplicationProperties properties) + public Producer kafkaProducer(ProducerFactory factory) { - 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", IntegerSerializer.class.getName()); - - return new KafkaProducer<>(props); + return factory.createProducer(); } } diff --git a/src/main/java/de/juplo/kafka/ApplicationProperties.java b/src/main/java/de/juplo/kafka/ApplicationProperties.java index 2bcbb7a..7d5f105 100644 --- a/src/main/java/de/juplo/kafka/ApplicationProperties.java +++ b/src/main/java/de/juplo/kafka/ApplicationProperties.java @@ -13,24 +13,8 @@ import javax.validation.constraints.NotNull; @Setter public class ApplicationProperties { - @NotNull - @NotEmpty - private String bootstrapServer; - @NotNull - @NotEmpty - private String clientId; @NotNull @NotEmpty private String topic; private Integer partition; - @NotNull - @NotEmpty - private String acks; - @NotNull - private Integer batchSize; - @NotNull - private Integer lingerMs; - @NotNull - @NotEmpty - private String compressionType; } diff --git a/src/main/java/de/juplo/kafka/RestGateway.java b/src/main/java/de/juplo/kafka/RestGateway.java index c50f465..53a87df 100644 --- a/src/main/java/de/juplo/kafka/RestGateway.java +++ b/src/main/java/de/juplo/kafka/RestGateway.java @@ -2,7 +2,7 @@ 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.Producer; import org.apache.kafka.clients.producer.ProducerRecord; import org.springframework.http.HttpStatus; import org.springframework.web.bind.annotation.*; @@ -18,7 +18,7 @@ public class RestGateway private final String id; private final String topic; private final Integer partition; - private final KafkaProducer producer; + private final Producer producer; private long produced = 0; diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index 218aaa1..deeb60a 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -1,12 +1,6 @@ sumup: gateway: - bootstrap-server: :9092 - client-id: DEV topic: test - acks: -1 - batch-size: 16384 - linger-ms: 0 - compression-type: gzip management: endpoint: shutdown: @@ -22,16 +16,38 @@ management: enabled: true info: kafka: - bootstrap-server: ${producer.bootstrap-server} - client-id: ${producer.client-id} - topic: ${producer.topic} - acks: ${producer.acks} - batch-size: ${producer.batch-size} - linger-ms: ${producer.linger-ms} - compression-type: ${producer.compression-type} + bootstrap-servers: ${spring.kafka.bootstrap-servers} + client-id: ${spring.kafka.client-id} + group-id: ${spring.kafka.consumer.group-id} + auto-offset-reset: ${spring.kafka.consumer.auto-offset-reset} + auto-commit-interval-ms: ${spring.kafka.consumer.properties.auto.commit.interval.ms} + topic: ${sumup.gateway.topic} + acks: ${spring.kafka.producer.acks} + batch-size: ${spring.kafka.producer.batch-size} + linger-ms: ${spring.kafka.producer.properties.linger.ms} + compression-type: ${spring.kafka.producer.compression-type} +spring: + kafka: + bootstrap-servers: :9092 + client-id: DEV + consumer: + auto-offset-reset: earliest + group-id: my-group + properties: + auto.commit.interval.ms: 5000 + producer: + acks: -1 + batch-size: 16384 + compression-type: gzip + key-serializer: org.apache.kafka.common.serialization.StringSerializer + value-serializer: org.springframework.kafka.support.serializer.IntegerSerializer + properties: + linger.ms: 0 + delivery.timeout.ms: 20000 # 20 Sekunden + request.timeout.ms: 10000 # 10 Sekunden logging: level: root: INFO - de.juplo: TRACE + de.juplo: DEBUG server: - port: 8880 + port: 8881 -- 2.20.1