Kafka instrumentation (#1920)

fixes gh-1906

pom.xml

@@ -92,6 +92,7 @@
 		<awaitility.version>4.0.3</awaitility.version>
 		<brave-propagation-aws.version>0.21.3</brave-propagation-aws.version>
 		<archunit-junit5.version>0.14.1</archunit-junit5.version>
+		<testcontainers.version>1.15.3</testcontainers.version>
 	</properties>
 
 	<build>
@@ -293,6 +294,13 @@
 				<artifactId>archunit-junit5</artifactId>
 				<version>${archunit-junit5.version}</version>
 			</dependency>
+			<dependency>
+				<groupId>org.testcontainers</groupId>
+				<artifactId>testcontainers-bom</artifactId>
+				<version>${testcontainers.version}</version>
+				<type>pom</type>
+				<scope>import</scope>
+			</dependency>
 		</dependencies>
 	</dependencyManagement>
 

spring-cloud-sleuth-instrumentation/pom.xml

@@ -52,6 +52,11 @@
 			<artifactId>reactor-core</artifactId>
 			<optional>true</optional>
 		</dependency>
+		<dependency>
+			<groupId>io.projectreactor.kafka</groupId>
+			<artifactId>reactor-kafka</artifactId>
+			<optional>true</optional>
+		</dependency>
 		<dependency>
 			<groupId>org.reactivestreams</groupId>
 			<artifactId>reactive-streams</artifactId>

spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/KafkaTracingCallback.java

@@ -0,0 +1,70 @@
+/*
+ * 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.instrument.kafka;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.kafka.clients.producer.Callback;
+import org.apache.kafka.clients.producer.RecordMetadata;
+
+import org.springframework.cloud.sleuth.Span;
+import org.springframework.cloud.sleuth.Tracer;
+
+/**
+ * This decorates a Kafka {@link Callback} and completes the {@link Span.Kind#PRODUCER}
+ * span created for the record when {@code onCompletion()} is invoked (i.e. the broker has
+ * acknowledged or an {@link Exception}) was thrown.
+ *
+ * @author Anders Clausen
+ * @author Flaviu Muresan
+ * @since 3.0.3
+ */
+public class KafkaTracingCallback implements Callback {
+
+	private static final Log log = LogFactory.getLog(KafkaTracingCallback.class);
+
+	private final Callback callback;
+
+	private final Tracer tracer;
+
+	private final Span span;
+
+	public KafkaTracingCallback(Callback callback, Tracer tracer, Span span) {
+		this.callback = callback;
+		this.tracer = tracer;
+		this.span = span;
+	}
+
+	@Override
+	public void onCompletion(RecordMetadata recordMetadata, Exception e) {
+		try (Tracer.SpanInScope spanInScope = tracer.withSpan(this.span)) {
+			if (this.callback != null) {
+				this.callback.onCompletion(recordMetadata, e);
+			}
+		}
+		finally {
+			if (e != null) {
+				this.span.error(e);
+			}
+			this.span.end();
+			if (log.isDebugEnabled()) {
+				log.debug("Finished producer span " + span);
+			}
+		}
+	}
+
+}

spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/KafkaTracingUtils.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2013-2020 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.instrument.kafka;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+
+import org.springframework.cloud.sleuth.Span;
+import org.springframework.cloud.sleuth.propagation.Propagator;
+
+final class KafkaTracingUtils {
+
+	private static final Log log = LogFactory.getLog(KafkaTracingUtils.class);
+
+	private KafkaTracingUtils() {
+	}
+
+	static <K, V> void buildAndFinishSpan(ConsumerRecord<K, V> consumerRecord, Propagator propagator,
+			Propagator.Getter<ConsumerRecord<?, ?>> extractor) {
+		Span.Builder spanBuilder = propagator.extract(consumerRecord, extractor).kind(Span.Kind.CONSUMER)
+				.name("kafka.consume").tag("kafka.topic", consumerRecord.topic())
+				.tag("kafka.offset", Long.toString(consumerRecord.offset()))
+				.tag("kafka.partition", Integer.toString(consumerRecord.partition()));
+		Span span = spanBuilder.start();
+		if (log.isDebugEnabled()) {
+			log.debug("Extracted span from event headers " + span);
+		}
+		span.end();
+	}
+
+}

spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/TracingKafkaConsumer.java

@@ -0,0 +1,314 @@
+/*
+ * 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.instrument.kafka;
+
+import java.time.Duration;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerGroupMetadata;
+import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
+import org.apache.kafka.clients.consumer.OffsetCommitCallback;
+import org.apache.kafka.common.Metric;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+
+import org.springframework.cloud.sleuth.Span;
+import org.springframework.cloud.sleuth.propagation.Propagator;
+
+/**
+ * This decorates a Kafka {@link Consumer}. It creates and completes a
+ * {@link Span.Kind#CONSUMER} span for each record received. This span will be a child
+ * span of the one extracted from the record headers.
+ *
+ * @author Anders Clausen
+ * @author Flaviu Muresan
+ * @since 3.0.3
+ */
+public class TracingKafkaConsumer<K, V> implements Consumer<K, V> {
+
+	private final Consumer<K, V> delegate;
+
+	private final Propagator propagator;
+
+	private final Propagator.Getter<ConsumerRecord<?, ?>> extractor;
+
+	public TracingKafkaConsumer(Consumer<K, V> consumer, Propagator propagator,
+			Propagator.Getter<ConsumerRecord<?, ?>> getter) {
+		this.delegate = consumer;
+		this.propagator = propagator;
+		this.extractor = getter;
+	}
+
+	@Override
+	public Set<TopicPartition> assignment() {
+		return this.delegate.assignment();
+	}
+
+	@Override
+	public Set<String> subscription() {
+		return this.delegate.subscription();
+	}
+
+	@Override
+	public void subscribe(Collection<String> collection) {
+		this.delegate.subscribe(collection);
+	}
+
+	@Override
+	public void subscribe(Collection<String> collection, ConsumerRebalanceListener consumerRebalanceListener) {
+		this.delegate.subscribe(collection, consumerRebalanceListener);
+	}
+
+	@Override
+	public void assign(Collection<TopicPartition> collection) {
+		this.delegate.assign(collection);
+	}
+
+	@Override
+	public void subscribe(Pattern pattern, ConsumerRebalanceListener consumerRebalanceListener) {
+		this.delegate.subscribe(pattern, consumerRebalanceListener);
+	}
+
+	@Override
+	public void subscribe(Pattern pattern) {
+		this.delegate.subscribe(pattern);
+	}
+
+	@Override
+	public void unsubscribe() {
+		this.delegate.unsubscribe();
+	}
+
+	@Deprecated
+	@Override
+	public ConsumerRecords<K, V> poll(long l) {
+		ConsumerRecords<K, V> consumerRecords = this.delegate.poll(l);
+		for (ConsumerRecord<K, V> consumerRecord : consumerRecords) {
+			KafkaTracingUtils.buildAndFinishSpan(consumerRecord, this.propagator, this.extractor);
+		}
+		return consumerRecords;
+	}
+
+	@Override
+	public ConsumerRecords<K, V> poll(Duration duration) {
+		ConsumerRecords<K, V> consumerRecords = this.delegate.poll(duration);
+		for (ConsumerRecord<K, V> consumerRecord : consumerRecords) {
+			KafkaTracingUtils.buildAndFinishSpan(consumerRecord, this.propagator, this.extractor);
+		}
+		return consumerRecords;
+	}
+
+	@Override
+	public void commitSync() {
+		this.delegate.commitSync();
+	}
+
+	@Override
+	public void commitSync(Duration duration) {
+		this.delegate.commitSync(duration);
+	}
+
+	@Override
+	public void commitSync(Map<TopicPartition, OffsetAndMetadata> map) {
+		this.delegate.commitSync(map);
+	}
+
+	@Override
+	public void commitSync(Map<TopicPartition, OffsetAndMetadata> map, Duration duration) {
+		this.delegate.commitSync(map, duration);
+	}
+
+	@Override
+	public void commitAsync() {
+		this.delegate.commitAsync();
+	}
+
+	@Override
+	public void commitAsync(OffsetCommitCallback offsetCommitCallback) {
+		this.delegate.commitAsync(offsetCommitCallback);
+	}
+
+	@Override
+	public void commitAsync(Map<TopicPartition, OffsetAndMetadata> map, OffsetCommitCallback offsetCommitCallback) {
+		this.delegate.commitAsync(map, offsetCommitCallback);
+	}
+
+	@Override
+	public void seek(TopicPartition topicPartition, long l) {
+		this.delegate.seek(topicPartition, l);
+	}
+
+	@Override
+	public void seek(TopicPartition topicPartition, OffsetAndMetadata offsetAndMetadata) {
+		this.delegate.seek(topicPartition, offsetAndMetadata);
+	}
+
+	@Override
+	public void seekToBeginning(Collection<TopicPartition> collection) {
+		this.delegate.seekToBeginning(collection);
+	}
+
+	@Override
+	public void seekToEnd(Collection<TopicPartition> collection) {
+		this.delegate.seekToEnd(collection);
+	}
+
+	@Override
+	public long position(TopicPartition topicPartition) {
+		return this.delegate.position(topicPartition);
+	}
+
+	@Override
+	public long position(TopicPartition topicPartition, Duration duration) {
+		return this.delegate.position(topicPartition, duration);
+	}
+
+	@Override
+	@Deprecated
+	public OffsetAndMetadata committed(TopicPartition topicPartition) {
+		return this.delegate.committed(topicPartition);
+	}
+
+	@Override
+	@Deprecated
+	public OffsetAndMetadata committed(TopicPartition topicPartition, Duration duration) {
+		return this.delegate.committed(topicPartition, duration);
+	}
+
+	@Override
+	public Map<TopicPartition, OffsetAndMetadata> committed(Set<TopicPartition> set) {
+		return this.delegate.committed(set);
+	}
+
+	@Override
+	public Map<TopicPartition, OffsetAndMetadata> committed(Set<TopicPartition> set, Duration duration) {
+		return this.delegate.committed(set, duration);
+	}
+
+	@Override
+	public Map<MetricName, ? extends Metric> metrics() {
+		return this.delegate.metrics();
+	}
+
+	@Override
+	public List<PartitionInfo> partitionsFor(String s) {
+		return this.delegate.partitionsFor(s);
+	}
+
+	@Override
+	public List<PartitionInfo> partitionsFor(String s, Duration duration) {
+		return this.delegate.partitionsFor(s, duration);
+	}
+
+	@Override
+	public Map<String, List<PartitionInfo>> listTopics() {
+		return this.delegate.listTopics();
+	}
+
+	@Override
+	public Map<String, List<PartitionInfo>> listTopics(Duration duration) {
+		return this.delegate.listTopics(duration);
+	}
+
+	@Override
+	public Set<TopicPartition> paused() {
+		return this.delegate.paused();
+	}
+
+	@Override
+	public void pause(Collection<TopicPartition> collection) {
+		this.delegate.pause(collection);
+	}
+
+	@Override
+	public void resume(Collection<TopicPartition> collection) {
+		this.delegate.resume(collection);
+	}
+
+	@Override
+	public Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(Map<TopicPartition, Long> map) {
+		return this.delegate.offsetsForTimes(map);
+	}
+
+	@Override
+	public Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(Map<TopicPartition, Long> map, Duration duration) {
+		return this.delegate.offsetsForTimes(map, duration);
+	}
+
+	@Override
+	public Map<TopicPartition, Long> beginningOffsets(Collection<TopicPartition> collection) {
+		return this.delegate.beginningOffsets(collection);
+	}
+
+	@Override
+	public Map<TopicPartition, Long> beginningOffsets(Collection<TopicPartition> collection, Duration duration) {
+		return this.delegate.beginningOffsets(collection, duration);
+	}
+
+	@Override
+	public Map<TopicPartition, Long> endOffsets(Collection<TopicPartition> collection) {
+		return this.delegate.endOffsets(collection);
+	}
+
+	@Override
+	public Map<TopicPartition, Long> endOffsets(Collection<TopicPartition> collection, Duration duration) {
+		return this.delegate.endOffsets(collection, duration);
+	}
+
+	@Override
+	public ConsumerGroupMetadata groupMetadata() {
+		return this.delegate.groupMetadata();
+	}
+
+	@Override
+	public void enforceRebalance() {
+		this.delegate.enforceRebalance();
+	}
+
+	@Override
+	public void close() {
+		this.delegate.close();
+	}
+
+	@Override
+	@Deprecated
+	public void close(long l, TimeUnit timeUnit) {
+		this.delegate.close(l, timeUnit);
+	}
+
+	@Override
+	public void close(Duration duration) {
+		this.delegate.close(duration);
+	}
+
+	@Override
+	public void wakeup() {
+		this.delegate.wakeup();
+	}
+
+}

spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/TracingKafkaProducer.java

@@ -0,0 +1,147 @@
+/*
+ * 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.instrument.kafka;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Future;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.kafka.clients.consumer.ConsumerGroupMetadata;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.clients.producer.Callback;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.Metric;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ProducerFencedException;
+
+import org.springframework.cloud.sleuth.Span;
+import org.springframework.cloud.sleuth.Tracer;
+import org.springframework.cloud.sleuth.propagation.Propagator;
+
+/**
+ * This decorates a Kafka {@link Producer} and creates a {@link Span.Kind#PRODUCER} span
+ * for each record sent. This span is also injected onto each record (via headers) so it
+ * becomes the parent when a consumer later receives the record.
+ *
+ * @author Anders Clausen
+ * @author Flaviu Muresan
+ * @since 3.0.3
+ */
+public class TracingKafkaProducer<K, V> implements Producer<K, V> {
+
+	private static final Log log = LogFactory.getLog(TracingKafkaProducer.class);
+
+	private final Producer<K, V> delegate;
+
+	private final Tracer tracer;
+
+	private final Propagator propagator;
+
+	private final Propagator.Setter<ProducerRecord<?, ?>> injector;
+
+	public TracingKafkaProducer(Producer<K, V> producer, Tracer tracer, Propagator propagator,
+			Propagator.Setter<ProducerRecord<?, ?>> setter) {
+		this.delegate = producer;
+		this.tracer = tracer;
+		this.propagator = propagator;
+		this.injector = setter;
+	}
+
+	@Override
+	public void initTransactions() {
+		this.delegate.initTransactions();
+	}
+
+	@Override
+	public void beginTransaction() throws ProducerFencedException {
+		this.delegate.beginTransaction();
+	}
+
+	@Override
+	public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> map, String s)
+			throws ProducerFencedException {
+		this.delegate.sendOffsetsToTransaction(map, s);
+	}
+
+	@Override
+	public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> map,
+			ConsumerGroupMetadata consumerGroupMetadata) throws ProducerFencedException {
+		this.delegate.sendOffsetsToTransaction(map, consumerGroupMetadata);
+	}
+
+	@Override
+	public void commitTransaction() throws ProducerFencedException {
+		this.delegate.commitTransaction();
+	}
+
+	@Override
+	public void abortTransaction() throws ProducerFencedException {
+		this.delegate.abortTransaction();
+	}
+
+	@Override
+	public Future<RecordMetadata> send(ProducerRecord<K, V> producerRecord) {
+		return send(producerRecord, null);
+	}
+
+	@Override
+	public Future<RecordMetadata> send(ProducerRecord<K, V> producerRecord, Callback callback) {
+		Span.Builder spanBuilder = tracer.spanBuilder().kind(Span.Kind.PRODUCER).name("kafka.produce")
+				.tag("kafka.topic", producerRecord.topic());
+		Span span = spanBuilder.start();
+		this.propagator.inject(span.context(), producerRecord, this.injector);
+		try (Tracer.SpanInScope spanInScope = tracer.withSpan(span)) {
+			if (log.isDebugEnabled()) {
+				log.debug("Created producer span " + span);
+			}
+			return this.delegate.send(producerRecord, new KafkaTracingCallback(callback, tracer, span));
+		}
+	}
+
+	@Override
+	public void flush() {
+		this.delegate.flush();
+	}
+
+	@Override
+	public List<PartitionInfo> partitionsFor(String s) {
+		return this.delegate.partitionsFor(s);
+	}
+
+	@Override
+	public Map<MetricName, ? extends Metric> metrics() {
+		return this.delegate.metrics();
+	}
+
+	@Override
+	public void close() {
+		this.delegate.close();
+	}
+
+	@Override
+	public void close(Duration duration) {
+		this.delegate.close(duration);
+	}
+
+}

spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/TracingKafkaProducerFactory.java

@@ -0,0 +1,54 @@
+/*
+ * 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.instrument.kafka;
+
+import org.apache.kafka.clients.producer.Producer;
+import reactor.kafka.sender.KafkaSender;
+import reactor.kafka.sender.SenderOptions;
+import reactor.kafka.sender.internals.ProducerFactory;
+
+import org.springframework.cloud.sleuth.Tracer;
+import org.springframework.cloud.sleuth.propagation.Propagator;
+
+/**
+ * This decorates a Reactor Kafka {@link ProducerFactory} to create decorated producers of
+ * type {@link TracingKafkaProducer}. This can be used by the {@link KafkaSender} factory
+ * methods to create instrumented senders.
+ *
+ * @author Anders Clausen
+ * @author Flaviu Muresan
+ * @since 3.0.3
+ */
+public class TracingKafkaProducerFactory extends ProducerFactory {
+
+	private final Tracer tracer;
+
+	private final Propagator propagator;
+
+	public TracingKafkaProducerFactory(Tracer tracer, Propagator propagator) {
+		super();
+		this.tracer = tracer;
+		this.propagator = propagator;
+	}
+
+	@Override
+	public <K, V> Producer<K, V> createProducer(SenderOptions<K, V> senderOptions) {
+		return new TracingKafkaProducer<>(super.createProducer(senderOptions), tracer, propagator,
+				new TracingKafkaPropagatorSetter());
+	}
+
+}

spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/TracingKafkaPropagatorGetter.java

@@ -0,0 +1,44 @@
+/*
+ * 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.instrument.kafka;
+
+import java.util.Iterator;
+import java.util.Optional;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.header.Header;
+
+import org.springframework.cloud.sleuth.propagation.Propagator;
+
+/**
+ * Getter extracting the values from the {@link ConsumerRecord} headers for Kafka based
+ * communication.
+ *
+ * @author Anders Clausen
+ * @author Flaviu Muresan
+ * @since 3.0.3
+ */
+public class TracingKafkaPropagatorGetter implements Propagator.Getter<ConsumerRecord<?, ?>> {
+
+	@Override
+	public String get(ConsumerRecord<?, ?> carrier, String key) {
+		return Optional.ofNullable(carrier).map(ConsumerRecord::headers).map(headers -> headers.headers(key))
+				.map(Iterable::iterator).filter(Iterator::hasNext).map(Iterator::next).map(Header::value)
+				.map(String::new).orElse(null);
+	}
+
+}

spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/TracingKafkaPropagatorSetter.java

@@ -0,0 +1,40 @@
+/*
+ * 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.instrument.kafka;
+
+import org.apache.kafka.clients.producer.ProducerRecord;
+
+import org.springframework.cloud.sleuth.propagation.Propagator;
+
+/**
+ * Setter injecting the values onto the {@link ProducerRecord} headers for Kafka based
+ * communication.
+ *
+ * @author Anders Clausen
+ * @author Flaviu Muresan
+ * @since 3.0.3
+ */
+public class TracingKafkaPropagatorSetter implements Propagator.Setter<ProducerRecord<?, ?>> {
+
+	@Override
+	public void set(ProducerRecord<?, ?> carrier, String key, String value) {
+		if (carrier != null) {
+			carrier.headers().add(key, value.getBytes());
+		}
+	}
+
+}

spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/TracingKafkaReceiver.java

@@ -0,0 +1,112 @@
+/*
+ * 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.instrument.kafka;
+
+import java.util.function.Function;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.kafka.receiver.KafkaReceiver;
+import reactor.kafka.receiver.ReceiverRecord;
+import reactor.kafka.sender.TransactionManager;
+
+import org.springframework.cloud.sleuth.Span;
+import org.springframework.cloud.sleuth.propagation.Propagator;
+
+/**
+ * This decorates a reactive {@link KafkaReceiver} and creates and completes a
+ * {@link Span.Kind#CONSUMER} span for each record received. This span will be a child
+ * span of the one extracted from the record headers.
+ *
+ * @author Anders Clausen
+ * @author Flaviu Muresan
+ * @since 3.0.3
+ */
+public class TracingKafkaReceiver<K, V> implements KafkaReceiver<K, V> {
+
+	private final KafkaReceiver<K, V> delegate;
+
+	private final Propagator propagator;
+
+	private final Propagator.Getter<ConsumerRecord<?, ?>> extractor;
+
+	public TracingKafkaReceiver(KafkaReceiver<K, V> receiver, Propagator propagator,
+			Propagator.Getter<ConsumerRecord<?, ?>> getter) {
+		this.delegate = receiver;
+		this.propagator = propagator;
+		this.extractor = getter;
+	}
+
+	@Override
+	public Flux<ReceiverRecord<K, V>> receive(Integer integer) {
+		return buildAndFinishSpanOnNextReceiverRecord(this.delegate.receive(integer));
+	}
+
+	@Override
+	public Flux<ReceiverRecord<K, V>> receive() {
+		return buildAndFinishSpanOnNextReceiverRecord(this.delegate.receive());
+	}
+
+	@Override
+	public Flux<Flux<ConsumerRecord<K, V>>> receiveAutoAck(Integer integer) {
+		return this.delegate.receiveAutoAck(integer).map(this::buildAndFinishSpanOnNextConsumerRecord);
+	}
+
+	@Override
+	public Flux<Flux<ConsumerRecord<K, V>>> receiveAutoAck() {
+		return this.delegate.receiveAutoAck().map(this::buildAndFinishSpanOnNextConsumerRecord);
+	}
+
+	@Override
+	public Flux<ConsumerRecord<K, V>> receiveAtmostOnce(Integer integer) {
+		return this.buildAndFinishSpanOnNextConsumerRecord(this.delegate.receiveAtmostOnce(integer));
+	}
+
+	@Override
+	public Flux<ConsumerRecord<K, V>> receiveAtmostOnce() {
+		return this.buildAndFinishSpanOnNextConsumerRecord(this.delegate.receiveAtmostOnce());
+	}
+
+	@Override
+	public Flux<Flux<ConsumerRecord<K, V>>> receiveExactlyOnce(TransactionManager transactionManager) {
+		return this.delegate.receiveExactlyOnce(transactionManager).map(this::buildAndFinishSpanOnNextConsumerRecord);
+	}
+
+	@Override
+	public Flux<Flux<ConsumerRecord<K, V>>> receiveExactlyOnce(TransactionManager transactionManager, Integer integer) {
+		return this.delegate.receiveExactlyOnce(transactionManager, integer)
+				.map(this::buildAndFinishSpanOnNextConsumerRecord);
+	}
+
+	@Override
+	public <T> Mono<T> doOnConsumer(Function<Consumer<K, V>, ? extends T> function) {
+		return this.delegate.doOnConsumer(function);
+	}
+
+	private Flux<ConsumerRecord<K, V>> buildAndFinishSpanOnNextConsumerRecord(Flux<ConsumerRecord<K, V>> flux) {
+		return flux.doOnNext(consumerRecord -> KafkaTracingUtils.buildAndFinishSpan(consumerRecord, this.propagator,
+				this.extractor));
+	}
+
+	private Flux<ReceiverRecord<K, V>> buildAndFinishSpanOnNextReceiverRecord(Flux<ReceiverRecord<K, V>> flux) {
+		return flux.doOnNext(consumerRecord -> KafkaTracingUtils.buildAndFinishSpan(consumerRecord, this.propagator,
+				this.extractor));
+	}
+
+}

