SpringCloud/spring-cloud-sleuth
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Added reactor-kafka autoconfig to spring.factories (#1943)

Browse Source
This commit is contained in:
Flaviu Mureșan
2021-05-08 14:14:57 +02:00
committed by GitHub
parent ac729f141d
commit 4070712f11
13 changed files with 266 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
spring-cloud-sleuth-autoconfigure/src/main/resources/META-INF/spring.factories
View File

@@ -1,6 +1,7 @@
# Auto Configuration
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
org.springframework.cloud.sleuth.autoconfig.instrument.kafka.TracingKafkaAutoConfiguration,\
org.springframework.cloud.sleuth.autoconfig.instrument.kafka.TracingReactorKafkaAutoConfiguration,\
org.springframework.cloud.sleuth.autoconfig.instrument.async.TraceAsyncAutoConfiguration,\
org.springframework.cloud.sleuth.autoconfig.instrument.async.TraceAsyncCustomAutoConfiguration,\
org.springframework.cloud.sleuth.autoconfig.instrument.async.TraceAsyncDefaultAutoConfiguration,\

76
spring-cloud-sleuth-autoconfigure/src/test/java/org/springframework/cloud/sleuth/autoconfig/instrument/kafka/TraceKafkaAutoConfigurationTests.java Normal file
View File

@@ -0,0 +1,76 @@
/*
* Copyright 2013-2021 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.sleuth.autoconfig.instrument.kafka;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.MockConsumer;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
import org.apache.kafka.clients.producer.MockProducer;
import org.apache.kafka.clients.producer.Producer;
import org.junit.jupiter.api.Test;
import org.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import org.springframework.cloud.sleuth.autoconfig.TraceNoOpAutoConfiguration;
import org.springframework.cloud.sleuth.instrument.kafka.TracingKafkaConsumer;
import org.springframework.cloud.sleuth.instrument.kafka.TracingKafkaProducer;
import org.springframework.cloud.sleuth.instrument.kafka.TracingKafkaPropagatorGetter;
import org.springframework.cloud.sleuth.instrument.kafka.TracingKafkaPropagatorSetter;
import static org.assertj.core.api.Assertions.assertThat;
class TraceKafkaAutoConfigurationTests {
private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
.withPropertyValues("spring.sleuth.noop.enabled=true")
.withConfiguration(AutoConfigurations.of(TraceNoOpAutoConfiguration.class,
TracingKafkaAutoConfiguration.class, TracingReactorKafkaAutoConfiguration.class));
@Test
void should_inject_beans_for_getter_setter_kafka_propagation() {
this.contextRunner.run(context -> assertThat(context).hasSingleBean(TracingKafkaPropagatorGetter.class)
.hasSingleBean(TracingKafkaPropagatorSetter.class));
}
@Test
void should_decorate_kafka_producer() {
this.contextRunner.withBean(Producer.class, MockProducer::new)
.run(context -> assertThat(context).hasSingleBean(TracingKafkaProducer.class));
}
@Test
void should_decorate_kafka_consumer() {
this.contextRunner.withBean(Consumer.class, () -> new MockConsumer<>(OffsetResetStrategy.NONE))
.run(context -> assertThat(context).hasSingleBean(TracingKafkaConsumer.class));
}
@Test
void should_not_decorate_tracing_kafka_consumer() {
TracingKafkaConsumer<String, String> kafkaConsumer = new TracingKafkaConsumer<>(
new MockConsumer<>(OffsetResetStrategy.NONE), null);
this.contextRunner.withBean(TracingKafkaConsumer.class, () -> kafkaConsumer)
.run(context -> assertThat(context).getBean(Consumer.class).isEqualTo(kafkaConsumer));
}
@Test
void should_not_decorate_tracing_kafka_producer() {
TracingKafkaProducer<String, String> kafkaProducer = new TracingKafkaProducer<>(new MockProducer<>(), null);
this.contextRunner.withBean(TracingKafkaProducer.class, () -> kafkaProducer)
.run(context -> assertThat(context).getBean(Producer.class).isEqualTo(kafkaProducer));
}
}

51
spring-cloud-sleuth-autoconfigure/src/test/java/org/springframework/cloud/sleuth/autoconfig/instrument/kafka/TraceReactorKafkaAutoConfigurationTests.java Normal file
View File

@@ -0,0 +1,51 @@
/*
* Copyright 2013-2021 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.sleuth.autoconfig.instrument.kafka;
import org.junit.jupiter.api.Test;
import reactor.kafka.receiver.KafkaReceiver;
import org.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.test.context.FilteredClassLoader;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import org.springframework.cloud.sleuth.autoconfig.TraceNoOpAutoConfiguration;
import org.springframework.cloud.sleuth.instrument.kafka.TracingKafkaConsumerFactory;
import org.springframework.cloud.sleuth.instrument.kafka.TracingKafkaProducerFactory;
import static org.assertj.core.api.Assertions.assertThat;
class TraceReactorKafkaAutoConfigurationTests {
private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
.withPropertyValues("spring.sleuth.noop.enabled=true")
.withConfiguration(AutoConfigurations.of(TraceNoOpAutoConfiguration.class,
TracingKafkaAutoConfiguration.class, TracingReactorKafkaAutoConfiguration.class));
@Test
void should_not_create_factories_when_reactor_kafka_not_on_classpath() {
this.contextRunner.withClassLoader(new FilteredClassLoader(KafkaReceiver.class))
.run(context -> assertThat(context).doesNotHaveBean(TracingKafkaProducerFactory.class)
.doesNotHaveBean(TracingKafkaConsumerFactory.class));
}
@Test
void should_create_factories_when_reactor_kafka_on_classpath() {
this.contextRunner.run(context -> assertThat(context).hasSingleBean(TracingKafkaProducerFactory.class)
.hasSingleBean(TracingKafkaConsumerFactory.class));
}
}

2
tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/kafka/KafkaConsumerTest.java
View File

@@ -52,7 +52,7 @@ public class KafkaConsumerTest extends org.springframework.cloud.sleuth.instrume
kafkaProducer.send(producerRecord);
kafkaProducer.close();
await().atMost(Duration.ofSeconds(5)).until(() -> receivedCounter.intValue() == 1);
await().atMost(Duration.ofSeconds(15)).until(() -> receivedCounter.intValue() == 1);
BDDAssertions.then(this.tracer.currentSpan()).isNull();
BDDAssertions.then(this.spans).hasSize(1);

2
tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/kafka/KafkaProducerTest.java
View File

@@ -46,7 +46,7 @@ public class KafkaProducerTest extends org.springframework.cloud.sleuth.instrume
startKafkaConsumer();
this.kafkaProducer.send(producerRecord);
ConsumerRecord<String, String> consumerRecord = consumerRecords.poll(5, TimeUnit.SECONDS);
ConsumerRecord<String, String> consumerRecord = consumerRecords.poll(15, TimeUnit.SECONDS);
BDDAssertions.then(consumerRecord).isNotNull();
BDDAssertions.then(getHeaderValueOrNull(consumerRecord, "X-B3-TraceId")).isNotNull();

2
tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/kafka/KafkaReceiverTest.java
View File

@@ -51,7 +51,7 @@ public class KafkaReceiverTest extends org.springframework.cloud.sleuth.instrume
producerRecord.headers().add("b3", "000000000000000a-000000000000000b-1-000000000000000a".getBytes());
kafkaProducer.send(producerRecord);
await().atMost(Duration.ofSeconds(5)).until(() -> receivedCounter.intValue() == 1);
await().atMost(Duration.ofSeconds(15)).until(() -> receivedCounter.intValue() == 1);
BDDAssertions.then(this.tracer.currentSpan()).isNull();
BDDAssertions.then(this.spans).hasSize(1);

2
tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/kafka/KafkaSenderTest.java
View File

@@ -53,7 +53,7 @@ public class KafkaSenderTest extends org.springframework.cloud.sleuth.instrument
Flux<SenderResult<Object>> senderResultFlux = this.kafkaSender
.send(Mono.just(SenderRecord.create(producerRecord, null)));
StepVerifier.create(senderResultFlux).expectNextCount(1).verifyComplete();
ConsumerRecord<String, String> consumerRecord = consumerRecords.poll(5, TimeUnit.SECONDS);
ConsumerRecord<String, String> consumerRecord = consumerRecords.poll(15, TimeUnit.SECONDS);
BDDAssertions.then(consumerRecord).isNotNull();
BDDAssertions.then(getHeaderValueOrNull(consumerRecord, "X-B3-TraceId")).isNotNull();

74
tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/kafka/TracingKafkaAutoConfigurationTest.java Normal file
View File

@@ -0,0 +1,74 @@
/*
* Copyright 2013-2021 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.sleuth.brave.instrument.kafka;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.MockConsumer;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
import org.apache.kafka.clients.producer.MockProducer;
import org.apache.kafka.clients.producer.Producer;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.cloud.sleuth.instrument.kafka.TracingKafkaConsumer;
import org.springframework.cloud.sleuth.instrument.kafka.TracingKafkaProducer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import static org.assertj.core.api.BDDAssertions.then;
@SpringBootTest(classes = TracingKafkaAutoConfigurationTest.Config.class,
webEnvironment = SpringBootTest.WebEnvironment.NONE)
public class TracingKafkaAutoConfigurationTest {
@Autowired
Consumer<String, String> kafkaConsumer;
@Autowired
Producer<String, String> kafkaProducer;
@Test
public void should_wrap_kafka_consumer() {
then(this.kafkaConsumer).isNotNull();
then(this.kafkaConsumer).isInstanceOf(TracingKafkaConsumer.class);
}
@Test
public void should_wrap_kafka_producer() {
then(this.kafkaProducer).isNotNull();
then(this.kafkaProducer).isInstanceOf(TracingKafkaProducer.class);
}
@Configuration(proxyBeanMethods = false)
@EnableAutoConfiguration
protected static class Config {
@Bean
Consumer<String, String> kafkaConsumer() {
return new MockConsumer<>(OffsetResetStrategy.NONE);
}
@Bean
Producer<String, String> kafkaProducer() {
return new MockProducer<>();
}
}
}

56
tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/kafka/TracingReactorKafkaAutoConfigurationTest.java Normal file
View File

@@ -0,0 +1,56 @@
/*
* Copyright 2013-2021 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.sleuth.brave.instrument.kafka;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.cloud.sleuth.instrument.kafka.TracingKafkaConsumerFactory;
import org.springframework.cloud.sleuth.instrument.kafka.TracingKafkaProducerFactory;
import org.springframework.context.annotation.Configuration;
import static org.assertj.core.api.BDDAssertions.then;
@SpringBootTest(classes = TracingReactorKafkaAutoConfigurationTest.Config.class,
webEnvironment = SpringBootTest.WebEnvironment.NONE)
public class TracingReactorKafkaAutoConfigurationTest {
@Autowired
TracingKafkaConsumerFactory kafkaConsumerFactory;
@Autowired
TracingKafkaProducerFactory kafkaProducerFactory;
@Test
public void should_register_consumer_factory() {
then(this.kafkaConsumerFactory).isNotNull();
}
@Test
public void should_register_producer_factory() {
then(this.kafkaProducerFactory).isNotNull();
}
@Configuration(proxyBeanMethods = false)
@EnableAutoConfiguration
protected static class Config {
}
}

2
tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/KafkaConsumerTest.java
View File

@@ -128,7 +128,7 @@ public abstract class KafkaConsumerTest implements TestTracingAwareSupplier {
kafkaProducer.send(producerRecord);
kafkaProducer.close();
await().atMost(Duration.ofSeconds(5)).until(() -> receivedCounter.intValue() == 1);
await().atMost(Duration.ofSeconds(15)).until(() -> receivedCounter.intValue() == 1);
BDDAssertions.then(this.tracer.currentSpan()).isNull();
BDDAssertions.then(this.spans).hasSize(1);

2
tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/KafkaProducerTest.java
View File

@@ -129,7 +129,7 @@ public abstract class KafkaProducerTest implements TestTracingAwareSupplier {
ProducerRecord<String, String> producerRecord = new ProducerRecord<>(testTopic, "test", "test");
this.kafkaProducer.send(producerRecord, callback);
await().atMost(Duration.ofSeconds(5)).until(acknowledged::get);
await().atMost(Duration.ofSeconds(15)).until(acknowledged::get);
BDDAssertions.then(this.tracer.currentSpan()).isNull();
BDDAssertions.then(this.spans).hasSize(1);

2
tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/KafkaReceiverTest.java
View File

@@ -130,7 +130,7 @@ public abstract class KafkaReceiverTest implements TestTracingAwareSupplier {
ProducerRecord<String, String> producerRecord = new ProducerRecord<>(testTopic, "test", "test");
kafkaProducer.send(producerRecord);
await().atMost(Duration.ofSeconds(5)).until(() -> receivedCounter.intValue() == 1);
await().atMost(Duration.ofSeconds(15)).until(() -> receivedCounter.intValue() == 1);
BDDAssertions.then(this.tracer.currentSpan()).isNull();
BDDAssertions.then(this.spans).hasSize(1);

2
tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/KafkaSenderTest.java
View File

@@ -134,7 +134,7 @@ public abstract class KafkaSenderTest implements TestTracingAwareSupplier {
Flux<SenderResult<Object>> senderResultFlux = this.kafkaSender
.send(Mono.just(SenderRecord.create(producerRecord, null)));
StepVerifier.create(senderResultFlux).expectNextCount(1).verifyComplete();
consumerRecords.poll(5, TimeUnit.SECONDS);
consumerRecords.poll(15, TimeUnit.SECONDS);
BDDAssertions.then(this.tracer.currentSpan()).isNull();
BDDAssertions.then(this.spans).hasSize(1);