diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/async/ExecutorBeanPostProcessor.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/async/ExecutorBeanPostProcessor.java index 107abf03f..116568a6b 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/async/ExecutorBeanPostProcessor.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/async/ExecutorBeanPostProcessor.java @@ -20,7 +20,11 @@ import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; import java.lang.reflect.Modifier; import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.function.Function; +import java.util.function.Supplier; +import org.aopalliance.aop.Advice; import org.aopalliance.intercept.MethodInterceptor; import org.aopalliance.intercept.MethodInvocation; import org.apache.commons.logging.Log; @@ -64,42 +68,64 @@ class ExecutorBeanPostProcessor implements BeanPostProcessor { @Override public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { - if (bean instanceof Executor && !(bean instanceof ThreadPoolTaskExecutor)) { - Method execute = ReflectionUtils.findMethod(bean.getClass(), "execute", - Runnable.class); - boolean methodFinal = Modifier.isFinal(execute.getModifiers()); - boolean classFinal = Modifier.isFinal(bean.getClass().getModifiers()); - boolean cglibProxy = !methodFinal && !classFinal; - Executor executor = (Executor) bean; - try { - return createProxy(bean, cglibProxy, executor); - } - catch (AopConfigException ex) { - if (cglibProxy) { - if (log.isDebugEnabled()) { - log.debug( - "Exception occurred while trying to create a proxy, falling back to JDK proxy", - ex); - } - return createProxy(bean, false, executor); - } - throw ex; - } - } - else if (bean instanceof ThreadPoolTaskExecutor) { + if (bean instanceof ThreadPoolTaskExecutor) { if (isProxyNeeded(beanName)) { - boolean classFinal = Modifier.isFinal(bean.getClass().getModifiers()); - boolean cglibProxy = !classFinal; - ThreadPoolTaskExecutor executor = (ThreadPoolTaskExecutor) bean; - return createThreadPoolTaskExecutorProxy(bean, cglibProxy, executor); + return wrapThreadPoolTaskExecutor(bean); } else { log.info("Not instrumenting bean " + beanName); } + } else if (bean instanceof ExecutorService) { + if (isProxyNeeded(beanName)) { + return wrapExecutorService(bean); + } + else { + log.info("Not instrumenting bean " + beanName); + } + } else if (bean instanceof Executor) { + return wrapExecutor(bean); } return bean; } + private Object wrapExecutor(Object bean) { + Method execute = ReflectionUtils.findMethod(bean.getClass(), "execute", + Runnable.class); + boolean methodFinal = Modifier.isFinal(execute.getModifiers()); + boolean classFinal = Modifier.isFinal(bean.getClass().getModifiers()); + boolean cglibProxy = !methodFinal && !classFinal; + Executor executor = (Executor) bean; + try { + return createProxy(bean, cglibProxy, + new ExecutorMethodInterceptor(executor, this.beanFactory)); + } + catch (AopConfigException ex) { + if (cglibProxy) { + if (log.isDebugEnabled()) { + log.debug( + "Exception occurred while trying to create a proxy, falling back to JDK proxy", + ex); + } + return createProxy(bean, false, new ExecutorMethodInterceptor(executor, this.beanFactory)); + } + throw ex; + } + } + + private Object wrapThreadPoolTaskExecutor(Object bean) { + boolean classFinal = Modifier.isFinal(bean.getClass().getModifiers()); + boolean cglibProxy = !classFinal; + ThreadPoolTaskExecutor executor = (ThreadPoolTaskExecutor) bean; + return createThreadPoolTaskExecutorProxy(bean, cglibProxy, executor); + } + + private Object wrapExecutorService(Object bean) { + boolean classFinal = Modifier.isFinal(bean.getClass().getModifiers()); + boolean cglibProxy = !classFinal; + ExecutorService executor = (ExecutorService) bean; + return createExecutorServiceProxy(bean, cglibProxy, executor); + } + boolean isProxyNeeded(String beanName) { SleuthAsyncProperties sleuthAsyncProperties = asyncConfigurationProperties(); return !sleuthAsyncProperties.getIgnoredBeans().contains(beanName); @@ -107,24 +133,47 @@ class ExecutorBeanPostProcessor implements BeanPostProcessor { Object createThreadPoolTaskExecutorProxy(Object bean, boolean cglibProxy, ThreadPoolTaskExecutor executor) { + return getProxiedObject(bean, cglibProxy, executor, + () -> new LazyTraceThreadPoolTaskExecutor(this.beanFactory, executor)); + } + + Object createExecutorServiceProxy(Object bean, boolean cglibProxy, + ExecutorService executor) { + return getProxiedObject(bean, cglibProxy, executor, + () -> new TraceableExecutorService(this.beanFactory, executor)); + } + + private Object getProxiedObject(Object bean, boolean cglibProxy, Executor executor, + Supplier supplier) { ProxyFactoryBean factory = new ProxyFactoryBean(); factory.setProxyTargetClass(cglibProxy); - factory.addAdvice(new ExecutorMethodInterceptor(executor, + factory.addAdvice(new ExecutorMethodInterceptor(executor, this.beanFactory) { @Override - Executor executor(BeanFactory beanFactory, ThreadPoolTaskExecutor executor) { - return new LazyTraceThreadPoolTaskExecutor(beanFactory, executor); + T executor(BeanFactory beanFactory, T executor) { + return (T) supplier.get(); } }); factory.setTarget(bean); + try { + return getObject(factory); + } catch (Exception e) { + if (log.isDebugEnabled()) { + log.debug("Exception occurred while trying to get a proxy. Will fallback to a different implementation", e); + } + return supplier.get(); + } + } + + Object getObject(ProxyFactoryBean factory) { return factory.getObject(); } @SuppressWarnings("unchecked") - Object createProxy(Object bean, boolean cglibProxy, Executor executor) { + Object createProxy(Object bean, boolean cglibProxy, Advice advice) { ProxyFactoryBean factory = new ProxyFactoryBean(); factory.setProxyTargetClass(cglibProxy); - factory.addAdvice(new ExecutorMethodInterceptor(executor, this.beanFactory)); + factory.addAdvice(advice); factory.setTarget(bean); if (JavaVersion.current().isJava11Compatible()) { if (log.isDebugEnabled()) { @@ -134,7 +183,7 @@ class ExecutorBeanPostProcessor implements BeanPostProcessor { }); } - return factory.getObject(); + return getObject(factory); } private SleuthAsyncProperties asyncConfigurationProperties() { @@ -166,7 +215,7 @@ class ExecutorMethodInterceptor implements MethodInterceptor @Override public Object invoke(MethodInvocation invocation) throws Throwable { - Executor executor = executor(this.beanFactory, this.delegate); + T executor = executor(this.beanFactory, this.delegate); Method methodOnTracedBean = getMethod(invocation, executor); if (methodOnTracedBean != null) { try { @@ -187,8 +236,8 @@ class ExecutorMethodInterceptor implements MethodInterceptor method.getParameterTypes()); } - Executor executor(BeanFactory beanFactory, T executor) { - return new LazyTraceExecutor(beanFactory, executor); + T executor(BeanFactory beanFactory, T executor) { + return (T) new LazyTraceExecutor(beanFactory, executor); } } diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientAutoConfiguration.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientAutoConfiguration.java index 7691c890f..98ae2e68d 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientAutoConfiguration.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientAutoConfiguration.java @@ -19,10 +19,7 @@ package org.springframework.cloud.sleuth.instrument.web.client; import java.io.IOException; import java.util.ArrayList; import java.util.List; -import java.util.Map; -import java.util.Set; -import java.util.concurrent.atomic.AtomicReference; -import java.util.function.BiFunction; +import java.util.function.BiConsumer; import brave.Span; import brave.Tracer; @@ -34,18 +31,17 @@ import brave.propagation.Propagation; import brave.propagation.TraceContext; import brave.spring.web.TracingClientHttpRequestInterceptor; import io.netty.handler.codec.http.HttpHeaders; -import io.netty.handler.codec.http.HttpMethod; -import io.netty.handler.codec.http.HttpVersion; -import io.netty.handler.codec.http.cookie.Cookie; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; +import io.netty.util.Attribute; +import io.netty.util.AttributeKey; import org.apache.http.impl.client.HttpClientBuilder; import org.apache.http.impl.nio.client.HttpAsyncClientBuilder; -import org.aspectj.lang.ProceedingJoinPoint; -import org.aspectj.lang.annotation.Around; -import org.aspectj.lang.annotation.Aspect; -import org.aspectj.lang.annotation.Pointcut; -import org.reactivestreams.Publisher; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import reactor.netty.Connection; +import reactor.netty.http.client.HttpClient; +import reactor.netty.http.client.HttpClientRequest; +import reactor.netty.http.client.HttpClientResponse; + import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.ListableBeanFactory; @@ -70,11 +66,6 @@ import org.springframework.http.client.ClientHttpResponse; import org.springframework.security.oauth2.client.OAuth2RestTemplate; import org.springframework.web.client.RestTemplate; import org.springframework.web.reactive.function.client.WebClient; -import reactor.core.publisher.Mono; -import reactor.netty.NettyOutbound; -import reactor.netty.http.client.HttpClient; -import reactor.netty.http.client.HttpClientRequest; -import reactor.netty.http.client.HttpClientResponse; /** * {@link org.springframework.boot.autoconfigure.EnableAutoConfiguration @@ -165,8 +156,8 @@ public class TraceWebClientAutoConfiguration { static class NettyConfiguration { @Bean - public NettyAspect traceNetyAspect(HttpTracing httpTracing) { - return new NettyAspect(httpTracing); + public HttpClientBeanPostProcessor httpClientBeanPostProcessor(BeanFactory beanFactory) { + return new HttpClientBeanPostProcessor(beanFactory); } } @@ -318,40 +309,31 @@ class LazyTracingClientHttpRequestInterceptor implements ClientHttpRequestInterc } return this.interceptor; } - } -@Aspect -class NettyAspect { - private final TracingHttpClientInstrumentation instrumentation; +class HttpClientBeanPostProcessor implements BeanPostProcessor { - NettyAspect(HttpTracing httpTracing) { - this.instrumentation = TracingHttpClientInstrumentation.create(httpTracing); + private final BeanFactory beanFactory; + + HttpClientBeanPostProcessor(BeanFactory beanFactory) { + this.beanFactory = beanFactory; } - @Pointcut("execution(public * reactor.netty.http.client.HttpClient.RequestSender.send(..)) && args(function)") - private void anyHttpClientRequestSending( - BiFunction> function) { - } // NOSONAR - - @Around("anyHttpClientRequestSending(function)") - public Object wrapHttpClientRequestSending(ProceedingJoinPoint pjp, - BiFunction> function) - throws Throwable { - return Mono.defer(() -> { - try { - return this.instrumentation.wrapHttpClientRequestSending(pjp, function); - } - catch (Throwable e) { - return Mono.error(e); - } - }); + @Override + public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { + if (bean instanceof HttpClient) { + return ((HttpClient) bean) + .doOnRequest(TracingDoOnRequest.create(this.beanFactory)) + .doOnResponse(TracingDoOnResponse.create(this.beanFactory)); + } + return bean; } - } -class TracingHttpClientInstrumentation { +class TracingDoOnRequest implements BiConsumer { + + private static final Logger log = LoggerFactory.getLogger(TracingDoOnRequest.class); static final Propagation.Setter SETTER = new Propagation.Setter() { @Override @@ -378,206 +360,95 @@ class TracingHttpClientInstrumentation { } }; - private static final Log log = LogFactory - .getLog(TracingHttpClientInstrumentation.class); - final Tracer tracer; - final HttpClientHandler handler; - final TraceContext.Injector injector; - final HttpTracing httpTracing; - TracingHttpClientInstrumentation(HttpTracing httpTracing) { + TracingDoOnRequest(HttpTracing httpTracing) { this.tracer = httpTracing.tracing().tracer(); this.handler = HttpClientHandler.create(httpTracing, new HttpAdapter()); this.injector = httpTracing.tracing().propagation().injector(SETTER); this.httpTracing = httpTracing; } - static TracingHttpClientInstrumentation create(HttpTracing httpTracing) { - return new TracingHttpClientInstrumentation(httpTracing); + static TracingDoOnRequest create(BeanFactory beanFactory) { + return new TracingDoOnRequest(beanFactory.getBean(HttpTracing.class)); } - Mono wrapHttpClientRequestSending(ProceedingJoinPoint pjp, - BiFunction> function) - throws Throwable { - // add headers and set CS + @Override + public void accept(HttpClientRequest req, Connection connection) { final Span currentSpan = this.tracer.currentSpan(); - final AtomicReference span = new AtomicReference<>(); - BiFunction> combinedFunction = ( - req, nettyOutbound) -> { - try (Tracer.SpanInScope spanInScope = this.tracer - .withSpanInScope(currentSpan)) { - io.netty.handler.codec.http.HttpHeaders originalHeaders = req - .requestHeaders().copy(); - io.netty.handler.codec.http.HttpHeaders tracedHeaders = req - .requestHeaders(); - span.set(this.handler.handleSend(this.injector, tracedHeaders, req)); - if (log.isDebugEnabled()) { - log.debug("Handled send of " + span.get()); - } - io.netty.handler.codec.http.HttpHeaders addedHeaders = tracedHeaders - .copy(); - originalHeaders.forEach(header -> addedHeaders.remove(header.getKey())); - try (Tracer.SpanInScope clientInScope = this.tracer - .withSpanInScope(span.get())) { - if (log.isDebugEnabled()) { - log.debug("Created a new client span for Netty client"); - } - return handle(function, - new TracedHttpClientRequest(req, addedHeaders), - nettyOutbound); - } + try (Tracer.SpanInScope spanInScope = this.tracer + .withSpanInScope(currentSpan)) { + if (log.isDebugEnabled()) { + log.debug("Wrapping do on request"); } - }; - // run - Mono responseMono = (Mono) pjp - .proceed(new Object[] { combinedFunction }); - // get response - return responseMono.doOnSuccessOrError((httpClientResponse, throwable) -> { - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span.get())) { - // status codes and CR - if (span.get() != null) { - this.handler.handleReceive(httpClientResponse, throwable, span.get()); - if (log.isDebugEnabled()) { - log.debug("Setting client sent spans"); - } - } + Span span = this.handler.handleSend(this.injector, req + .requestHeaders(), req); + Attribute attribute = connection.channel() + .attr(AttributeKey.valueOf("span")); + attribute.set(span); + } + } +} + +class TracingDoOnResponse implements BiConsumer { + + private static final Logger log = LoggerFactory.getLogger(TracingDoOnResponse.class); + + final Tracer tracer; + final HttpClientHandler handler; + + TracingDoOnResponse(HttpTracing httpTracing) { + this.tracer = httpTracing.tracing().tracer(); + this.handler = HttpClientHandler.create(httpTracing, new HttpAdapter()); + } + + static TracingDoOnResponse create(BeanFactory beanFactory) { + return new TracingDoOnResponse(beanFactory.getBean(HttpTracing.class)); + } + + @Override + public void accept(HttpClientResponse httpClientResponse, Connection connection) { + Attribute spanAttr = connection.channel().attr(AttributeKey.valueOf("span")); + Span span = (Span) spanAttr.get(); + if (span == null) { + return; + } + try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span)) { + if (log.isDebugEnabled()) { + log.debug("Setting client sent spans"); } - }); + // status codes and CR + // TODO: Add throwable + this.handler.handleReceive(httpClientResponse, null, span); + } + } +} + +class HttpAdapter + extends brave.http.HttpClientAdapter { + + @Override + public String method(HttpClientRequest request) { + return request.method().name(); } - private Publisher handle( - BiFunction> handler, - HttpClientRequest req, NettyOutbound nettyOutbound) { - if (handler != null) { - return handler.apply(req, nettyOutbound); - } - return nettyOutbound; + @Override + public String url(HttpClientRequest request) { + return request.uri(); } - /** - * The `org.springframework.cloud.gateway.filter.NettyRoutingFilter` in SC Gateway is - * adding only these headers that were set when the request came in. That means that - * adding any additional headers (via instrumentation) is completely ignored. That's - * why we're wrapping the `HttpClientRequest` in such a wrapper that when `setHeaders` - * is called (that clears any current headers), will also add the tracing headers - */ - static class TracedHttpClientRequest implements HttpClientRequest { - - private final io.netty.handler.codec.http.HttpHeaders addedHeaders; - - private HttpClientRequest delegate; - - TracedHttpClientRequest(HttpClientRequest delegate, HttpHeaders addedHeaders) { - this.delegate = delegate; - this.addedHeaders = addedHeaders; - } - - @Override - public HttpClientRequest addCookie(Cookie cookie) { - this.delegate = this.delegate.addCookie(cookie); - return this; - } - - @Override - public HttpClientRequest addHeader(CharSequence name, CharSequence value) { - this.delegate = this.delegate.addHeader(name, value); - return this; - } - - @Override - public HttpClientRequest header(CharSequence name, CharSequence value) { - this.delegate = this.delegate.header(name, value); - return this; - } - - @Override - public HttpClientRequest headers(HttpHeaders headers) { - HttpHeaders copy = headers.copy(); - copy.add(this.addedHeaders); - this.delegate = this.delegate.headers(copy); - return this; - } - - @Override - public boolean isFollowRedirect() { - return this.delegate.isFollowRedirect(); - } - - @Override - public String[] redirectedFrom() { - return this.delegate.redirectedFrom(); - } - - @Override - public HttpHeaders requestHeaders() { - return this.delegate.requestHeaders(); - } - - @Override - public Map> cookies() { - return this.delegate.cookies(); - } - - @Override - public boolean isKeepAlive() { - return this.delegate.isKeepAlive(); - } - - @Override - public boolean isWebsocket() { - return this.delegate.isWebsocket(); - } - - @Override - public HttpMethod method() { - return this.delegate.method(); - } - - @Override - public String path() { - return this.delegate.path(); - } - - @Override - public String uri() { - return this.delegate.uri(); - } - - @Override - public HttpVersion version() { - return this.delegate.version(); - } - + @Override + public String requestHeader(HttpClientRequest request, String name) { + Object result = request.requestHeaders().get(name); + return result != null ? result.toString() : ""; } - static final class HttpAdapter - extends brave.http.HttpClientAdapter { - - @Override - public String method(HttpClientRequest request) { - return request.method().name(); - } - - @Override - public String url(HttpClientRequest request) { - return request.uri(); - } - - @Override - public String requestHeader(HttpClientRequest request, String name) { - Object result = request.requestHeaders().get(name); - return result != null ? result.toString() : ""; - } - - @Override - public Integer statusCode(HttpClientResponse response) { - return response.status().code(); - } - + @Override + public Integer statusCode(HttpClientResponse response) { + return response.status().code(); } } diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/async/ExecutorBeanPostProcessorTests.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/async/ExecutorBeanPostProcessorTests.java index 50b0bb2b7..e817ae6b7 100644 --- a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/async/ExecutorBeanPostProcessorTests.java +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/async/ExecutorBeanPostProcessorTests.java @@ -16,22 +16,36 @@ package org.springframework.cloud.sleuth.instrument.async; +import java.util.Collection; import java.util.Collections; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutionException; import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import brave.Tracer; import brave.Tracing; +import org.aopalliance.aop.Advice; +import org.junit.After; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; +import org.mockito.BDDMockito; import org.mockito.Mock; import org.mockito.Mockito; import org.mockito.junit.MockitoJUnitRunner; import org.springframework.aop.framework.AopConfigException; +import org.springframework.aop.framework.ProxyFactoryBean; import org.springframework.beans.factory.BeanFactory; import org.springframework.cloud.sleuth.DefaultSpanNamer; +import org.springframework.cloud.sleuth.SpanName; import org.springframework.cloud.sleuth.SpanNamer; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; @@ -49,6 +63,7 @@ public class ExecutorBeanPostProcessorTests { @Mock BeanFactory beanFactory; + Tracing tracing = Tracing.newBuilder().build(); private SleuthAsyncProperties sleuthAsyncProperties; @@ -57,6 +72,15 @@ public class ExecutorBeanPostProcessorTests { this.sleuthAsyncProperties = new SleuthAsyncProperties(); Mockito.when(beanFactory.getBean(SleuthAsyncProperties.class)) .thenReturn(this.sleuthAsyncProperties); + Mockito.when(beanFactory.getBean(Tracing.class)) + .thenReturn(this.tracing); + Mockito.when(beanFactory.getBean(SpanNamer.class)) + .thenReturn(new DefaultSpanNamer()); + } + + @After + public void clear() { + this.tracing.close(); } @Test @@ -78,30 +102,32 @@ public class ExecutorBeanPostProcessorTests { } @Test - public void should_create_jdk_proxy_when_cglib_fails_to_be_done() throws Exception { + public void should_fallback_to_sleuth_implementation_when_cglib_cannot_be_created() throws Exception { ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor(); Object o = new ExecutorBeanPostProcessor(this.beanFactory) .postProcessAfterInitialization(service, "foo"); - then(o).isInstanceOf(ScheduledExecutorService.class); - then(ClassUtils.isCglibProxy(o)).isFalse(); + then(o).isInstanceOf(TraceableExecutorService.class); service.shutdown(); } @Test - public void should_throw_exception_when_it_is_not_possible_to_create_any_proxy() + public void should_fallback_to_default_implementation_when_exception_thrown() throws Exception { ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor(); ExecutorBeanPostProcessor bpp = new ExecutorBeanPostProcessor(this.beanFactory) { + @Override - Object createProxy(Object bean, boolean cglibProxy, Executor executor) { + Object createProxy(Object bean, boolean cglibProxy, Advice advice) { throw new AopConfigException("foo"); } + }; - thenThrownBy(() -> bpp.postProcessAfterInitialization(service, "foo")) - .isInstanceOf(AopConfigException.class).hasMessage("foo"); + Object wrappedService = bpp.postProcessAfterInitialization(service, "foo"); + + then(wrappedService).isInstanceOf(TraceableExecutorService.class); service.shutdown(); } @@ -120,7 +146,7 @@ public class ExecutorBeanPostProcessorTests { } @Test - public void should_throw_exception_when_it_is_not_possible_to_create_any_proxyfor_ThreadPoolTaskExecutor() + public void should_throw_exception_when_it_is_not_possible_to_create_any_proxy_for_ThreadPoolTaskExecutor() throws Exception { ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor(); ExecutorBeanPostProcessor bpp = new ExecutorBeanPostProcessor(this.beanFactory) { @@ -135,6 +161,104 @@ public class ExecutorBeanPostProcessorTests { .isInstanceOf(AopConfigException.class).hasMessage("foo"); } + @Test + public void should_fallback_to_sleuth_impl_when_it_is_not_possible_to_create_any_proxy_for_ExecutorService() + throws Exception { + ExecutorService service = BDDMockito.mock(ExecutorService.class); + ExecutorBeanPostProcessor bpp = new ExecutorBeanPostProcessor(this.beanFactory) { + @Override + Object getObject(ProxyFactoryBean factory) { + throw new AopConfigException("foo"); + } + }; + + Object o = bpp.postProcessAfterInitialization(service, "foo"); + + then(o).isInstanceOf(TraceableExecutorService.class); + } + + @Test + public void should_throw_exception_from_the_wrapped_object() + throws Exception { + ExecutorService service = exceptionThrowingExecutorService(); + ExecutorBeanPostProcessor bpp = new ExecutorBeanPostProcessor(this.beanFactory); + + ExecutorService o = (ExecutorService) bpp.postProcessAfterInitialization(service, "foo"); + + thenThrownBy(() -> o.submit((Callable) () -> "hello")) + .hasMessage("foo") + .isInstanceOf(IllegalStateException.class); + } + + private ExecutorService exceptionThrowingExecutorService() { + return new ExecutorService() { + @Override + public void execute(Runnable command) { + + } + + @Override + public void shutdown() { + + } + + @Override + public List shutdownNow() { + return null; + } + + @Override + public boolean isShutdown() { + return false; + } + + @Override + public boolean isTerminated() { + return false; + } + + @Override + public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException { + return false; + } + + @Override + public Future submit(Callable task) { + throw new IllegalStateException("foo"); + } + + @Override + public Future submit(Runnable task, T result) { + return null; + } + + @Override + public Future submit(Runnable task) { + return null; + } + + @Override + public List> invokeAll(Collection> tasks) throws InterruptedException { + return null; + } + + @Override + public List> invokeAll(Collection> tasks, long timeout, TimeUnit unit) throws InterruptedException { + return null; + } + + @Override + public T invokeAny(Collection> tasks) throws InterruptedException, ExecutionException { + return null; + } + + @Override + public T invokeAny(Collection> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { + return null; + } + }; + } + @Test public void proxy_is_not_needed() throws Exception { this.sleuthAsyncProperties diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/integration/WebClientTests.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/integration/WebClientTests.java index 531a94466..7005feeff 100644 --- a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/integration/WebClientTests.java +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/integration/WebClientTests.java @@ -280,8 +280,6 @@ public class WebClientTests { then(this.reporter.getSpans()).isNotEmpty(); } - // TODO: Fix me - @Ignore("reactor is broken") @Test @SuppressWarnings("unchecked") public void shouldAttachTraceIdWhenCallingAnotherServiceForNettyHttpClient()