spring-cloud-sleuth-instrumentation/src/test/java/org/springframework/cloud/sleuth/instrument/kafka/KafkaTracingCallbackTest.java

@@ -0,0 +1,65 @@
+/*
+ * 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.instrument.kafka;
+
+import org.apache.kafka.clients.producer.Callback;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import org.springframework.cloud.sleuth.Span;
+import org.springframework.cloud.sleuth.Tracer;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.isNull;
+
+@ExtendWith(MockitoExtension.class)
+public class KafkaTracingCallbackTest {
+
+	@Mock
+	Tracer tracer;
+
+	@Mock
+	Span span;
+
+	@Mock
+	Callback callback;
+
+	@Test
+	void should_call_on_completion_on_user_callback_success() {
+		KafkaTracingCallback tracingCallback = new KafkaTracingCallback(callback, tracer, span);
+		RecordMetadata recordMetadata = new RecordMetadata(null, 0, 0, 0, 0L, 0, 0);
+
+		tracingCallback.onCompletion(recordMetadata, null);
+
+		Mockito.verify(callback).onCompletion(eq(recordMetadata), isNull());
+	}
+
+	@Test
+	void should_call_on_completion_on_user_callback_error() {
+		KafkaTracingCallback tracingCallback = new KafkaTracingCallback(callback, tracer, span);
+
+		tracingCallback.onCompletion(null, new RuntimeException());
+
+		Mockito.verify(callback).onCompletion(isNull(), any(RuntimeException.class));
+	}
+
+}

spring-cloud-sleuth-instrumentation/src/test/java/org/springframework/cloud/sleuth/instrument/kafka/TracingKafkaConsumerTest.java

@@ -0,0 +1,67 @@
+/*
+ * 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.instrument.kafka;
+
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.TopicPartition;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Answers;
+import org.mockito.BDDMockito;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import org.springframework.cloud.sleuth.propagation.Propagator;
+
+import static org.mockito.ArgumentMatchers.eq;
+
+@ExtendWith(MockitoExtension.class)
+public class TracingKafkaConsumerTest {
+
+	@Mock
+	KafkaConsumer<String, String> kafkaConsumer;
+
+	@Mock(answer = Answers.RETURNS_DEEP_STUBS)
+	Propagator propagator;
+
+	@Test
+	void should_delegate_poll_calls() {
+		Duration pollTimeout = Duration.of(5, ChronoUnit.SECONDS);
+		ConsumerRecord<String, String> record = new ConsumerRecord<>("topic", 0, 1, "test-key", "test-value");
+		Map<TopicPartition, List<ConsumerRecord<String, String>>> map = new HashMap<>();
+		map.put(new TopicPartition("topic", 0), Collections.singletonList(record));
+		ConsumerRecords<String, String> records = new ConsumerRecords<>(map);
+		BDDMockito.given(kafkaConsumer.poll(pollTimeout)).willReturn(records);
+		TracingKafkaConsumer<String, String> tracingKafkaConsumer = new TracingKafkaConsumer<>(kafkaConsumer,
+				propagator, new TracingKafkaPropagatorGetter());
+
+		tracingKafkaConsumer.poll(pollTimeout);
+
+		Mockito.verify(kafkaConsumer).poll(eq(pollTimeout));
+	}
+
+}

spring-cloud-sleuth-instrumentation/src/test/java/org/springframework/cloud/sleuth/instrument/kafka/TracingKafkaProducerTest.java

@@ -0,0 +1,79 @@
+/*
+ * 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.instrument.kafka;
+
+import org.apache.kafka.clients.producer.Callback;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.assertj.core.api.BDDAssertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Answers;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import org.springframework.cloud.sleuth.Tracer;
+import org.springframework.cloud.sleuth.propagation.Propagator;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+
+@ExtendWith(MockitoExtension.class)
+public class TracingKafkaProducerTest {
+
+	@Mock
+	KafkaProducer<String, String> kafkaProducer;
+
+	@Mock
+	Propagator propagator;
+
+	@Mock(answer = Answers.RETURNS_DEEP_STUBS)
+	Tracer tracer;
+
+	@Test
+	void should_delegate_send_calls() {
+		ProducerRecord<String, String> testRecord = new ProducerRecord<>("test", "test");
+		Callback callback = (record, ex) -> {
+		};
+		TracingKafkaProducer<String, String> tracingKafkaProducer = new TracingKafkaProducer<>(kafkaProducer, tracer,
+				propagator, new TracingKafkaPropagatorSetter());
+
+		tracingKafkaProducer.send(testRecord, callback);
+
+		Mockito.verify(kafkaProducer).send(eq(testRecord), any());
+	}
+
+	@Test
+	void should_wrap_user_callback_on_send() {
+		ProducerRecord<String, String> testRecord = new ProducerRecord<>("test", "test");
+		Callback callback = (record, ex) -> {
+		};
+		TracingKafkaProducer<String, String> tracingKafkaProducer = new TracingKafkaProducer<>(kafkaProducer, tracer,
+				propagator, new TracingKafkaPropagatorSetter());
+
+		tracingKafkaProducer.send(testRecord, callback);
+
+		ArgumentCaptor<KafkaTracingCallback> callbackArgument = ArgumentCaptor.forClass(KafkaTracingCallback.class);
+		Mockito.verify(kafkaProducer).send(any(), callbackArgument.capture());
+		BDDAssertions.then(callbackArgument.getValue()).isNotNull();
+		BDDAssertions.then(ReflectionTestUtils.getField(callbackArgument.getValue(), "callback")).isEqualTo(callback);
+	}
+
+}

spring-cloud-sleuth-instrumentation/src/test/java/org/springframework/cloud/sleuth/instrument/kafka/TracingKafkaReceiverTest.java

@@ -0,0 +1,61 @@
+/*
+ * 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.instrument.kafka;
+
+import java.util.function.Predicate;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Answers;
+import org.mockito.BDDMockito;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.junit.jupiter.MockitoExtension;
+import reactor.core.publisher.Flux;
+import reactor.kafka.receiver.KafkaReceiver;
+import reactor.kafka.receiver.ReceiverOffset;
+import reactor.kafka.receiver.ReceiverRecord;
+import reactor.test.StepVerifier;
+
+import org.springframework.cloud.sleuth.propagation.Propagator;
+
+@ExtendWith(MockitoExtension.class)
+public class TracingKafkaReceiverTest {
+
+	@Mock
+	KafkaReceiver<String, String> kafkaReceiver;
+
+	@Mock(answer = Answers.RETURNS_DEEP_STUBS)
+	Propagator propagator;
+
+	@Test
+	void should_delegate_receive_calls() {
+		ReceiverOffset receiverOffset = BDDMockito.mock(ReceiverOffset.class);
+		ConsumerRecord<String, String> record = new ConsumerRecord<>("topic", 0, 1, "test-key", "test-value");
+		ReceiverRecord<String, String> receiverRecord = new ReceiverRecord<>(record, receiverOffset);
+		BDDMockito.given(kafkaReceiver.receive()).willReturn(Flux.just(receiverRecord));
+		TracingKafkaReceiver<String, String> tracingKafkaReceiver = new TracingKafkaReceiver<>(kafkaReceiver,
+				propagator, new TracingKafkaPropagatorGetter());
+
+		StepVerifier.create(tracingKafkaReceiver.receive()).expectNextMatches(Predicate.isEqual(receiverRecord))
+				.verifyComplete();
+
+		Mockito.verify(kafkaReceiver).receive();
+	}
+
+}

tests/brave/pom.xml

@@ -44,6 +44,7 @@
 		<module>spring-cloud-sleuth-instrumentation-feign-tests</module>
 		<module>spring-cloud-sleuth-instrumentation-gateway-tests</module>
 		<module>spring-cloud-sleuth-instrumentation-grpc-tests</module>
+		<module>spring-cloud-sleuth-instrumentation-kafka-tests</module>
 		<module>spring-cloud-sleuth-instrumentation-lettuce-tests</module>
 		<module>spring-cloud-sleuth-instrumentation-messaging-tests</module>
 		<module>spring-cloud-sleuth-instrumentation-mvc-tests</module>

tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/pom.xml

@@ -0,0 +1,97 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+  ~
+  ~
+  -->
+
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+		 xmlns="http://maven.apache.org/POM/4.0.0"
+		 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+
+	<artifactId>spring-cloud-sleuth-instrumentation-kafka-tests</artifactId>
+	<packaging>jar</packaging>
+	<name>Spring Cloud Sleuth Brave Kafka Instrumentation Tests</name>
+	<description>Spring Cloud Sleuth Brave Kafka Instrumentation Tests</description>
+
+	<parent>
+		<groupId>org.springframework.cloud</groupId>
+		<artifactId>spring-cloud-sleuth-tests-brave</artifactId>
+		<version>3.1.0-SNAPSHOT</version>
+		<relativePath>..</relativePath>
+	</parent>
+
+	<properties>
+		<sonar.skip>true</sonar.skip>
+	</properties>
+
+	<build>
+		<plugins>
+			<plugin>
+				<!--skip deploy -->
+				<artifactId>maven-deploy-plugin</artifactId>
+				<configuration>
+					<skip>true</skip>
+				</configuration>
+			</plugin>
+		</plugins>
+	</build>
+
+	<dependencies>
+		<dependency>
+			<groupId>${project.groupId}</groupId>
+			<artifactId>spring-cloud-sleuth-tests-common</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-webflux</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-starter-sleuth</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-test</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>io.projectreactor.kafka</groupId>
+			<artifactId>reactor-kafka</artifactId>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>org.testcontainers</groupId>
+			<artifactId>testcontainers</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.testcontainers</groupId>
+			<artifactId>junit-jupiter</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.testcontainers</groupId>
+			<artifactId>kafka</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>io.zipkin.brave</groupId>
+			<artifactId>brave-tests</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.awaitility</groupId>
+			<artifactId>awaitility</artifactId>
+		</dependency>
+	</dependencies>
+
+</project>

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

@@ -0,0 +1,34 @@
+/*
+ * 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.springframework.cloud.sleuth.brave.BraveTestTracing;
+import org.springframework.cloud.sleuth.test.TestTracingAware;
+
+public class KafkaProducerTest extends org.springframework.cloud.sleuth.instrument.kafka.KafkaProducerTest {
+
+	BraveTestTracing testTracing;
+
+	@Override
+	public TestTracingAware tracerTest() {
+		if (this.testTracing == null) {
+			this.testTracing = new BraveTestTracing();
+		}
+		return this.testTracing;
+	}
+
+}

tests/common/pom.xml

@@ -144,6 +144,26 @@
 			<artifactId>brave-tests</artifactId>
 			<optional>true</optional>
 		</dependency>
+		<dependency>
+			<groupId>io.projectreactor.kafka</groupId>
+			<artifactId>reactor-kafka</artifactId>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>org.testcontainers</groupId>
+			<artifactId>testcontainers</artifactId>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>org.testcontainers</groupId>
+			<artifactId>junit-jupiter</artifactId>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>org.testcontainers</groupId>
+			<artifactId>kafka</artifactId>
+			<optional>true</optional>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.cloud</groupId>
 			<artifactId>spring-cloud-sleuth-zipkin</artifactId>

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

@@ -0,0 +1,98 @@
+/*
+ * 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.instrument.kafka;
+
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.kafka.clients.producer.Callback;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.assertj.core.api.BDDAssertions;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.KafkaContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.DockerImageName;
+
+import org.springframework.cloud.sleuth.Span;
+import org.springframework.cloud.sleuth.Tracer;
+import org.springframework.cloud.sleuth.propagation.Propagator;
+import org.springframework.cloud.sleuth.test.TestSpanHandler;
+import org.springframework.cloud.sleuth.test.TestTracingAwareSupplier;
+
+import static org.awaitility.Awaitility.await;
+
+@Testcontainers
+public abstract class KafkaProducerTest implements TestTracingAwareSupplier {
+
+	protected Tracer tracer = tracerTest().tracing().tracer();
+
+	protected Propagator propagator = tracerTest().tracing().propagator();
+
+	protected TestSpanHandler spans = tracerTest().handler();
+
+	protected TracingKafkaProducer<String, String> kafkaProducer;
+
+	@Container
+	protected final KafkaContainer kafkaContainer = new KafkaContainer(
+			DockerImageName.parse("confluentinc/cp-kafka:5.2.1")).withExposedPorts(9093)
+					.waitingFor(Wait.forListeningPort());
+
+	@BeforeEach
+	void setup() {
+		kafkaContainer.start();
+		Map<String, Object> properties = new HashMap<>();
+		properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaContainer.getBootstrapServers());
+		properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+		properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+		kafkaProducer = new TracingKafkaProducer<>(new KafkaProducer<>(properties), tracer, propagator,
+				new TracingKafkaPropagatorSetter());
+	}
+
+	@AfterEach
+	void destroy() {
+		kafkaContainer.stop();
+	}
+
+	@Test
+	public void should_create_and_finish_producer_span() {
+		AtomicBoolean acknowledged = new AtomicBoolean(false);
+		Callback callback = (metadata, ex) -> acknowledged.set(true);
+		ProducerRecord<String, String> producerRecord = new ProducerRecord<>("spring-cloud-sleuth-otel-topic", "test",
+				"test");
+		this.kafkaProducer.send(producerRecord, callback);
+		await().atMost(Duration.ofSeconds(5)).until(acknowledged::get);
+
+		BDDAssertions.then(this.tracer.currentSpan()).isNull();
+		BDDAssertions.then(this.spans).isNotEmpty();
+		BDDAssertions.then(this.spans.get(0).getKind()).isEqualTo(Span.Kind.PRODUCER);
+	}
+
+	@Override
+	public void cleanUpTracing() {
+		this.spans.clear();
+	}
+
+}