diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/SleuthKafkaStreamsConfiguration.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/SleuthKafkaStreamsConfiguration.java index 32e0e6dc5..8570db8d9 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/SleuthKafkaStreamsConfiguration.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/SleuthKafkaStreamsConfiguration.java @@ -45,7 +45,8 @@ import org.springframework.kafka.config.StreamsBuilderFactoryBean; @ConditionalOnBean(Tracing.class) @AutoConfigureAfter({ TraceAutoConfiguration.class }) @OnMessagingEnabled -@ConditionalOnProperty(value = "spring.sleuth.messaging.kafka.streams.enabled", matchIfMissing = true) +@ConditionalOnProperty(value = "spring.sleuth.messaging.kafka.streams.enabled", + matchIfMissing = true) @ConditionalOnClass(KafkaStreams.class) public class SleuthKafkaStreamsConfiguration { @@ -55,7 +56,6 @@ public class SleuthKafkaStreamsConfiguration { /** * Expose {@link KafkaStreamsTracing} as bean to allow for filter/map/peek/transform * operations. - * * @param tracing Brave Tracing instance from TraceAutoConfiguration * @return instance for use in further manual instrumentation */ @@ -66,9 +66,9 @@ public class SleuthKafkaStreamsConfiguration { } /** - * Call {@link StreamsBuilderFactoryBean#setClientSupplier(org.apache.kafka.streams.KafkaClientSupplier)} with - * Brave's TracingKafkaClientSupplier. - * + * Call + * {@link StreamsBuilderFactoryBean#setClientSupplier(org.apache.kafka.streams.KafkaClientSupplier)} + * with Brave's TracingKafkaClientSupplier. * @param objectProvider provides KafkaStreamsTracing; prevents eager initialization * @return */ @@ -91,22 +91,29 @@ public class SleuthKafkaStreamsConfiguration { */ class KafkaStreamsBuilderFactoryBeanPostProcessor implements BeanPostProcessor { - private static final Log log = LogFactory.getLog(KafkaStreamsBuilderFactoryBeanPostProcessor.class); + private static final Log log = LogFactory + .getLog(KafkaStreamsBuilderFactoryBeanPostProcessor.class); private final ObjectProvider objectProvider; - KafkaStreamsBuilderFactoryBeanPostProcessor(ObjectProvider objectProvider) { + KafkaStreamsBuilderFactoryBeanPostProcessor( + ObjectProvider objectProvider) { this.objectProvider = objectProvider; } @Override - public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { + public Object postProcessAfterInitialization(Object bean, String beanName) + throws BeansException { if (bean instanceof StreamsBuilderFactoryBean) { - // KafkaStreamsTracing is created in SleuthKafkaStreamsConfiguration above, so should not be null here - KafkaStreamsTracing kafkaStreamsTracing = this.objectProvider.getIfAvailable(); - ((StreamsBuilderFactoryBean) bean).setClientSupplier(kafkaStreamsTracing.kafkaClientSupplier()); + // KafkaStreamsTracing is created in SleuthKafkaStreamsConfiguration above, so + // should not be null here + KafkaStreamsTracing kafkaStreamsTracing = this.objectProvider + .getIfAvailable(); + ((StreamsBuilderFactoryBean) bean) + .setClientSupplier(kafkaStreamsTracing.kafkaClientSupplier()); if (log.isDebugEnabled()) { - log.debug("StreamsBuilderFactoryBean bean is auto-configured to enable tracing."); + log.debug( + "StreamsBuilderFactoryBean bean is auto-configured to enable tracing."); } } return bean; diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/feign/TracingFeignClient.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/feign/TracingFeignClient.java index 601d487ca..3039ecb54 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/feign/TracingFeignClient.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/feign/TracingFeignClient.java @@ -176,7 +176,8 @@ final class TracingFeignClient implements Client { String url = delegate.url(); byte[] body = delegate.body(); Charset charset = delegate.charset(); - return Request.create(delegate.httpMethod(), url, headers, body, charset, delegate.requestTemplate()); + return Request.create(delegate.httpMethod(), url, headers, body, charset, + delegate.requestTemplate()); } } diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/SleuthKafkaStreamsConfigurationIntegrationTests.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/SleuthKafkaStreamsConfigurationIntegrationTests.java index 77cc0a07a..5aba918e5 100644 --- a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/SleuthKafkaStreamsConfigurationIntegrationTests.java +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/SleuthKafkaStreamsConfigurationIntegrationTests.java @@ -46,60 +46,60 @@ import static org.mockito.Mockito.verify; class SleuthKafkaStreamsConfigurationIntegrationTests { private final ApplicationContextRunner contextRunner = new ApplicationContextRunner() - .withConfiguration(AutoConfigurations.of( - TraceAutoConfiguration.class, + .withConfiguration(AutoConfigurations.of(TraceAutoConfiguration.class, SleuthKafkaStreamsConfiguration.class)) .withUserConfiguration(UserConfig.class); @Test void should_create_KafkaStreamsTracing() { - this.contextRunner - .run(context -> assertThat(context).hasSingleBean(KafkaStreamsTracing.class)); + this.contextRunner.run( + context -> assertThat(context).hasSingleBean(KafkaStreamsTracing.class)); } @Test void should_not_create_KafkaStreamsTracing_when_KafkaStreams_not_present() { - this.contextRunner - .withClassLoader(new FilteredClassLoader(KafkaStreams.class)) - .run(context -> assertThat(context).doesNotHaveBean(KafkaStreamsTracing.class)); + this.contextRunner.withClassLoader(new FilteredClassLoader(KafkaStreams.class)) + .run(context -> assertThat(context) + .doesNotHaveBean(KafkaStreamsTracing.class)); } @Test void should_not_create_KafkaStreamsTracing_when_kafkastreams_disabled() { this.contextRunner .withPropertyValues("spring.sleuth.messaging.kafka.streams.enabled=false") - .run(context -> assertThat(context).doesNotHaveBean(KafkaStreamsTracing.class)); + .run(context -> assertThat(context) + .doesNotHaveBean(KafkaStreamsTracing.class)); } @Test void should_not_create_KafkaStreamsTracing_when_messaging_disabled() { - this.contextRunner - .withPropertyValues("spring.sleuth.messaging.enabled=false") - .run(context -> assertThat(context).doesNotHaveBean(KafkaStreamsTracing.class)); + this.contextRunner.withPropertyValues("spring.sleuth.messaging.enabled=false") + .run(context -> assertThat(context) + .doesNotHaveBean(KafkaStreamsTracing.class)); } @Test void should_set_KafkaClientSupplier_on_StreamsBuilderFactoryBean() { - this.contextRunner - .run(context -> verify(UserConfig.streamsBuilderFactoryBean) - .setClientSupplier(any(KafkaClientSupplier.class))); + this.contextRunner.run(context -> verify(UserConfig.streamsBuilderFactoryBean) + .setClientSupplier(any(KafkaClientSupplier.class))); } @Test void should_not_complain_about_eager_initialization() { - this.contextRunner - .withUserConfiguration(EagerInitializationConfig.class) + this.contextRunner.withUserConfiguration(EagerInitializationConfig.class) .run(context -> verify(UserConfig.streamsBuilderFactoryBean) .setClientSupplier(any(KafkaClientSupplier.class))); } @AfterEach void afterEach(CapturedOutput output) { - assertThat(output).doesNotContain("is not eligible for getting processed by all BeanPostProcessors"); + assertThat(output).doesNotContain( + "is not eligible for getting processed by all BeanPostProcessors"); } @Configuration static class UserConfig { + static StreamsBuilderFactoryBean streamsBuilderFactoryBean; @Bean @@ -107,25 +107,31 @@ class SleuthKafkaStreamsConfigurationIntegrationTests { streamsBuilderFactoryBean = mock(StreamsBuilderFactoryBean.class); return UserConfig.streamsBuilderFactoryBean; } + } @Configuration static class EagerInitializationConfig { + @Bean EagerInitializationComponent eagerInitializationComponent() { return new EagerInitializationComponent(); } + } static class EagerInitializationComponent { @Autowired private Tracing tracing; + private KafkaStreamsTracing kafkaStreamsTracing; @PostConstruct void init() { kafkaStreamsTracing = KafkaStreamsTracing.create(tracing); } + } + } diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/feign/TracingFeignClientTests.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/feign/TracingFeignClientTests.java index 55f21872e..14ab048fc 100644 --- a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/feign/TracingFeignClientTests.java +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/feign/TracingFeignClientTests.java @@ -28,24 +28,32 @@ import brave.http.HttpTracing; import brave.propagation.StrictCurrentTraceContext; import feign.Client; import feign.Request; +import feign.RequestTemplate; import org.assertj.core.api.BDDAssertions; import org.junit.After; +import org.junit.Assert; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.BDDMockito; import org.mockito.Mock; +import org.mockito.invocation.InvocationOnMock; import org.mockito.junit.MockitoJUnitRunner; +import org.mockito.stubbing.Answer; import static org.assertj.core.api.BDDAssertions.then; /** * @author Marcin Grzejszczak + * @author Hash.Jang */ @RunWith(MockitoJUnitRunner.class) public class TracingFeignClientTests { - Request request = Request.create("GET", "https://foo", new HashMap<>(), null, null); + RequestTemplate requestTemplate = new RequestTemplate(); + + Request request = Request.create(Request.HttpMethod.GET, "https://foo", + new HashMap<>(), null, null, requestTemplate); Request.Options options = new Request.Options(); @@ -112,4 +120,18 @@ public class TracingFeignClientTests { then(this.spans.get(0).tags()).containsEntry("error", "exception has occurred"); } + @Test + public void keep_requestTemplate() throws IOException { + BDDMockito.given(this.client.execute(BDDMockito.any(), BDDMockito.any())) + .willAnswer(new Answer() { + public Object answer(InvocationOnMock invocation) { + Object[] args = invocation.getArguments(); + Assert.assertEquals(((Request) args[0]).requestTemplate(), + requestTemplate); + return null; + } + }); + this.traceFeignClient.execute(this.request, this.options); + } + }