X-Git-Url: http://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Ftest%2Fjava%2Fde%2Fjuplo%2Fkafka%2FApplicationTests.java;h=d446bbed8a922c5542beb86bc28a04f1c4a5080a;hb=d8f4e839b57b8405ca909b5472a8d4639937a101;hp=40dc149e519a7f95709d7a7966dddf5b8de722df;hpb=ac56a4e914c4a7d687c8c85b2608f0e397372d56;p=demos%2Fkafka%2Ftraining diff --git a/src/test/java/de/juplo/kafka/ApplicationTests.java b/src/test/java/de/juplo/kafka/ApplicationTests.java index 40dc149..d446bbe 100644 --- a/src/test/java/de/juplo/kafka/ApplicationTests.java +++ b/src/test/java/de/juplo/kafka/ApplicationTests.java @@ -11,6 +11,7 @@ import org.apache.kafka.common.serialization.*; import org.apache.kafka.common.utils.Bytes; import org.junit.jupiter.api.*; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.kafka.KafkaProperties; import org.springframework.boot.test.context.ConfigDataApplicationContextInitializer; import org.springframework.boot.test.context.TestConfiguration; import org.springframework.context.annotation.Bean; @@ -39,7 +40,7 @@ import static org.awaitility.Awaitility.*; @TestMethodOrder(MethodOrderer.OrderAnnotation.class) @TestPropertySource( properties = { - "consumer.bootstrap-server=${spring.embedded.kafka.brokers}", + "spring.kafka.consumer.bootstrap-servers=${spring.embedded.kafka.brokers}", "consumer.topic=" + TOPIC }) @EmbeddedKafka(topics = TOPIC, partitions = PARTITIONS) @Slf4j @@ -60,7 +61,9 @@ class ApplicationTests @Autowired KafkaConsumer offsetConsumer; @Autowired - ApplicationProperties properties; + ApplicationProperties applicationProperties; + @Autowired + KafkaProperties kafkaProperties; @Autowired ExecutorService executor; @@ -255,8 +258,8 @@ class ApplicationTests endlessConsumer = new EndlessConsumer<>( executor, - properties.getClientId(), - properties.getTopic(), + kafkaProperties.getConsumer().getClientId(), + applicationProperties.getTopic(), kafkaConsumer, captureOffsetAndExecuteTestHandler); @@ -288,10 +291,10 @@ class ApplicationTests } @Bean - KafkaProducer kafkaProducer(ApplicationProperties properties) + KafkaProducer kafkaProducer(KafkaProperties properties) { Properties props = new Properties(); - props.put("bootstrap.servers", properties.getBootstrapServer()); + props.put("bootstrap.servers", properties.getConsumer().getBootstrapServers()); props.put("linger.ms", 100); props.put("key.serializer", StringSerializer.class.getName()); props.put("value.serializer", BytesSerializer.class.getName()); @@ -300,12 +303,12 @@ class ApplicationTests } @Bean - KafkaConsumer offsetConsumer(ApplicationProperties properties) + KafkaConsumer offsetConsumer(KafkaProperties properties) { Properties props = new Properties(); - props.put("bootstrap.servers", properties.getBootstrapServer()); + props.put("bootstrap.servers", properties.getConsumer().getBootstrapServers()); props.put("client.id", "OFFSET-CONSUMER"); - props.put("group.id", properties.getGroupId()); + props.put("group.id", properties.getConsumer().getGroupId()); props.put("key.deserializer", BytesDeserializer.class.getName()); props.put("value.deserializer", BytesDeserializer.class.getName());