diff --git a/spring-cloud-sleuth-autoconfigure/src/main/resources/META-INF/spring.factories b/spring-cloud-sleuth-autoconfigure/src/main/resources/META-INF/spring.factories index 7d7e98a50..7b8848a99 100644 --- a/spring-cloud-sleuth-autoconfigure/src/main/resources/META-INF/spring.factories +++ b/spring-cloud-sleuth-autoconfigure/src/main/resources/META-INF/spring.factories @@ -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,\ diff --git a/spring-cloud-sleuth-autoconfigure/src/test/java/org/springframework/cloud/sleuth/autoconfig/instrument/kafka/TraceKafkaAutoConfigurationTests.java b/spring-cloud-sleuth-autoconfigure/src/test/java/org/springframework/cloud/sleuth/autoconfig/instrument/kafka/TraceKafkaAutoConfigurationTests.java new file mode 100644 index 000000000..ea6560e38 --- /dev/null +++ b/spring-cloud-sleuth-autoconfigure/src/test/java/org/springframework/cloud/sleuth/autoconfig/instrument/kafka/TraceKafkaAutoConfigurationTests.java @@ -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 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 kafkaProducer = new TracingKafkaProducer<>(new MockProducer<>(), null); + this.contextRunner.withBean(TracingKafkaProducer.class, () -> kafkaProducer) + .run(context -> assertThat(context).getBean(Producer.class).isEqualTo(kafkaProducer)); + } + +} diff --git a/spring-cloud-sleuth-autoconfigure/src/test/java/org/springframework/cloud/sleuth/autoconfig/instrument/kafka/TraceReactorKafkaAutoConfigurationTests.java b/spring-cloud-sleuth-autoconfigure/src/test/java/org/springframework/cloud/sleuth/autoconfig/instrument/kafka/TraceReactorKafkaAutoConfigurationTests.java new file mode 100644 index 000000000..4386651e8 --- /dev/null +++ b/spring-cloud-sleuth-autoconfigure/src/test/java/org/springframework/cloud/sleuth/autoconfig/instrument/kafka/TraceReactorKafkaAutoConfigurationTests.java @@ -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)); + } + +} diff --git a/tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/kafka/KafkaConsumerTest.java b/tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/kafka/KafkaConsumerTest.java index fbe5cca26..46a7aaff3 100644 --- a/tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/kafka/KafkaConsumerTest.java +++ b/tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/kafka/KafkaConsumerTest.java @@ -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); diff --git a/tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/kafka/KafkaProducerTest.java b/tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/kafka/KafkaProducerTest.java index 78f9e7a6e..79b25ed2b 100644 --- a/tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/kafka/KafkaProducerTest.java +++ b/tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/kafka/KafkaProducerTest.java @@ -46,7 +46,7 @@ public class KafkaProducerTest extends org.springframework.cloud.sleuth.instrume startKafkaConsumer(); this.kafkaProducer.send(producerRecord); - ConsumerRecord consumerRecord = consumerRecords.poll(5, TimeUnit.SECONDS); + ConsumerRecord consumerRecord = consumerRecords.poll(15, TimeUnit.SECONDS); BDDAssertions.then(consumerRecord).isNotNull(); BDDAssertions.then(getHeaderValueOrNull(consumerRecord, "X-B3-TraceId")).isNotNull(); diff --git a/tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/kafka/KafkaReceiverTest.java b/tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/kafka/KafkaReceiverTest.java index 5a3f14d75..6b26069bf 100644 --- a/tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/kafka/KafkaReceiverTest.java +++ b/tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/kafka/KafkaReceiverTest.java @@ -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); diff --git a/tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/kafka/KafkaSenderTest.java b/tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/kafka/KafkaSenderTest.java index fb88a5b99..52b763679 100644 --- a/tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/kafka/KafkaSenderTest.java +++ b/tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/kafka/KafkaSenderTest.java @@ -53,7 +53,7 @@ public class KafkaSenderTest extends org.springframework.cloud.sleuth.instrument Flux> senderResultFlux = this.kafkaSender .send(Mono.just(SenderRecord.create(producerRecord, null))); StepVerifier.create(senderResultFlux).expectNextCount(1).verifyComplete(); - ConsumerRecord consumerRecord = consumerRecords.poll(5, TimeUnit.SECONDS); + ConsumerRecord consumerRecord = consumerRecords.poll(15, TimeUnit.SECONDS); BDDAssertions.then(consumerRecord).isNotNull(); BDDAssertions.then(getHeaderValueOrNull(consumerRecord, "X-B3-TraceId")).isNotNull(); diff --git a/tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/kafka/TracingKafkaAutoConfigurationTest.java b/tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/kafka/TracingKafkaAutoConfigurationTest.java new file mode 100644 index 000000000..a0d0dd3cb --- /dev/null +++ b/tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/kafka/TracingKafkaAutoConfigurationTest.java @@ -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 kafkaConsumer; + + @Autowired + Producer 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 kafkaConsumer() { + return new MockConsumer<>(OffsetResetStrategy.NONE); + } + + @Bean + Producer kafkaProducer() { + return new MockProducer<>(); + } + + } + +} diff --git a/tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/kafka/TracingReactorKafkaAutoConfigurationTest.java b/tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/kafka/TracingReactorKafkaAutoConfigurationTest.java new file mode 100644 index 000000000..4992bc281 --- /dev/null +++ b/tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/kafka/TracingReactorKafkaAutoConfigurationTest.java @@ -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 { + + } + +} diff --git a/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/KafkaConsumerTest.java b/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/KafkaConsumerTest.java index cd84d04b3..9dc76c744 100644 --- a/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/KafkaConsumerTest.java +++ b/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/KafkaConsumerTest.java @@ -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); diff --git a/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/KafkaProducerTest.java b/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/KafkaProducerTest.java index f399eb190..8efc776be 100644 --- a/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/KafkaProducerTest.java +++ b/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/KafkaProducerTest.java @@ -129,7 +129,7 @@ public abstract class KafkaProducerTest implements TestTracingAwareSupplier { ProducerRecord 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); diff --git a/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/KafkaReceiverTest.java b/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/KafkaReceiverTest.java index e33f3eb7c..24a700620 100644 --- a/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/KafkaReceiverTest.java +++ b/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/KafkaReceiverTest.java @@ -130,7 +130,7 @@ public abstract class KafkaReceiverTest implements TestTracingAwareSupplier { ProducerRecord 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); diff --git a/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/KafkaSenderTest.java b/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/KafkaSenderTest.java index 67db8e9cc..896132dbe 100644 --- a/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/KafkaSenderTest.java +++ b/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/KafkaSenderTest.java @@ -134,7 +134,7 @@ public abstract class KafkaSenderTest implements TestTracingAwareSupplier { Flux> 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);