SpringCloud/spring-cloud-sleuth
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Kafka instrumentation enhancements (#1936)

Browse Source 
* Bump javadocs since version to 3.1.0

* Extend integration tests to cover all Kafka clients

* Add autoconfiguration module for kafka instrumentation

* Refactor instrumentation for reactive Kafka Receiver

* Refactor instrumentation for reactive Kafka Receiver

* Add docs for Kafka instrumentation

* Revert "Refactor instrumentation for reactive Kafka Receiver"

This reverts commit 58c8f2fa

* Revert "Revert "Refactor instrumentation for reactive Kafka Receiver""

This reverts commit 450a9f8c

* Remove empty test

* Resolve comments from PR 1936

* Revert whitespaces

* Revert whitespaces in common tests pom.xml

* Fix autoconfig to consider generics when registering beans.
Only register reactor-kafka beans if the dependency is on the classpath.

* Split autoconfig for kafka and reactor-kafka
This commit is contained in:
Flaviu Mureșan
2021-05-07 22:20:37 +02:00
committed by GitHub
parent 507e918c39
commit 59c170d147
32 changed files with 1266 additions and 232 deletions
Download Patch File Download Diff File Expand all files Collapse all files

159
tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/KafkaConsumerTest.java Normal file
View File

@@ -0,0 +1,159 @@
/*
* 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.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.regex.Pattern;
import org.apache.kafka.clients.consumer.ConsumerConfig;
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.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.assertj.core.api.BDDAssertions;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
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.junit.jupiter.MockitoExtension;
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.beans.factory.BeanFactory;
import org.springframework.cloud.sleuth.Span;
import org.springframework.cloud.sleuth.Tracer;
import org.springframework.cloud.sleuth.exporter.FinishedSpan;
import org.springframework.cloud.sleuth.propagation.Propagator;
import org.springframework.cloud.sleuth.test.TestSpanHandler;
import org.springframework.cloud.sleuth.test.TestTracingAwareSupplier;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.core.ResolvableType;
import static org.awaitility.Awaitility.await;
@Testcontainers
@ExtendWith(MockitoExtension.class)
public abstract class KafkaConsumerTest implements TestTracingAwareSupplier {
protected String testTopic;
protected Tracer tracer = tracerTest().tracing().tracer();
protected Propagator propagator = tracerTest().tracing().propagator();
protected TestSpanHandler spans = tracerTest().handler();
protected TracingKafkaConsumer<String, String> kafkaConsumer;
private final AtomicBoolean consumerRun = new AtomicBoolean();
protected final AtomicInteger receivedCounter = new AtomicInteger(0);
@Mock(answer = Answers.RETURNS_DEEP_STUBS)
BeanFactory beanFactory;
@Container
protected static final KafkaContainer kafkaContainer = new KafkaContainer(
DockerImageName.parse("confluentinc/cp-kafka:6.1.1")).withExposedPorts(9093)
.waitingFor(Wait.forListeningPort());
@BeforeAll
static void setupAll() {
kafkaContainer.start();
}
@AfterAll
static void destroyAll() {
kafkaContainer.stop();
}
@BeforeEach
void setup() {
BDDMockito.given(this.beanFactory.getBean(Propagator.class)).willReturn(this.propagator);
BDDMockito.given(this.beanFactory.getBeanProvider(ResolvableType.forClassWithGenerics(Propagator.Getter.class,
ResolvableType.forType(new ParameterizedTypeReference<ConsumerRecord<?, ?>>() {
}))).getIfAvailable()).willReturn(new TracingKafkaPropagatorGetter());
testTopic = UUID.randomUUID().toString();
Map<String, Object> consumerProperties = new HashMap<>();
consumerProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaContainer.getBootstrapServers());
consumerProperties.put(ConsumerConfig.GROUP_ID_CONFIG, "test-consumer-group");
consumerProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
consumerProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
kafkaConsumer = new TracingKafkaConsumer<>(new KafkaConsumer<>(consumerProperties), beanFactory);
consumerRun.set(true);
Executors.newSingleThreadExecutor().execute(() -> doStartKafkaConsumer(receivedCounter));
}
@AfterEach
void destroy() {
consumerRun.set(false);
}
@Test
public void should_create_and_finish_consumer_span() {
KafkaProducer<String, String> kafkaProducer = KafkaTestUtils
.buildTestKafkaProducer(kafkaContainer.getBootstrapServers());
ProducerRecord<String, String> producerRecord = new ProducerRecord<>(testTopic, "test", "test");
kafkaProducer.send(producerRecord);
kafkaProducer.close();
await().atMost(Duration.ofSeconds(5)).until(() -> receivedCounter.intValue() == 1);
BDDAssertions.then(this.tracer.currentSpan()).isNull();
BDDAssertions.then(this.spans).hasSize(1);
FinishedSpan span = this.spans.get(0);
BDDAssertions.then(span.getKind()).isEqualTo(Span.Kind.CONSUMER);
BDDAssertions.then(span.getTags()).isNotEmpty();
BDDAssertions.then(span.getTags().get("kafka.topic")).isEqualTo(testTopic);
BDDAssertions.then(span.getTags().get("kafka.offset")).isEqualTo("0");
BDDAssertions.then(span.getTags().get("kafka.partition")).isEqualTo("0");
}
private void doStartKafkaConsumer(AtomicInteger receivedCounter) {
this.kafkaConsumer.subscribe(Pattern.compile(this.testTopic));
while (this.consumerRun.get()) {
ConsumerRecords<String, String> records = this.kafkaConsumer.poll(Duration.ofSeconds(1));
for (ConsumerRecord<String, String> record : records) {
receivedCounter.incrementAndGet();
}
}
this.kafkaConsumer.close();
}
@Override
public void cleanUpTracing() {
this.spans.clear();
}
}

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

@@ -19,34 +19,56 @@ package org.springframework.cloud.sleuth.instrument.kafka;
import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.regex.Pattern;
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.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.AfterAll;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
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.junit.jupiter.MockitoExtension;
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.beans.factory.BeanFactory;
import org.springframework.cloud.sleuth.Span;
import org.springframework.cloud.sleuth.Tracer;
import org.springframework.cloud.sleuth.exporter.FinishedSpan;
import org.springframework.cloud.sleuth.propagation.Propagator;
import org.springframework.cloud.sleuth.test.TestSpanHandler;
import org.springframework.cloud.sleuth.test.TestTracingAwareSupplier;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.core.ResolvableType;
import static org.awaitility.Awaitility.await;
@Testcontainers
@ExtendWith(MockitoExtension.class)
public abstract class KafkaProducerTest implements TestTracingAwareSupplier {
protected String testTopic;
protected Tracer tracer = tracerTest().tracing().tracer();
protected Propagator propagator = tracerTest().tracing().propagator();
@@ -55,39 +77,82 @@ public abstract class KafkaProducerTest implements TestTracingAwareSupplier {
protected TracingKafkaProducer<String, String> kafkaProducer;
private final AtomicBoolean consumerRun = new AtomicBoolean();
protected final BlockingQueue<ConsumerRecord<String, String>> consumerRecords = new LinkedBlockingQueue<>();
@Mock(answer = Answers.RETURNS_DEEP_STUBS)
BeanFactory beanFactory;
@Container
protected final KafkaContainer kafkaContainer = new KafkaContainer(
DockerImageName.parse("confluentinc/cp-kafka:5.2.1")).withExposedPorts(9093)
protected static final KafkaContainer kafkaContainer = new KafkaContainer(
DockerImageName.parse("confluentinc/cp-kafka:6.1.1")).withExposedPorts(9093)
.waitingFor(Wait.forListeningPort());
@BeforeAll
static void setupAll() {
kafkaContainer.start();
}
@AfterAll
static void destroyAll() {
kafkaContainer.stop();
}
@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());
BDDMockito.given(this.beanFactory.getBean(Tracer.class)).willReturn(this.tracer);
BDDMockito.given(this.beanFactory.getBean(Propagator.class)).willReturn(this.propagator);
BDDMockito.given(this.beanFactory.getBeanProvider(ResolvableType.forClassWithGenerics(Propagator.Setter.class,
ResolvableType.forType(new ParameterizedTypeReference<ProducerRecord<?, ?>>() {
}))).getIfAvailable()).willReturn(new TracingKafkaPropagatorSetter());
testTopic = UUID.randomUUID().toString();
Map<String, Object> producerProperties = new HashMap<>();
producerProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaContainer.getBootstrapServers());
producerProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
producerProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
kafkaProducer = new TracingKafkaProducer<>(new KafkaProducer<>(producerProperties), beanFactory);
consumerRun.set(true);
consumerRecords.clear();
}
@AfterEach
void destroy() {
kafkaContainer.stop();
this.kafkaProducer.close();
consumerRun.set(false);
}
@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");
ProducerRecord<String, String> producerRecord = new ProducerRecord<>(testTopic, "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);
BDDAssertions.then(this.spans).hasSize(1);
FinishedSpan span = this.spans.get(0);
BDDAssertions.then(span.getKind()).isEqualTo(Span.Kind.PRODUCER);
BDDAssertions.then(span.getTags().get("kafka.topic")).isEqualTo(testTopic);
}
protected void startKafkaConsumer() {
Executors.newSingleThreadExecutor().execute(this::doStartKafkaConsumer);
}
private void doStartKafkaConsumer() {
KafkaConsumer<String, String> kafkaConsumer = KafkaTestUtils
.buildTestKafkaConsumer(kafkaContainer.getBootstrapServers());
kafkaConsumer.subscribe(Pattern.compile(testTopic));
while (consumerRun.get()) {
ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofSeconds(1));
for (ConsumerRecord<String, String> record : records) {
consumerRecords.offer(record);
}
}
kafkaConsumer.close();
}
@Override

150
tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/KafkaReceiverTest.java Normal file
View File

@@ -0,0 +1,150 @@
/*
* 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.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.assertj.core.api.BDDAssertions;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
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.junit.jupiter.MockitoExtension;
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 reactor.core.Disposable;
import reactor.core.scheduler.Schedulers;
import reactor.kafka.receiver.KafkaReceiver;
import reactor.kafka.receiver.ReceiverOptions;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.cloud.sleuth.Span;
import org.springframework.cloud.sleuth.Tracer;
import org.springframework.cloud.sleuth.exporter.FinishedSpan;
import org.springframework.cloud.sleuth.propagation.Propagator;
import org.springframework.cloud.sleuth.test.TestSpanHandler;
import org.springframework.cloud.sleuth.test.TestTracingAwareSupplier;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.core.ResolvableType;
import static org.awaitility.Awaitility.await;
@Testcontainers
@ExtendWith(MockitoExtension.class)
public abstract class KafkaReceiverTest implements TestTracingAwareSupplier {
protected String testTopic;
protected Tracer tracer = tracerTest().tracing().tracer();
protected Propagator propagator = tracerTest().tracing().propagator();
protected TestSpanHandler spans = tracerTest().handler();
private Disposable consumerSubscription;
protected final AtomicInteger receivedCounter = new AtomicInteger(0);
@Mock(answer = Answers.RETURNS_DEEP_STUBS)
BeanFactory beanFactory;
@Container
protected static final KafkaContainer kafkaContainer = new KafkaContainer(
DockerImageName.parse("confluentinc/cp-kafka:6.1.1")).withExposedPorts(9093)
.waitingFor(Wait.forListeningPort());
@BeforeAll
static void setupAll() {
kafkaContainer.start();
}
@AfterAll
static void destroyAll() {
kafkaContainer.stop();
}
@BeforeEach
void setup() {
BDDMockito.given(this.beanFactory.getBean(Propagator.class)).willReturn(this.propagator);
BDDMockito.given(this.beanFactory.getBeanProvider(ResolvableType.forClassWithGenerics(Propagator.Getter.class,
ResolvableType.forType(new ParameterizedTypeReference<ConsumerRecord<?, ?>>() {
}))).getIfAvailable()).willReturn(new TracingKafkaPropagatorGetter());
testTopic = UUID.randomUUID().toString();
Map<String, Object> consumerProperties = new HashMap<>();
consumerProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaContainer.getBootstrapServers());
consumerProperties.put(ConsumerConfig.GROUP_ID_CONFIG, "test-consumer-group");
consumerProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
consumerProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
ReceiverOptions<String, String> options = ReceiverOptions.create(consumerProperties);
options = options.withKeyDeserializer(new StringDeserializer()).withValueDeserializer(new StringDeserializer())
.subscription(Collections.singletonList(testTopic));
KafkaReceiver<String, String> kafkaReceiver = KafkaReceiver.create(new TracingKafkaConsumerFactory(beanFactory),
options);
this.consumerSubscription = kafkaReceiver.receive().subscribeOn(Schedulers.single())
.subscribe(record -> receivedCounter.incrementAndGet());
this.receivedCounter.set(0);
}
@AfterEach
void destroy() {
this.consumerSubscription.dispose();
}
@Test
public void should_create_and_finish_consumer_span() {
KafkaProducer<String, String> kafkaProducer = KafkaTestUtils
.buildTestKafkaProducer(kafkaContainer.getBootstrapServers());
ProducerRecord<String, String> producerRecord = new ProducerRecord<>(testTopic, "test", "test");
kafkaProducer.send(producerRecord);
await().atMost(Duration.ofSeconds(5)).until(() -> receivedCounter.intValue() == 1);
BDDAssertions.then(this.tracer.currentSpan()).isNull();
BDDAssertions.then(this.spans).hasSize(1);
FinishedSpan span = this.spans.get(0);
BDDAssertions.then(span.getKind()).isEqualTo(Span.Kind.CONSUMER);
BDDAssertions.then(span.getTags()).isNotEmpty();
BDDAssertions.then(span.getTags().get("kafka.topic")).isEqualTo(testTopic);
BDDAssertions.then(span.getTags().get("kafka.offset")).isEqualTo("0");
BDDAssertions.then(span.getTags().get("kafka.partition")).isEqualTo("0");
}
@Override
public void cleanUpTracing() {
this.spans.clear();
}
}

169
tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/KafkaSenderTest.java Normal file
View File

@@ -0,0 +1,169 @@
/*
* 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.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.regex.Pattern;
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.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.AfterAll;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
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.junit.jupiter.MockitoExtension;
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 reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.kafka.sender.KafkaSender;
import reactor.kafka.sender.SenderOptions;
import reactor.kafka.sender.SenderRecord;
import reactor.kafka.sender.SenderResult;
import reactor.test.StepVerifier;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.cloud.sleuth.Span;
import org.springframework.cloud.sleuth.Tracer;
import org.springframework.cloud.sleuth.exporter.FinishedSpan;
import org.springframework.cloud.sleuth.propagation.Propagator;
import org.springframework.cloud.sleuth.test.TestSpanHandler;
import org.springframework.cloud.sleuth.test.TestTracingAwareSupplier;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.core.ResolvableType;
@Testcontainers
@ExtendWith(MockitoExtension.class)
public abstract class KafkaSenderTest implements TestTracingAwareSupplier {
protected String testTopic;
protected Tracer tracer = tracerTest().tracing().tracer();
protected Propagator propagator = tracerTest().tracing().propagator();
protected TestSpanHandler spans = tracerTest().handler();
protected KafkaSender<String, String> kafkaSender;
private final AtomicBoolean consumerRun = new AtomicBoolean();
protected final BlockingQueue<ConsumerRecord<String, String>> consumerRecords = new LinkedBlockingQueue<>();
@Mock(answer = Answers.RETURNS_DEEP_STUBS)
BeanFactory beanFactory;
@Container
protected static final KafkaContainer kafkaContainer = new KafkaContainer(
DockerImageName.parse("confluentinc/cp-kafka:6.1.1")).withExposedPorts(9093)
.waitingFor(Wait.forListeningPort());
@BeforeAll
static void setupAll() {
kafkaContainer.start();
}
@AfterAll
static void destroyAll() {
kafkaContainer.stop();
}
@BeforeEach
void setup() {
BDDMockito.given(this.beanFactory.getBean(Tracer.class)).willReturn(this.tracer);
BDDMockito.given(this.beanFactory.getBean(Propagator.class)).willReturn(this.propagator);
BDDMockito.given(this.beanFactory.getBeanProvider(ResolvableType.forClassWithGenerics(Propagator.Setter.class,
ResolvableType.forType(new ParameterizedTypeReference<ProducerRecord<?, ?>>() {
}))).getIfAvailable()).willReturn(new TracingKafkaPropagatorSetter());
testTopic = UUID.randomUUID().toString();
Map<String, Object> producerProperties = new HashMap<>();
producerProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaContainer.getBootstrapServers());
producerProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
producerProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
this.kafkaSender = KafkaSender.create(new TracingKafkaProducerFactory(beanFactory),
SenderOptions.create(producerProperties));
consumerRun.set(true);
consumerRecords.clear();
}
@AfterEach
void destroy() {
consumerRun.set(false);
}
@Test
public void should_create_and_finish_producer_span() throws InterruptedException {
ProducerRecord<String, String> producerRecord = new ProducerRecord<>(testTopic, "test", "test");
startKafkaConsumer();
Flux<SenderResult<Object>> senderResultFlux = this.kafkaSender
.send(Mono.just(SenderRecord.create(producerRecord, null)));
StepVerifier.create(senderResultFlux).expectNextCount(1).verifyComplete();
consumerRecords.poll(5, TimeUnit.SECONDS);
BDDAssertions.then(this.tracer.currentSpan()).isNull();
BDDAssertions.then(this.spans).hasSize(1);
FinishedSpan span = this.spans.get(0);
BDDAssertions.then(span.getKind()).isEqualTo(Span.Kind.PRODUCER);
BDDAssertions.then(span.getTags().get("kafka.topic")).isEqualTo(testTopic);
}
protected void startKafkaConsumer() {
Executors.newSingleThreadExecutor().execute(this::doStartKafkaConsumer);
}
private void doStartKafkaConsumer() {
KafkaConsumer<String, String> kafkaConsumer = KafkaTestUtils
.buildTestKafkaConsumer(kafkaContainer.getBootstrapServers());
kafkaConsumer.subscribe(Pattern.compile(this.testTopic));
while (this.consumerRun.get()) {
ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofSeconds(1));
for (ConsumerRecord<String, String> record : records) {
this.consumerRecords.offer(record);
}
}
kafkaConsumer.close();
}
@Override
public void cleanUpTracing() {
this.spans.clear();
}
}

52
tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/kafka/KafkaTestUtils.java Normal file
View File

@@ -0,0 +1,52 @@
/*
* Copyright 2013-2021 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.sleuth.instrument.kafka;
import java.util.HashMap;
import java.util.Map;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
public final class KafkaTestUtils {
private KafkaTestUtils() {
}
public static KafkaProducer<String, String> buildTestKafkaProducer(String bootstrapServers) {
Map<String, Object> producerProperties = new HashMap<>();
producerProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
producerProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
producerProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
return new KafkaProducer<>(producerProperties);
}
public static KafkaConsumer<String, String> buildTestKafkaConsumer(String bootstrapServers) {
Map<String, Object> consumerProperties = new HashMap<>();
consumerProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
consumerProperties.put(ConsumerConfig.GROUP_ID_CONFIG, "test-consumer-group");
consumerProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
consumerProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
return new KafkaConsumer<>(consumerProperties);
}
}