Spring kafka autoconfiguration (#2013)

* Add auto-configuration for spring kafka

* Add auto-configuration for spring kafka

* Make spring-kafka instrumentation autoconfig conditional on missing brave

Co-authored-by: flavium31 <w7og3bljlmmn742puqw2zwzu7ycuwmflr2nb4swrwzivccx6arpq>

docs/src/main/asciidoc/integrations.adoc

@@ -24,6 +24,8 @@ KafkaReceiver<String, String> 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
 

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);
+	}
+
+}

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<K, V> implements ConsumerPostProcessor<K, V> {
+
+	private final BeanFactory beanFactory;
+
+	SpringKafkaConsumerPostProcessor(BeanFactory beanFactory) {
+		this.beanFactory = beanFactory;
+	}
+
+	@Override
+	public Consumer<K, V> apply(Consumer<K, V> kvConsumer) {
+		return new TracingKafkaConsumer<>(kvConsumer, beanFactory);
+	}
+
+}

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;
+	}
+
+}

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<K, V> implements ProducerPostProcessor<K, V> {
+
+	private final BeanFactory beanFactory;
+
+	SpringKafkaProducerPostProcessor(BeanFactory beanFactory) {
+		this.beanFactory = beanFactory;
+	}
+
+	@Override
+	public Producer<K, V> apply(Producer<K, V> kvProducer) {
+		return new TracingKafkaProducer<>(kvProducer, beanFactory);
+	}
+
+}

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,\

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<ConsumerPostProcessor> 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<ConsumerPostProcessor> getPostProcessors() {
+			return this.postProcessors;
+		}
+
+	}
+
+	class TestProducerFactory implements ProducerFactory {
+
+		List<ProducerPostProcessor> postProcessors = new ArrayList<>();
+
+		@Override
+		public Producer createProducer() {
+			return null;
+		}
+
+		@Override
+		public void addPostProcessor(ProducerPostProcessor postProcessor) {
+			this.postProcessors.add(postProcessor);
+		}
+
+		@Override
+		public List<ProducerPostProcessor> getPostProcessors() {
+			return this.postProcessors;
+		}
+
+	}
+
+}