From 7adff476ad862d30d668d75212d1ca1c7cf16b03 Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Fri, 22 Apr 2022 11:24:55 +0200 Subject: [PATCH] =?utf8?q?Springify:=20Der=20Kafka-`Consumer`=20wird=20?= =?utf8?q?=C3=BCber=20die=20Spring-Factory=20erzeugt?= MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit * Per `git cherry-pick` aus `springified-consumer--config' übernommen. * Conflicts: ** src/main/java/de/juplo/kafka/ApplicationConfiguration.java ** src/test/java/de/juplo/kafka/ApplicationTests.java * Damit Spring Kafka den Consumer instanziieren kann, musste insbesondere noch der Teil der Konfiguration, der fix ist, aus der Konfig-Klasse `ApplicationConfiguration` in die YAML-Datei `application.yml` verschoben werden: ** Die Auswahl des `StickyAssignor` als Partition-Assignment-Strategy ** Die Konfiguration der Deserialisierer --- .../juplo/kafka/ApplicationConfiguration.java | 29 ++++--------------- src/main/resources/application.yml | 6 ++++ .../juplo/kafka/GenericApplicationTests.java | 13 ++++++--- 3 files changed, 21 insertions(+), 27 deletions(-) diff --git a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java index 523707f..bae5d51 100644 --- a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java +++ b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java @@ -1,15 +1,14 @@ package de.juplo.kafka; -import org.apache.kafka.clients.consumer.KafkaConsumer; -import org.apache.kafka.common.serialization.StringDeserializer; +import org.apache.kafka.clients.consumer.Consumer; 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.support.serializer.JsonDeserializer; import java.util.Optional; -import java.util.Properties; +import org.springframework.kafka.core.ConsumerFactory; + import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -53,7 +52,7 @@ public class ApplicationConfiguration @Bean public EndlessConsumer endlessConsumer( - KafkaConsumer kafkaConsumer, + Consumer kafkaConsumer, ExecutorService executor, ApplicationRebalanceListener rebalanceListener, ApplicationRecordHandler recordHandler, @@ -77,24 +76,8 @@ public class ApplicationConfiguration } @Bean(destroyMethod = "close") - public KafkaConsumer kafkaConsumer(KafkaProperties kafkaProperties) + public Consumer kafkaConsumer(ConsumerFactory factory) { - Properties props = new Properties(); - - props.put("bootstrap.servers", kafkaProperties.getBootstrapServers()); - props.put("partition.assignment.strategy", "org.apache.kafka.clients.consumer.StickyAssignor"); - props.put("group.id", kafkaProperties.getConsumer().getGroupId()); - props.put("client.id", kafkaProperties.getClientId()); - props.put("auto.offset.reset", kafkaProperties.getConsumer().getAutoOffsetReset()); - props.put("auto.commit.interval.ms", (int)kafkaProperties.getConsumer().getAutoCommitInterval().toMillis()); - 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"); - props.put(JsonDeserializer.TYPE_MAPPINGS, - Message.Type.ADD + ":" + MessageAddNumber.class.getName() + "," + - Message.Type.CALC + ":" + MessageCalculateSum.class.getName()); - - return new KafkaConsumer<>(props); + return factory.createConsumer(); } } diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index a899340..92f3a6b 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -33,8 +33,14 @@ spring: group-id: my-group auto-offset-reset: earliest auto-commit-interval: 5s + key-deserializer: org.apache.kafka.common.serialization.StringDeserializer + value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer properties: + partition.assignment.strategy: org.apache.kafka.clients.consumer.StickyAssignor metadata.max.age.ms: 1000 + spring.json.type.mapping: > + ADD:de.juplo.kafka.MessageAddNumber, + CALC:de.juplo.kafka.MessageCalculateSum logging: level: root: INFO diff --git a/src/test/java/de/juplo/kafka/GenericApplicationTests.java b/src/test/java/de/juplo/kafka/GenericApplicationTests.java index 869b5d9..21c3f7f 100644 --- a/src/test/java/de/juplo/kafka/GenericApplicationTests.java +++ b/src/test/java/de/juplo/kafka/GenericApplicationTests.java @@ -14,6 +14,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.KafkaAutoConfiguration; import org.springframework.boot.autoconfigure.kafka.KafkaProperties; import org.springframework.boot.autoconfigure.mongo.MongoProperties; import org.springframework.boot.test.autoconfigure.data.mongo.AutoConfigureDataMongo; @@ -38,7 +39,11 @@ import static org.assertj.core.api.Assertions.*; import static org.awaitility.Awaitility.*; -@SpringJUnitConfig(initializers = ConfigDataApplicationContextInitializer.class) +@SpringJUnitConfig( + initializers = ConfigDataApplicationContextInitializer.class, + classes = { + KafkaAutoConfiguration.class, + ApplicationTests.Configuration.class }) @TestPropertySource( properties = { "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}", @@ -56,13 +61,13 @@ abstract class GenericApplicationTests @Autowired - KafkaConsumer kafkaConsumer; + org.apache.kafka.clients.consumer.Consumer kafkaConsumer; @Autowired Consumer> consumer; @Autowired ApplicationProperties applicationProperties; - @Autowired - KafkaProperties kafkaProperties; + @Autowired + KafkaProperties kafkaProperties; @Autowired ExecutorService executor; @Autowired -- 2.20.1