diff --git a/docs/src/main/asciidoc/integrations.adoc b/docs/src/main/asciidoc/integrations.adoc index b3dfadea7..d550f365a 100644 --- a/docs/src/main/asciidoc/integrations.adoc +++ b/docs/src/main/asciidoc/integrations.adoc @@ -24,6 +24,8 @@ KafkaReceiver reactiveKafkaReceiver(TracingKafkaConsumerFactory } ---- +Additionally, we decorate any https://docs.spring.io/spring-kafka/docs/current/reference/html/[Spring Kafka] `ProducerFactory` and `ConsumerFactory` available in the context. However, this is disabled if Brave instrumentation is on the classpath. + [[sleuth-async-integration]] == Asynchronous Communication diff --git a/spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/instrument/kafka/SpringKafkaAutoConfiguration.java b/spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/instrument/kafka/SpringKafkaAutoConfiguration.java new file mode 100644 index 000000000..415e4a926 --- /dev/null +++ b/spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/instrument/kafka/SpringKafkaAutoConfiguration.java @@ -0,0 +1,52 @@ +/* + * 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.springframework.beans.factory.BeanFactory; +import org.springframework.boot.autoconfigure.AutoConfigureAfter; +import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingClass; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.cloud.sleuth.Tracer; +import org.springframework.cloud.sleuth.autoconfig.brave.BraveAutoConfiguration; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.kafka.core.ProducerFactory; + +/** + * {@link org.springframework.boot.autoconfigure.EnableAutoConfiguration + * Auto-configuration} that registers instrumentation for Spring Kafka. + * + * @author Anders Clausen + * @author Flaviu Muresan + * @since 3.1.0 + */ +@Configuration(proxyBeanMethods = false) +@ConditionalOnClass(ProducerFactory.class) +@ConditionalOnMissingClass("brave.kafka.clients.KafkaTracing") +@ConditionalOnBean(Tracer.class) +@AutoConfigureAfter(BraveAutoConfiguration.class) +@ConditionalOnProperty(value = "spring.sleuth.kafka.enabled", matchIfMissing = true) +public class SpringKafkaAutoConfiguration { + + @Bean + static SpringKafkaFactoryBeanPostProcessor springKafkaFactoryBeanPostProcessor(BeanFactory beanFactory) { + return new SpringKafkaFactoryBeanPostProcessor(beanFactory); + } + +} diff --git a/spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/instrument/kafka/SpringKafkaConsumerPostProcessor.java b/spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/instrument/kafka/SpringKafkaConsumerPostProcessor.java new file mode 100644 index 000000000..f57bbe494 --- /dev/null +++ b/spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/instrument/kafka/SpringKafkaConsumerPostProcessor.java @@ -0,0 +1,38 @@ +/* + * 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.springframework.beans.factory.BeanFactory; +import org.springframework.cloud.sleuth.instrument.kafka.TracingKafkaConsumer; +import org.springframework.kafka.core.ConsumerPostProcessor; + +class SpringKafkaConsumerPostProcessor implements ConsumerPostProcessor { + + private final BeanFactory beanFactory; + + SpringKafkaConsumerPostProcessor(BeanFactory beanFactory) { + this.beanFactory = beanFactory; + } + + @Override + public Consumer apply(Consumer kvConsumer) { + return new TracingKafkaConsumer<>(kvConsumer, beanFactory); + } + +} diff --git a/spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/instrument/kafka/SpringKafkaFactoryBeanPostProcessor.java b/spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/instrument/kafka/SpringKafkaFactoryBeanPostProcessor.java new file mode 100644 index 000000000..20dae307f --- /dev/null +++ b/spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/instrument/kafka/SpringKafkaFactoryBeanPostProcessor.java @@ -0,0 +1,58 @@ +/* + * 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.springframework.beans.BeansException; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.config.BeanPostProcessor; +import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.kafka.core.ProducerFactory; + +/** + * Bean post processor for {@link org.springframework.kafka.core.ProducerFactory} and + * {@link org.springframework.kafka.core.ConsumerFactory}. + * + * @author Anders Clausen + * @author Flaviu Muresan + * @since 3.1.0 + */ +public class SpringKafkaFactoryBeanPostProcessor implements BeanPostProcessor { + + private final BeanFactory beanFactory; + + public SpringKafkaFactoryBeanPostProcessor(BeanFactory beanFactory) { + this.beanFactory = beanFactory; + } + + @Override + public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { + if (bean instanceof ConsumerFactory) { + ConsumerFactory factory = (ConsumerFactory) bean; + if (factory.getPostProcessors().stream().noneMatch(o -> o instanceof SpringKafkaConsumerPostProcessor)) { + factory.addPostProcessor(new SpringKafkaConsumerPostProcessor(this.beanFactory)); + } + } + else if (bean instanceof ProducerFactory) { + ProducerFactory factory = (ProducerFactory) bean; + if (factory.getPostProcessors().stream().noneMatch(o -> o instanceof SpringKafkaProducerPostProcessor)) { + factory.addPostProcessor(new SpringKafkaProducerPostProcessor(this.beanFactory)); + } + } + return bean; + } + +} diff --git a/spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/instrument/kafka/SpringKafkaProducerPostProcessor.java b/spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/instrument/kafka/SpringKafkaProducerPostProcessor.java new file mode 100644 index 000000000..fa514664b --- /dev/null +++ b/spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/instrument/kafka/SpringKafkaProducerPostProcessor.java @@ -0,0 +1,38 @@ +/* + * 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.producer.Producer; + +import org.springframework.beans.factory.BeanFactory; +import org.springframework.cloud.sleuth.instrument.kafka.TracingKafkaProducer; +import org.springframework.kafka.core.ProducerPostProcessor; + +class SpringKafkaProducerPostProcessor implements ProducerPostProcessor { + + private final BeanFactory beanFactory; + + SpringKafkaProducerPostProcessor(BeanFactory beanFactory) { + this.beanFactory = beanFactory; + } + + @Override + public Producer apply(Producer kvProducer) { + return new TracingKafkaProducer<>(kvProducer, beanFactory); + } + +} 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 d931b5211..1bdfee08b 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 @@ -3,6 +3,7 @@ org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ org.springframework.cloud.sleuth.autoconfig.actuate.TraceSleuthActuatorAutoConfiguration,\ org.springframework.cloud.sleuth.autoconfig.instrument.kafka.TracingKafkaAutoConfiguration,\ org.springframework.cloud.sleuth.autoconfig.instrument.kafka.TracingReactorKafkaAutoConfiguration,\ +org.springframework.cloud.sleuth.autoconfig.instrument.kafka.SpringKafkaAutoConfiguration,\ 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/SpringKafkaAutoConfigurationTests.java b/spring-cloud-sleuth-autoconfigure/src/test/java/org/springframework/cloud/sleuth/autoconfig/instrument/kafka/SpringKafkaAutoConfigurationTests.java new file mode 100644 index 000000000..62f3d8582 --- /dev/null +++ b/spring-cloud-sleuth-autoconfigure/src/test/java/org/springframework/cloud/sleuth/autoconfig/instrument/kafka/SpringKafkaAutoConfigurationTests.java @@ -0,0 +1,116 @@ +/* + * 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 java.util.ArrayList; +import java.util.List; + +import brave.kafka.clients.KafkaTracing; +import org.apache.kafka.clients.consumer.Consumer; +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.FilteredClassLoader; +import org.springframework.boot.test.context.runner.ApplicationContextRunner; +import org.springframework.cloud.sleuth.autoconfig.TraceNoOpAutoConfiguration; +import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.kafka.core.ConsumerPostProcessor; +import org.springframework.kafka.core.ProducerFactory; +import org.springframework.kafka.core.ProducerPostProcessor; + +import static org.assertj.core.api.Assertions.assertThat; + +class SpringKafkaAutoConfigurationTests { + + private final ApplicationContextRunner contextRunner = new ApplicationContextRunner() + .withPropertyValues("spring.sleuth.noop.enabled=true").withConfiguration( + AutoConfigurations.of(TraceNoOpAutoConfiguration.class, TracingKafkaAutoConfiguration.class, + TracingReactorKafkaAutoConfiguration.class, SpringKafkaAutoConfiguration.class)); + + @Test + void should_be_disabled_when_brave_on_classpath() { + this.contextRunner + .run((context) -> assertThat(context).doesNotHaveBean(SpringKafkaFactoryBeanPostProcessor.class)); + } + + @Test + void should_decorate_spring_kafka_producer_factory() { + this.contextRunner.withClassLoader(new FilteredClassLoader(KafkaTracing.class)) + .withBean(ProducerFactory.class, TestProducerFactory::new) + .run(context -> assertThat(context).getBean(ProducerFactory.class) + .extracting(ProducerFactory::getPostProcessors).matches(postProcessors -> postProcessors + .stream().filter(p -> p instanceof SpringKafkaProducerPostProcessor).count() == 1)); + } + + @Test + void should_decorate_spring_kafka_consumer_factory() { + this.contextRunner.withClassLoader(new FilteredClassLoader(KafkaTracing.class)) + .withBean(ConsumerFactory.class, TestConsumerFactory::new) + .run(context -> assertThat(context).getBean(ConsumerFactory.class) + .extracting(ConsumerFactory::getPostProcessors).matches(postProcessors -> postProcessors + .stream().filter(p -> p instanceof SpringKafkaConsumerPostProcessor).count() == 1)); + } + + class TestConsumerFactory implements ConsumerFactory { + + List postProcessors = new ArrayList<>(); + + @Override + public Consumer createConsumer(String groupId, String clientIdPrefix, String clientIdSuffix) { + return null; + } + + @Override + public boolean isAutoCommit() { + return false; + } + + @Override + public void addPostProcessor(ConsumerPostProcessor postProcessor) { + this.postProcessors.add(postProcessor); + } + + @Override + public List getPostProcessors() { + return this.postProcessors; + } + + } + + class TestProducerFactory implements ProducerFactory { + + List postProcessors = new ArrayList<>(); + + @Override + public Producer createProducer() { + return null; + } + + @Override + public void addPostProcessor(ProducerPostProcessor postProcessor) { + this.postProcessors.add(postProcessor); + } + + @Override + public List getPostProcessors() { + return this.postProcessors; + } + + } + +}