diff --git a/pom.xml b/pom.xml index 08c6c9d63..7b1064f32 100644 --- a/pom.xml +++ b/pom.xml @@ -92,6 +92,7 @@ 4.0.3 0.21.3 0.14.1 + 1.15.3 @@ -293,6 +294,13 @@ archunit-junit5 ${archunit-junit5.version} + + org.testcontainers + testcontainers-bom + ${testcontainers.version} + pom + import + diff --git a/spring-cloud-sleuth-instrumentation/pom.xml b/spring-cloud-sleuth-instrumentation/pom.xml index 356f74fe0..4793fa853 100644 --- a/spring-cloud-sleuth-instrumentation/pom.xml +++ b/spring-cloud-sleuth-instrumentation/pom.xml @@ -52,6 +52,11 @@ reactor-core true + + io.projectreactor.kafka + reactor-kafka + true + org.reactivestreams reactive-streams diff --git a/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/KafkaTracingCallback.java b/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/KafkaTracingCallback.java new file mode 100644 index 000000000..4e8dfd44a --- /dev/null +++ b/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); + } + } + } + +} diff --git a/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/KafkaTracingUtils.java b/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/KafkaTracingUtils.java new file mode 100644 index 000000000..3c3b23461 --- /dev/null +++ b/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 void buildAndFinishSpan(ConsumerRecord consumerRecord, Propagator propagator, + Propagator.Getter> 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(); + } + +} diff --git a/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/TracingKafkaConsumer.java b/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/TracingKafkaConsumer.java new file mode 100644 index 000000000..9698f15cd --- /dev/null +++ b/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 implements Consumer { + + private final Consumer delegate; + + private final Propagator propagator; + + private final Propagator.Getter> extractor; + + public TracingKafkaConsumer(Consumer consumer, Propagator propagator, + Propagator.Getter> getter) { + this.delegate = consumer; + this.propagator = propagator; + this.extractor = getter; + } + + @Override + public Set assignment() { + return this.delegate.assignment(); + } + + @Override + public Set subscription() { + return this.delegate.subscription(); + } + + @Override + public void subscribe(Collection collection) { + this.delegate.subscribe(collection); + } + + @Override + public void subscribe(Collection collection, ConsumerRebalanceListener consumerRebalanceListener) { + this.delegate.subscribe(collection, consumerRebalanceListener); + } + + @Override + public void assign(Collection 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 poll(long l) { + ConsumerRecords consumerRecords = this.delegate.poll(l); + for (ConsumerRecord consumerRecord : consumerRecords) { + KafkaTracingUtils.buildAndFinishSpan(consumerRecord, this.propagator, this.extractor); + } + return consumerRecords; + } + + @Override + public ConsumerRecords poll(Duration duration) { + ConsumerRecords consumerRecords = this.delegate.poll(duration); + for (ConsumerRecord 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 map) { + this.delegate.commitSync(map); + } + + @Override + public void commitSync(Map 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 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 collection) { + this.delegate.seekToBeginning(collection); + } + + @Override + public void seekToEnd(Collection 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 committed(Set set) { + return this.delegate.committed(set); + } + + @Override + public Map committed(Set set, Duration duration) { + return this.delegate.committed(set, duration); + } + + @Override + public Map metrics() { + return this.delegate.metrics(); + } + + @Override + public List partitionsFor(String s) { + return this.delegate.partitionsFor(s); + } + + @Override + public List partitionsFor(String s, Duration duration) { + return this.delegate.partitionsFor(s, duration); + } + + @Override + public Map> listTopics() { + return this.delegate.listTopics(); + } + + @Override + public Map> listTopics(Duration duration) { + return this.delegate.listTopics(duration); + } + + @Override + public Set paused() { + return this.delegate.paused(); + } + + @Override + public void pause(Collection collection) { + this.delegate.pause(collection); + } + + @Override + public void resume(Collection collection) { + this.delegate.resume(collection); + } + + @Override + public Map offsetsForTimes(Map map) { + return this.delegate.offsetsForTimes(map); + } + + @Override + public Map offsetsForTimes(Map map, Duration duration) { + return this.delegate.offsetsForTimes(map, duration); + } + + @Override + public Map beginningOffsets(Collection collection) { + return this.delegate.beginningOffsets(collection); + } + + @Override + public Map beginningOffsets(Collection collection, Duration duration) { + return this.delegate.beginningOffsets(collection, duration); + } + + @Override + public Map endOffsets(Collection collection) { + return this.delegate.endOffsets(collection); + } + + @Override + public Map endOffsets(Collection 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(); + } + +} diff --git a/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/TracingKafkaProducer.java b/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/TracingKafkaProducer.java new file mode 100644 index 000000000..51d778076 --- /dev/null +++ b/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 implements Producer { + + private static final Log log = LogFactory.getLog(TracingKafkaProducer.class); + + private final Producer delegate; + + private final Tracer tracer; + + private final Propagator propagator; + + private final Propagator.Setter> injector; + + public TracingKafkaProducer(Producer producer, Tracer tracer, Propagator propagator, + Propagator.Setter> 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 map, String s) + throws ProducerFencedException { + this.delegate.sendOffsetsToTransaction(map, s); + } + + @Override + public void sendOffsetsToTransaction(Map 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 send(ProducerRecord producerRecord) { + return send(producerRecord, null); + } + + @Override + public Future send(ProducerRecord 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 partitionsFor(String s) { + return this.delegate.partitionsFor(s); + } + + @Override + public Map metrics() { + return this.delegate.metrics(); + } + + @Override + public void close() { + this.delegate.close(); + } + + @Override + public void close(Duration duration) { + this.delegate.close(duration); + } + +} diff --git a/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/TracingKafkaProducerFactory.java b/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/TracingKafkaProducerFactory.java new file mode 100644 index 000000000..184df69b7 --- /dev/null +++ b/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 Producer createProducer(SenderOptions senderOptions) { + return new TracingKafkaProducer<>(super.createProducer(senderOptions), tracer, propagator, + new TracingKafkaPropagatorSetter()); + } + +} diff --git a/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/TracingKafkaPropagatorGetter.java b/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/TracingKafkaPropagatorGetter.java new file mode 100644 index 000000000..e09988e5c --- /dev/null +++ b/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> { + + @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); + } + +} diff --git a/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/TracingKafkaPropagatorSetter.java b/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/TracingKafkaPropagatorSetter.java new file mode 100644 index 000000000..afdb9b955 --- /dev/null +++ b/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> { + + @Override + public void set(ProducerRecord carrier, String key, String value) { + if (carrier != null) { + carrier.headers().add(key, value.getBytes()); + } + } + +} diff --git a/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/TracingKafkaReceiver.java b/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/TracingKafkaReceiver.java new file mode 100644 index 000000000..f855b5595 --- /dev/null +++ b/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 implements KafkaReceiver { + + private final KafkaReceiver delegate; + + private final Propagator propagator; + + private final Propagator.Getter> extractor; + + public TracingKafkaReceiver(KafkaReceiver receiver, Propagator propagator, + Propagator.Getter> getter) { + this.delegate = receiver; + this.propagator = propagator; + this.extractor = getter; + } + + @Override + public Flux> receive(Integer integer) { + return buildAndFinishSpanOnNextReceiverRecord(this.delegate.receive(integer)); + } + + @Override + public Flux> receive() { + return buildAndFinishSpanOnNextReceiverRecord(this.delegate.receive()); + } + + @Override + public Flux>> receiveAutoAck(Integer integer) { + return this.delegate.receiveAutoAck(integer).map(this::buildAndFinishSpanOnNextConsumerRecord); + } + + @Override + public Flux>> receiveAutoAck() { + return this.delegate.receiveAutoAck().map(this::buildAndFinishSpanOnNextConsumerRecord); + } + + @Override + public Flux> receiveAtmostOnce(Integer integer) { + return this.buildAndFinishSpanOnNextConsumerRecord(this.delegate.receiveAtmostOnce(integer)); + } + + @Override + public Flux> receiveAtmostOnce() { + return this.buildAndFinishSpanOnNextConsumerRecord(this.delegate.receiveAtmostOnce()); + } + + @Override + public Flux>> receiveExactlyOnce(TransactionManager transactionManager) { + return this.delegate.receiveExactlyOnce(transactionManager).map(this::buildAndFinishSpanOnNextConsumerRecord); + } + + @Override + public Flux>> receiveExactlyOnce(TransactionManager transactionManager, Integer integer) { + return this.delegate.receiveExactlyOnce(transactionManager, integer) + .map(this::buildAndFinishSpanOnNextConsumerRecord); + } + + @Override + public Mono doOnConsumer(Function, ? extends T> function) { + return this.delegate.doOnConsumer(function); + } + + private Flux> buildAndFinishSpanOnNextConsumerRecord(Flux> flux) { + return flux.doOnNext(consumerRecord -> KafkaTracingUtils.buildAndFinishSpan(consumerRecord, this.propagator, + this.extractor)); + } + + private Flux> buildAndFinishSpanOnNextReceiverRecord(Flux> flux) { + return flux.doOnNext(consumerRecord -> KafkaTracingUtils.buildAndFinishSpan(consumerRecord, this.propagator, + this.extractor)); + } + +} diff --git a/spring-cloud-sleuth-instrumentation/src/test/java/org/springframework/cloud/sleuth/instrument/kafka/KafkaTracingCallbackTest.java b/spring-cloud-sleuth-instrumentation/src/test/java/org/springframework/cloud/sleuth/instrument/kafka/KafkaTracingCallbackTest.java new file mode 100644 index 000000000..44fc895c9 --- /dev/null +++ b/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)); + } + +} diff --git a/spring-cloud-sleuth-instrumentation/src/test/java/org/springframework/cloud/sleuth/instrument/kafka/TracingKafkaConsumerTest.java b/spring-cloud-sleuth-instrumentation/src/test/java/org/springframework/cloud/sleuth/instrument/kafka/TracingKafkaConsumerTest.java new file mode 100644 index 000000000..c04c22d72 --- /dev/null +++ b/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 kafkaConsumer; + + @Mock(answer = Answers.RETURNS_DEEP_STUBS) + Propagator propagator; + + @Test + void should_delegate_poll_calls() { + Duration pollTimeout = Duration.of(5, ChronoUnit.SECONDS); + ConsumerRecord record = new ConsumerRecord<>("topic", 0, 1, "test-key", "test-value"); + Map>> map = new HashMap<>(); + map.put(new TopicPartition("topic", 0), Collections.singletonList(record)); + ConsumerRecords records = new ConsumerRecords<>(map); + BDDMockito.given(kafkaConsumer.poll(pollTimeout)).willReturn(records); + TracingKafkaConsumer tracingKafkaConsumer = new TracingKafkaConsumer<>(kafkaConsumer, + propagator, new TracingKafkaPropagatorGetter()); + + tracingKafkaConsumer.poll(pollTimeout); + + Mockito.verify(kafkaConsumer).poll(eq(pollTimeout)); + } + +} diff --git a/spring-cloud-sleuth-instrumentation/src/test/java/org/springframework/cloud/sleuth/instrument/kafka/TracingKafkaProducerTest.java b/spring-cloud-sleuth-instrumentation/src/test/java/org/springframework/cloud/sleuth/instrument/kafka/TracingKafkaProducerTest.java new file mode 100644 index 000000000..dcb3bab8b --- /dev/null +++ b/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 kafkaProducer; + + @Mock + Propagator propagator; + + @Mock(answer = Answers.RETURNS_DEEP_STUBS) + Tracer tracer; + + @Test + void should_delegate_send_calls() { + ProducerRecord testRecord = new ProducerRecord<>("test", "test"); + Callback callback = (record, ex) -> { + }; + TracingKafkaProducer 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 testRecord = new ProducerRecord<>("test", "test"); + Callback callback = (record, ex) -> { + }; + TracingKafkaProducer tracingKafkaProducer = new TracingKafkaProducer<>(kafkaProducer, tracer, + propagator, new TracingKafkaPropagatorSetter()); + + tracingKafkaProducer.send(testRecord, callback); + + ArgumentCaptor 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); + } + +} diff --git a/spring-cloud-sleuth-instrumentation/src/test/java/org/springframework/cloud/sleuth/instrument/kafka/TracingKafkaReceiverTest.java b/spring-cloud-sleuth-instrumentation/src/test/java/org/springframework/cloud/sleuth/instrument/kafka/TracingKafkaReceiverTest.java new file mode 100644 index 000000000..6707aeee6 --- /dev/null +++ b/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 kafkaReceiver; + + @Mock(answer = Answers.RETURNS_DEEP_STUBS) + Propagator propagator; + + @Test + void should_delegate_receive_calls() { + ReceiverOffset receiverOffset = BDDMockito.mock(ReceiverOffset.class); + ConsumerRecord record = new ConsumerRecord<>("topic", 0, 1, "test-key", "test-value"); + ReceiverRecord receiverRecord = new ReceiverRecord<>(record, receiverOffset); + BDDMockito.given(kafkaReceiver.receive()).willReturn(Flux.just(receiverRecord)); + TracingKafkaReceiver tracingKafkaReceiver = new TracingKafkaReceiver<>(kafkaReceiver, + propagator, new TracingKafkaPropagatorGetter()); + + StepVerifier.create(tracingKafkaReceiver.receive()).expectNextMatches(Predicate.isEqual(receiverRecord)) + .verifyComplete(); + + Mockito.verify(kafkaReceiver).receive(); + } + +} diff --git a/tests/brave/pom.xml b/tests/brave/pom.xml index 2e21ae75f..621d49fe6 100644 --- a/tests/brave/pom.xml +++ b/tests/brave/pom.xml @@ -44,6 +44,7 @@ spring-cloud-sleuth-instrumentation-feign-tests spring-cloud-sleuth-instrumentation-gateway-tests spring-cloud-sleuth-instrumentation-grpc-tests + spring-cloud-sleuth-instrumentation-kafka-tests spring-cloud-sleuth-instrumentation-lettuce-tests spring-cloud-sleuth-instrumentation-messaging-tests spring-cloud-sleuth-instrumentation-mvc-tests diff --git a/tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/pom.xml b/tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/pom.xml new file mode 100644 index 000000000..6faff3f17 --- /dev/null +++ b/tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/pom.xml @@ -0,0 +1,97 @@ + + + + + 4.0.0 + + spring-cloud-sleuth-instrumentation-kafka-tests + jar + Spring Cloud Sleuth Brave Kafka Instrumentation Tests + Spring Cloud Sleuth Brave Kafka Instrumentation Tests + + + org.springframework.cloud + spring-cloud-sleuth-tests-brave + 3.1.0-SNAPSHOT + .. + + + + true + + + + + + + maven-deploy-plugin + + true + + + + + + + + ${project.groupId} + spring-cloud-sleuth-tests-common + + + org.springframework.boot + spring-boot-starter-webflux + + + org.springframework.cloud + spring-cloud-starter-sleuth + + + org.springframework.boot + spring-boot-starter-test + + + io.projectreactor.kafka + reactor-kafka + true + + + org.testcontainers + testcontainers + + + org.testcontainers + junit-jupiter + + + org.testcontainers + kafka + + + io.zipkin.brave + brave-tests + + + org.awaitility + awaitility + + + + 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 new file mode 100644 index 000000000..b19b6efc3 --- /dev/null +++ b/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; + } + +} diff --git a/tests/common/pom.xml b/tests/common/pom.xml index f405ccd78..16cb95b52 100644 --- a/tests/common/pom.xml +++ b/tests/common/pom.xml @@ -144,6 +144,26 @@ brave-tests true + + io.projectreactor.kafka + reactor-kafka + true + + + org.testcontainers + testcontainers + true + + + org.testcontainers + junit-jupiter + true + + + org.testcontainers + kafka + true + org.springframework.cloud spring-cloud-sleuth-zipkin 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 new file mode 100644 index 000000000..c541eac73 --- /dev/null +++ b/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 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 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 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(); + } + +}