From: Kai Moritz Date: Fri, 22 Apr 2022 09:08:37 +0000 (+0200) Subject: Springify: Konfiguration erfolgt über `KafkaProperties` X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;h=026b9db62b3d6c3576fe654a3fdfffa143b08205;p=demos%2Fkafka%2Ftraining Springify: Konfiguration erfolgt über `KafkaProperties` * Per `git cherry-pick` aus `springified-consumer--config' übernommen. * Conflicts: ** src/main/java/de/juplo/kafka/ApplicationConfiguration.java ** src/main/java/de/juplo/kafka/ApplicationProperties.java ** src/main/resources/application.yml ** src/test/java/de/juplo/kafka/ApplicationTests.java --- diff --git a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java index 596be26..92c7abe 100644 --- a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java +++ b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java @@ -3,6 +3,7 @@ package de.juplo.kafka; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.common.serialization.StringDeserializer; +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; @@ -16,18 +17,19 @@ import java.util.concurrent.Executors; @Configuration -@EnableConfigurationProperties(ApplicationProperties.class) +@EnableConfigurationProperties({ KafkaProperties.class, ApplicationProperties.class }) public class ApplicationConfiguration { @Bean public ApplicationRecordHandler recordHandler( AdderResults adderResults, - ApplicationProperties properties) + KafkaProperties kafkaProperties, + ApplicationProperties applicationProperties) { return new ApplicationRecordHandler( adderResults, - Optional.ofNullable(properties.getThrottle()), - properties.getClientId()); + Optional.ofNullable(applicationProperties.getThrottle()), + kafkaProperties.getClientId()); } @Bean @@ -42,16 +44,17 @@ public class ApplicationConfiguration AdderResults adderResults, StateRepository stateRepository, Consumer consumer, - ApplicationProperties properties) + KafkaProperties kafkaProperties, + ApplicationProperties applicationProperties) { return new ApplicationRebalanceListener( recordHandler, adderResults, stateRepository, - properties.getClientId(), - properties.getTopic(), + kafkaProperties.getClientId(), + applicationProperties.getTopic(), Clock.systemDefaultZone(), - properties.getCommitInterval(), + kafkaProperties.getConsumer().getAutoCommitInterval(), consumer); } @@ -61,13 +64,14 @@ public class ApplicationConfiguration ExecutorService executor, ApplicationRebalanceListener rebalanceListener, ApplicationRecordHandler recordHandler, - ApplicationProperties properties) + KafkaProperties kafkaProperties, + ApplicationProperties applicationProperties) { return new EndlessConsumer<>( executor, - properties.getClientId(), - properties.getTopic(), + kafkaProperties.getClientId(), + applicationProperties.getTopic(), kafkaConsumer, rebalanceListener, recordHandler); @@ -80,17 +84,17 @@ public class ApplicationConfiguration } @Bean(destroyMethod = "close") - public KafkaConsumer kafkaConsumer(ApplicationProperties properties) + public KafkaConsumer kafkaConsumer(KafkaProperties kafkaProperties) { Properties props = new Properties(); - props.put("bootstrap.servers", properties.getBootstrapServer()); + props.put("bootstrap.servers", kafkaProperties.getBootstrapServers()); props.put("partition.assignment.strategy", "org.apache.kafka.clients.consumer.CooperativeStickyAssignor"); - props.put("group.id", properties.getGroupId()); - props.put("client.id", properties.getClientId()); + props.put("group.id", kafkaProperties.getConsumer().getGroupId()); + props.put("client.id", kafkaProperties.getClientId()); props.put("enable.auto.commit", false); - props.put("auto.offset.reset", properties.getAutoOffsetReset()); - props.put("metadata.max.age.ms", "1000"); + props.put("auto.offset.reset", kafkaProperties.getConsumer().getAutoOffsetReset()); + props.put("metadata.max.age.ms", kafkaProperties.getConsumer().getProperties().get("metadata.max.age.ms")); props.put("key.deserializer", StringDeserializer.class.getName()); props.put("value.deserializer", JsonDeserializer.class.getName()); props.put(JsonDeserializer.TRUSTED_PACKAGES, "de.juplo.kafka"); diff --git a/src/main/java/de/juplo/kafka/ApplicationProperties.java b/src/main/java/de/juplo/kafka/ApplicationProperties.java index f852c00..005460c 100644 --- a/src/main/java/de/juplo/kafka/ApplicationProperties.java +++ b/src/main/java/de/juplo/kafka/ApplicationProperties.java @@ -16,22 +16,8 @@ import java.time.Duration; @Setter public class ApplicationProperties { - @NotNull - @NotEmpty - private String bootstrapServer; - @NotNull - @NotEmpty - private String groupId; - @NotNull - @NotEmpty - private String clientId; @NotNull @NotEmpty private String topic; - @NotNull - @NotEmpty - private String autoOffsetReset; - @NotNull - private Duration commitInterval; private Duration throttle; } diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index 26948f5..a899340 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -1,11 +1,6 @@ sumup: adder: - bootstrap-server: :9092 - group-id: my-group - client-id: DEV topic: out - auto-offset-reset: earliest - commit-interval: 5s management: endpoint: shutdown: @@ -21,16 +16,25 @@ management: enabled: true info: kafka: - bootstrap-server: ${consumer.bootstrap-server} - client-id: ${consumer.client-id} - group-id: ${consumer.group-id} + bootstrap-server: ${spring.kafka.consumer.bootstrap-servers} + client-id: ${spring.kafka.consumer.client-id} + group-id: ${spring.kafka.consumer.group-id} topic: ${consumer.topic} - auto-offset-reset: ${consumer.auto-offset-reset} + auto-offset-reset: ${spring.kafka.consumer.auto-offset-reset} spring: data: mongodb: uri: mongodb://juplo:training@localhost:27017 database: juplo + kafka: + bootstrap-servers: :9092 + client-id: DEV + consumer: + group-id: my-group + auto-offset-reset: earliest + auto-commit-interval: 5s + properties: + metadata.max.age.ms: 1000 logging: level: root: INFO diff --git a/src/test/java/de/juplo/kafka/ApplicationIT.java b/src/test/java/de/juplo/kafka/ApplicationIT.java index dcac79b..4bb4f5b 100644 --- a/src/test/java/de/juplo/kafka/ApplicationIT.java +++ b/src/test/java/de/juplo/kafka/ApplicationIT.java @@ -14,7 +14,7 @@ import static de.juplo.kafka.ApplicationIT.TOPIC; @SpringBootTest( webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT, properties = { - "sumup.adder.bootstrap-server=${spring.embedded.kafka.brokers}", + "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}", "sumup.adder.topic=" + TOPIC, "spring.mongodb.embedded.version=4.4.13" }) @EmbeddedKafka(topics = TOPIC) diff --git a/src/test/java/de/juplo/kafka/GenericApplicationTests.java b/src/test/java/de/juplo/kafka/GenericApplicationTests.java index a8fa7ea..0e05dbe 100644 --- a/src/test/java/de/juplo/kafka/GenericApplicationTests.java +++ b/src/test/java/de/juplo/kafka/GenericApplicationTests.java @@ -13,6 +13,7 @@ import org.apache.kafka.common.utils.Bytes; import org.junit.jupiter.api.*; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.autoconfigure.kafka.KafkaProperties; import org.springframework.boot.autoconfigure.mongo.MongoProperties; import org.springframework.boot.test.autoconfigure.data.mongo.AutoConfigureDataMongo; import org.springframework.boot.test.context.ConfigDataApplicationContextInitializer; @@ -39,9 +40,9 @@ import static org.awaitility.Awaitility.*; @SpringJUnitConfig(initializers = ConfigDataApplicationContextInitializer.class) @TestPropertySource( properties = { - "sumup.adder.bootstrap-server=${spring.embedded.kafka.brokers}", + "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}", "sumup.adder.topic=" + TOPIC, - "sumup.adder.commit-interval=500ms", + "spring.kafka.consumer.auto-commit-interval=500ms", "spring.mongodb.embedded.version=4.4.13" }) @EmbeddedKafka(topics = TOPIC, partitions = PARTITIONS) @EnableAutoConfiguration @@ -58,7 +59,9 @@ abstract class GenericApplicationTests @Autowired Consumer> consumer; @Autowired - ApplicationProperties properties; + ApplicationProperties applicationProperties; + @Autowired + KafkaProperties kafkaProperties; @Autowired ExecutorService executor; @Autowired @@ -337,16 +340,16 @@ abstract class GenericApplicationTests { Properties props; props = new Properties(); - props.put("bootstrap.servers", properties.getBootstrapServer()); + props.put("bootstrap.servers", kafkaProperties.getBootstrapServers()); props.put("linger.ms", 100); props.put("key.serializer", BytesSerializer.class.getName()); props.put("value.serializer", BytesSerializer.class.getName()); testRecordProducer = new KafkaProducer<>(props); props = new Properties(); - props.put("bootstrap.servers", properties.getBootstrapServer()); + props.put("bootstrap.servers", kafkaProperties.getBootstrapServers()); props.put("client.id", "OFFSET-CONSUMER"); - props.put("group.id", properties.getGroupId()); + props.put("group.id", kafkaProperties.getConsumer().getGroupId()); props.put("key.deserializer", BytesDeserializer.class.getName()); props.put("value.deserializer", BytesDeserializer.class.getName()); offsetConsumer = new KafkaConsumer<>(props); @@ -380,8 +383,8 @@ abstract class GenericApplicationTests endlessConsumer = new EndlessConsumer<>( executor, - properties.getClientId(), - properties.getTopic(), + kafkaProperties.getClientId(), + applicationProperties.getTopic(), kafkaConsumer, rebalanceListener, captureOffsetAndExecuteTestHandler);