From f586661b78d096c4652f93a12caa4103f831ace2 Mon Sep 17 00:00:00 2001 From: Marcin Grzejszczak Date: Tue, 22 Oct 2019 12:41:53 +0200 Subject: [PATCH 1/2] Reusing input headers for Gateway instrumentation fixes gh-1469 --- .circleci/config.yml | 1 - .../client/TraceRequestHttpHeadersFilter.java | 58 +++++++++++-------- .../TraceRequestHttpHeadersFilterTests.java | 41 +++++++++++-- 3 files changed, 69 insertions(+), 31 deletions(-) diff --git a/.circleci/config.yml b/.circleci/config.yml index bbc27ed71..a9d1dd966 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -10,7 +10,6 @@ jobs: branches: ignore: - gh-pages # list of branches to ignore - resource_class: large steps: - checkout - restore_cache: diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceRequestHttpHeadersFilter.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceRequestHttpHeadersFilter.java index 4408c148c..5944aab85 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceRequestHttpHeadersFilter.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceRequestHttpHeadersFilter.java @@ -29,6 +29,7 @@ import org.springframework.cloud.gateway.filter.headers.HttpHeadersFilter; import org.springframework.http.HttpHeaders; import org.springframework.http.server.reactive.ServerHttpRequest; import org.springframework.http.server.reactive.ServerHttpResponse; +import org.springframework.lang.NonNull; import org.springframework.web.server.ServerWebExchange; final class TraceRequestHttpHeadersFilter extends AbstractHttpHeadersFilter { @@ -48,17 +49,16 @@ final class TraceRequestHttpHeadersFilter extends AbstractHttpHeadersFilter { if (log.isDebugEnabled()) { log.debug("Will instrument the HTTP request headers"); } - ServerHttpRequest.Builder builder = exchange.getRequest().mutate(); - Span span = this.handler.handleSend(this.injector, builder); + TraceCarrier carrier = new TraceCarrier(exchange.getRequest(), input); + Span span = this.handler.handleSend(this.injector, carrier); if (log.isDebugEnabled()) { - log.debug( - "Client span " + span + " created for the request. New headers are " - + builder.build().getHeaders().toSingleValueMap()); + log.debug("Client span " + span + " created for the request. New headers are " + + carrier.filteredHeaders.toSingleValueMap()); } exchange.getAttributes().put(SPAN_ATTRIBUTE, span); HttpHeaders headersWithInput = new HttpHeaders(); headersWithInput.addAll(input); - builder.build().getHeaders().forEach(headersWithInput::put); + carrier.filteredHeaders.forEach(headersWithInput::put); return headersWithInput; } @@ -69,10 +69,22 @@ final class TraceRequestHttpHeadersFilter extends AbstractHttpHeadersFilter { } +class TraceCarrier { + + final ServerHttpRequest originalRequest; + + final HttpHeaders filteredHeaders; + + TraceCarrier(@NonNull ServerHttpRequest originalRequest, @NonNull HttpHeaders filteredHeaders) { + this.originalRequest = originalRequest; + this.filteredHeaders = filteredHeaders; + } + +} + final class TraceResponseHttpHeadersFilter extends AbstractHttpHeadersFilter { - private static final Log log = LogFactory - .getLog(TraceResponseHttpHeadersFilter.class); + private static final Log log = LogFactory.getLog(TraceResponseHttpHeadersFilter.class); private TraceResponseHttpHeadersFilter(HttpTracing httpTracing) { super(httpTracing); @@ -109,23 +121,23 @@ abstract class AbstractHttpHeadersFilter implements HttpHeadersFilter { static final String SPAN_ATTRIBUTE = Span.class.getName(); - private static final Propagation.Setter SETTER = new Propagation.Setter() { + private static final Propagation.Setter SETTER = new Propagation.Setter() { @Override - public void put(ServerHttpRequest.Builder carrier, String key, String value) { - carrier.headers(httpHeaders -> httpHeaders.set(key, value)); + public void put(TraceCarrier carrier, String key, String value) { + carrier.filteredHeaders.set(key, value); } @Override public String toString() { - return "ServerHttpRequest.Builder::header"; + return "TraceCarrier::httpHeaders::set"; } }; final Tracer tracer; - final HttpClientHandler handler; + final HttpClientHandler handler; - final TraceContext.Injector injector; + final TraceContext.Injector injector; final HttpTracing httpTracing; @@ -136,29 +148,27 @@ abstract class AbstractHttpHeadersFilter implements HttpHeadersFilter { this.httpTracing = httpTracing; } - private static class ServerHttpAdapter extends - brave.http.HttpClientAdapter { + private static class ServerHttpAdapter extends brave.http.HttpClientAdapter { @Override - public String method(ServerHttpRequest.Builder request) { - return request.build().getMethodValue(); + public String method(TraceCarrier request) { + return request.originalRequest.getMethodValue(); } @Override - public String url(ServerHttpRequest.Builder request) { - return request.build().getURI().toString(); + public String url(TraceCarrier request) { + return request.originalRequest.getURI().toString(); } @Override - public String requestHeader(ServerHttpRequest.Builder request, String name) { - Object result = request.build().getHeaders().get(name); + public String requestHeader(TraceCarrier request, String name) { + Object result = request.filteredHeaders.get(name); return result != null ? result.toString() : ""; } @Override public Integer statusCode(ServerHttpResponse response) { - return response.getStatusCode() != null ? response.getStatusCode().value() - : null; + return response.getStatusCode() != null ? response.getStatusCode().value() : null; } } diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TraceRequestHttpHeadersFilterTests.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TraceRequestHttpHeadersFilterTests.java index 4917b4f4a..35ebce273 100644 --- a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TraceRequestHttpHeadersFilterTests.java +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TraceRequestHttpHeadersFilterTests.java @@ -48,14 +48,13 @@ public class TraceRequestHttpHeadersFilterTests { HttpHeadersFilter filter = TraceRequestHttpHeadersFilter.create(this.httpTracing); HttpHeaders httpHeaders = new HttpHeaders(); httpHeaders.set("X-Hello", "World"); - httpHeaders.set("X-Auth-User", "aaaa"); httpHeaders.set("X-B3-TraceId", "52f112af7472aff0"); httpHeaders.set("X-B3-SpanId", "53e6ab6fc5dfee58"); MockServerHttpRequest request = MockServerHttpRequest.post("foo/bar") .headers(httpHeaders).build(); MockServerWebExchange exchange = MockServerWebExchange.builder(request).build(); - HttpHeaders filteredHeaders = filter.filter(requestHeaders(), exchange); + HttpHeaders filteredHeaders = filter.filter(requestHeaders(httpHeaders), exchange); BDDAssertions.then(filteredHeaders.get("X-B3-TraceId")) .isNotEqualTo(httpHeaders.get("X-B3-TraceId")); @@ -81,7 +80,7 @@ public class TraceRequestHttpHeadersFilterTests { .headers(httpHeaders).build(); MockServerWebExchange exchange = MockServerWebExchange.builder(request).build(); - HttpHeaders filteredHeaders = filter.filter(requestHeaders(), exchange); + HttpHeaders filteredHeaders = filter.filter(requestHeaders(httpHeaders), exchange); BDDAssertions.then(filteredHeaders.get("X-B3-TraceId")).isNotEmpty(); BDDAssertions.then(filteredHeaders.get("X-B3-SpanId")).isNotEmpty(); @@ -95,18 +94,40 @@ public class TraceRequestHttpHeadersFilterTests { .isNotNull(); } - // #1352 + // #1469 @Test - public void should_set_tracing_headers_with_multiple_values() { + public void should_reuse_headers_only_from_input_since_exchange_may_contain_already_ignored_headers() { HttpHeadersFilter filter = TraceRequestHttpHeadersFilter.create(this.httpTracing); HttpHeaders httpHeaders = new HttpHeaders(); - httpHeaders.addAll("X-Hello", Arrays.asList("World1", "World2")); + httpHeaders.set("X-Hello", "World"); MockServerHttpRequest request = MockServerHttpRequest.post("foo/bar") .headers(httpHeaders).build(); MockServerWebExchange exchange = MockServerWebExchange.builder(request).build(); HttpHeaders filteredHeaders = filter.filter(requestHeaders(), exchange); + BDDAssertions.then(filteredHeaders.get("X-B3-TraceId")).isNotEmpty(); + BDDAssertions.then(filteredHeaders.get("X-B3-SpanId")).isNotEmpty(); + BDDAssertions.then(filteredHeaders.get("X-Hello")).isNullOrEmpty(); + BDDAssertions + .then((Object) exchange + .getAttribute(TraceRequestHttpHeadersFilter.SPAN_ATTRIBUTE)) + .isNotNull(); + } + + // #1352 + @Test + public void should_set_tracing_headers_with_multiple_values() { + HttpHeadersFilter filter = TraceRequestHttpHeadersFilter.create(this.httpTracing); + HttpHeaders httpHeaders = new HttpHeaders(); + httpHeaders.add("X-Hello-Request", "Request World"); + httpHeaders.addAll("X-Hello", Arrays.asList("World1", "World2")); + MockServerHttpRequest request = MockServerHttpRequest.post("foo/bar") + .headers(httpHeaders).build(); + MockServerWebExchange exchange = MockServerWebExchange.builder(request).build(); + + HttpHeaders filteredHeaders = filter.filter(httpHeaders, exchange); + BDDAssertions.then(filteredHeaders.get("X-B3-TraceId")).isNotEmpty(); BDDAssertions.then(filteredHeaders.get("X-B3-SpanId")).isNotEmpty(); BDDAssertions.then(filteredHeaders.get("X-Hello")) @@ -126,4 +147,12 @@ public class TraceRequestHttpHeadersFilterTests { return headers; } + private HttpHeaders requestHeaders(HttpHeaders originalHeaders) { + HttpHeaders headers = new HttpHeaders(); + headers.putAll(originalHeaders); + headers.add("X-Hello-Request", "Request World"); + headers.add("X-Auth-User", "aaaa"); + return headers; + } + } From 54a65bc9f33cce7683ee5ebdc633bf984ae1ed5b Mon Sep 17 00:00:00 2001 From: Marcin Grzejszczak Date: Tue, 22 Oct 2019 13:06:50 +0200 Subject: [PATCH 2/2] Ensures that there is no double instrumentation for executor services fixes gh-1463 --- .../async/ExecutorBeanPostProcessor.java | 107 +++++++----------- .../async/ExecutorBeanPostProcessorTests.java | 106 ++++++++--------- 2 files changed, 97 insertions(+), 116 deletions(-) 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 cb7b865f3..9b2144531 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 @@ -63,16 +63,14 @@ class ExecutorBeanPostProcessor implements BeanPostProcessor { } @Override - public Object postProcessBeforeInitialization(Object bean, String beanName) - throws BeansException { + public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { return bean; } @Override - public Object postProcessAfterInitialization(Object bean, String beanName) - throws BeansException { - if (bean instanceof ThreadPoolTaskExecutor - && !(bean instanceof LazyTraceThreadPoolTaskExecutor)) { + public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { + boolean alreadyTraced = alreadyTraced(bean); + if (bean instanceof ThreadPoolTaskExecutor && !alreadyTraced) { if (isProxyNeeded(beanName)) { return wrapThreadPoolTaskExecutor(bean); } @@ -80,8 +78,7 @@ class ExecutorBeanPostProcessor implements BeanPostProcessor { log.info("Not instrumenting bean " + beanName); } } - else if (bean instanceof ExecutorService - && !(bean instanceof TraceableExecutorService)) { + else if (bean instanceof ExecutorService && !alreadyTraced) { if (isProxyNeeded(beanName)) { return wrapExecutorService(bean); } @@ -89,8 +86,7 @@ class ExecutorBeanPostProcessor implements BeanPostProcessor { log.info("Not instrumenting bean " + beanName); } } - else if (bean instanceof AsyncTaskExecutor - && !(bean instanceof LazyTraceAsyncTaskExecutor)) { + else if (bean instanceof AsyncTaskExecutor && !alreadyTraced) { if (isProxyNeeded(beanName)) { return wrapAsyncTaskExecutor(bean); } @@ -98,30 +94,32 @@ class ExecutorBeanPostProcessor implements BeanPostProcessor { log.info("Not instrumenting bean " + beanName); } } - else if (bean instanceof Executor && !(bean instanceof LazyTraceExecutor)) { + else if (bean instanceof Executor && !alreadyTraced) { return wrapExecutor(bean); } return bean; } + private boolean alreadyTraced(Object bean) { + return bean instanceof LazyTraceThreadPoolTaskExecutor + || bean instanceof TraceableExecutorService || bean instanceof LazyTraceAsyncTaskExecutor + || bean instanceof LazyTraceExecutor; + } + private Object wrapExecutor(Object bean) { Executor executor = (Executor) bean; boolean methodFinal = anyFinalMethods(executor, Executor.class); boolean classFinal = Modifier.isFinal(bean.getClass().getModifiers()); boolean cglibProxy = !methodFinal && !classFinal; try { - return createProxy(bean, cglibProxy, - new ExecutorMethodInterceptor<>(executor, this.beanFactory)); + 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); + 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)); + return createProxy(bean, false, new ExecutorMethodInterceptor<>(executor, this.beanFactory)); } throw ex; } @@ -156,8 +154,7 @@ class ExecutorBeanPostProcessor implements BeanPostProcessor { return !sleuthAsyncProperties.getIgnoredBeans().contains(beanName); } - Object createThreadPoolTaskExecutorProxy(Object bean, boolean cglibProxy, - ThreadPoolTaskExecutor executor) { + Object createThreadPoolTaskExecutorProxy(Object bean, boolean cglibProxy, ThreadPoolTaskExecutor executor) { if (!cglibProxy) { return new LazyTraceThreadPoolTaskExecutor(this.beanFactory, executor); } @@ -165,45 +162,37 @@ class ExecutorBeanPostProcessor implements BeanPostProcessor { () -> new LazyTraceThreadPoolTaskExecutor(this.beanFactory, executor)); } - Supplier createThreadPoolTaskSchedulerProxy( - ThreadPoolTaskScheduler executor) { + Supplier createThreadPoolTaskSchedulerProxy(ThreadPoolTaskScheduler executor) { return () -> new LazyTraceThreadPoolTaskScheduler(this.beanFactory, executor); } - Supplier createScheduledThreadPoolExecutorProxy( - ScheduledThreadPoolExecutor executor) { - return () -> new LazyTraceScheduledThreadPoolExecutor(executor.getCorePoolSize(), - executor.getThreadFactory(), executor.getRejectedExecutionHandler(), - this.beanFactory, executor); + Supplier createScheduledThreadPoolExecutorProxy(ScheduledThreadPoolExecutor executor) { + return () -> new LazyTraceScheduledThreadPoolExecutor(executor.getCorePoolSize(), executor.getThreadFactory(), + executor.getRejectedExecutionHandler(), this.beanFactory, executor); } - Object createExecutorServiceProxy(Object bean, boolean cglibProxy, - ExecutorService executor) { + Object createExecutorServiceProxy(Object bean, boolean cglibProxy, ExecutorService executor) { return getProxiedObject(bean, cglibProxy, executor, () -> new TraceableExecutorService(this.beanFactory, executor)); } - Object createAsyncTaskExecutorProxy(Object bean, boolean cglibProxy, - AsyncTaskExecutor executor) { + Object createAsyncTaskExecutorProxy(Object bean, boolean cglibProxy, AsyncTaskExecutor executor) { return getProxiedObject(bean, cglibProxy, executor, () -> { if (bean instanceof ThreadPoolTaskScheduler) { - return new LazyTraceThreadPoolTaskScheduler(this.beanFactory, - (ThreadPoolTaskScheduler) executor); + return new LazyTraceThreadPoolTaskScheduler(this.beanFactory, (ThreadPoolTaskScheduler) executor); } return new LazyTraceAsyncTaskExecutor(this.beanFactory, executor); }); } - private Object getProxiedObject(Object bean, boolean cglibProxy, Executor executor, - Supplier supplier) { + private Object getProxiedObject(Object bean, boolean cglibProxy, Executor executor, Supplier supplier) { ProxyFactoryBean factory = proxyFactoryBean(bean, cglibProxy, executor, supplier); try { return getObject(factory); } catch (Exception ex) { if (log.isDebugEnabled()) { - log.debug( - "Exception occurred while trying to get a proxy. Will fallback to a different implementation", + log.debug("Exception occurred while trying to get a proxy. Will fallback to a different implementation", ex); } try { @@ -212,40 +201,35 @@ class ExecutorBeanPostProcessor implements BeanPostProcessor { log.debug( "Will wrap ThreadPoolTaskScheduler in its tracing representation due to previous errors"); } - return createThreadPoolTaskSchedulerProxy( - (ThreadPoolTaskScheduler) bean).get(); + return createThreadPoolTaskSchedulerProxy((ThreadPoolTaskScheduler) bean).get(); } else if (bean instanceof ScheduledThreadPoolExecutor) { if (log.isDebugEnabled()) { log.debug( "Will wrap ScheduledThreadPoolExecutor in its tracing representation due to previous errors"); } - return createScheduledThreadPoolExecutorProxy( - (ScheduledThreadPoolExecutor) bean).get(); + return createScheduledThreadPoolExecutorProxy((ScheduledThreadPoolExecutor) bean).get(); } } catch (Exception ex2) { if (log.isDebugEnabled()) { - log.debug( - "Fallback for special wrappers failed, will try the tracing representation instead", - ex2); + log.debug("Fallback for special wrappers failed, will try the tracing representation instead", ex2); } } return supplier.get(); } } - private ProxyFactoryBean proxyFactoryBean(Object bean, boolean cglibProxy, - Executor executor, Supplier supplier) { + private ProxyFactoryBean proxyFactoryBean(Object bean, boolean cglibProxy, Executor executor, + Supplier supplier) { ProxyFactoryBean factory = new ProxyFactoryBean(); factory.setProxyTargetClass(cglibProxy); - factory.addAdvice( - new ExecutorMethodInterceptor(executor, this.beanFactory) { - @Override - Executor executor(BeanFactory beanFactory, Executor executor) { - return supplier.get(); - } - }); + factory.addAdvice(new ExecutorMethodInterceptor(executor, this.beanFactory) { + @Override + Executor executor(BeanFactory beanFactory, Executor executor) { + return supplier.get(); + } + }); factory.setTarget(bean); return factory; } @@ -265,21 +249,17 @@ class ExecutorBeanPostProcessor implements BeanPostProcessor { private SleuthAsyncProperties asyncConfigurationProperties() { if (this.sleuthAsyncProperties == null) { - this.sleuthAsyncProperties = this.beanFactory - .getBean(SleuthAsyncProperties.class); + this.sleuthAsyncProperties = this.beanFactory.getBean(SleuthAsyncProperties.class); } return this.sleuthAsyncProperties; } private static boolean anyFinalMethods(T object, Class iface) { AtomicBoolean finalMethodPresent = new AtomicBoolean(); - ReflectionUtils.doWithMethods(iface, method -> finalMethodPresent.set(true), - method -> { - Method m = ReflectionUtils.findMethod(object.getClass(), - method.getName(), method.getParameterTypes()); - return m != null && !ReflectionUtils.isObjectMethod(m) - && Modifier.isFinal(m.getModifiers()); - }); + ReflectionUtils.doWithMethods(iface, method -> finalMethodPresent.set(true), method -> { + Method m = ReflectionUtils.findMethod(object.getClass(), method.getName(), method.getParameterTypes()); + return m != null && !ReflectionUtils.isObjectMethod(m) && Modifier.isFinal(m.getModifiers()); + }); return finalMethodPresent.get(); } @@ -321,8 +301,7 @@ class ExecutorMethodInterceptor implements MethodInterceptor private Method getMethod(MethodInvocation invocation, Object object) { Method method = invocation.getMethod(); - return ReflectionUtils.findMethod(object.getClass(), method.getName(), - method.getParameterTypes()); + return ReflectionUtils.findMethod(object.getClass(), method.getName(), method.getParameterTypes()); } T executor(BeanFactory beanFactory, T executor) { 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 661cb4797..f5f9ba032 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 @@ -33,6 +33,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import brave.Tracing; import org.aopalliance.aop.Advice; +import org.assertj.core.api.BDDAssertions; import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -71,8 +72,7 @@ public class ExecutorBeanPostProcessorTests { @Before public void setup() { this.sleuthAsyncProperties = new SleuthAsyncProperties(); - Mockito.when(this.beanFactory.getBean(SleuthAsyncProperties.class)) - .thenReturn(this.sleuthAsyncProperties); + Mockito.when(this.beanFactory.getBean(SleuthAsyncProperties.class)).thenReturn(this.sleuthAsyncProperties); } @After @@ -82,28 +82,24 @@ public class ExecutorBeanPostProcessorTests { @Test public void should_create_a_cglib_proxy_by_default() throws Exception { - Object o = new ExecutorBeanPostProcessor(this.beanFactory) - .postProcessAfterInitialization(new Foo(), "foo"); + Object o = new ExecutorBeanPostProcessor(this.beanFactory).postProcessAfterInitialization(new Foo(), "foo"); then(o).isInstanceOf(Foo.class); then(AopUtils.isCglibProxy(o)).isTrue(); } @Test - public void should_fallback_to_sleuth_implementation_when_cglib_cannot_be_created() - 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"); + Object o = new ExecutorBeanPostProcessor(this.beanFactory).postProcessAfterInitialization(service, "foo"); then(o).isInstanceOf(TraceableExecutorService.class); service.shutdown(); } @Test - public void should_fallback_to_default_implementation_when_exception_thrown() - throws Exception { + public void should_fallback_to_default_implementation_when_exception_thrown() throws Exception { ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor(); ExecutorBeanPostProcessor bpp = new ExecutorBeanPostProcessor(this.beanFactory) { @@ -121,8 +117,7 @@ public class ExecutorBeanPostProcessorTests { } @Test - public void should_create_a_cglib_proxy_by_default_for_ThreadPoolTaskExecutor() - throws Exception { + public void should_create_a_cglib_proxy_by_default_for_ThreadPoolTaskExecutor() throws Exception { Object o = new ExecutorBeanPostProcessor(this.beanFactory) .postProcessAfterInitialization(new FooThreadPoolTaskExecutor(), "foo"); @@ -136,8 +131,7 @@ public class ExecutorBeanPostProcessorTests { ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor(); ExecutorBeanPostProcessor bpp = new ExecutorBeanPostProcessor(this.beanFactory) { @Override - Object createThreadPoolTaskExecutorProxy(Object bean, boolean cglibProxy, - ThreadPoolTaskExecutor executor) { + Object createThreadPoolTaskExecutorProxy(Object bean, boolean cglibProxy, ThreadPoolTaskExecutor executor) { throw new AopConfigException("foo"); } }; @@ -167,8 +161,7 @@ public class ExecutorBeanPostProcessorTests { ExecutorService service = exceptionThrowingExecutorService(); ExecutorBeanPostProcessor bpp = new ExecutorBeanPostProcessor(this.beanFactory); - ExecutorService o = (ExecutorService) bpp.postProcessAfterInitialization(service, - "foo"); + ExecutorService o = (ExecutorService) bpp.postProcessAfterInitialization(service, "foo"); thenThrownBy(() -> o.submit((Callable) () -> "hello")).hasMessage("foo") .isInstanceOf(IllegalStateException.class); @@ -202,8 +195,7 @@ public class ExecutorBeanPostProcessorTests { } @Override - public boolean awaitTermination(long timeout, TimeUnit unit) - throws InterruptedException { + public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException { return false; } @@ -223,14 +215,13 @@ public class ExecutorBeanPostProcessorTests { } @Override - public List> invokeAll(Collection> tasks) - throws InterruptedException { + public List> invokeAll(Collection> tasks) throws InterruptedException { return null; } @Override - public List> invokeAll(Collection> tasks, - long timeout, TimeUnit unit) throws InterruptedException { + public List> invokeAll(Collection> tasks, long timeout, TimeUnit unit) + throws InterruptedException { return null; } @@ -241,8 +232,7 @@ public class ExecutorBeanPostProcessorTests { } @Override - public T invokeAny(Collection> tasks, long timeout, - TimeUnit unit) + public T invokeAny(Collection> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { return null; } @@ -251,11 +241,9 @@ public class ExecutorBeanPostProcessorTests { @Test public void should_use_jdk_proxy_when_executor_has_final_methods() { - ExecutorBeanPostProcessor beanPostProcessor = new ExecutorBeanPostProcessor( - this.beanFactory); + ExecutorBeanPostProcessor beanPostProcessor = new ExecutorBeanPostProcessor(this.beanFactory); Executor executor = Runnable::run; - Executor wrappedExecutor = (Executor) beanPostProcessor - .postProcessAfterInitialization(executor, "executor"); + Executor wrappedExecutor = (Executor) beanPostProcessor.postProcessAfterInitialization(executor, "executor"); then(AopUtils.isJdkDynamicProxy(wrappedExecutor)).isTrue(); then(AopUtils.isCglibProxy(wrappedExecutor)).isFalse(); @@ -268,10 +256,8 @@ public class ExecutorBeanPostProcessorTests { } @Test - public void should_use_jdk_proxy_when_executor_service_has_final_methods() - throws Exception { - ExecutorBeanPostProcessor beanPostProcessor = new ExecutorBeanPostProcessor( - this.beanFactory); + public void should_use_jdk_proxy_when_executor_service_has_final_methods() throws Exception { + ExecutorBeanPostProcessor beanPostProcessor = new ExecutorBeanPostProcessor(this.beanFactory); ExecutorService executorService = new DelegatingSecurityContextExecutorService( Executors.newSingleThreadExecutor()); ExecutorService wrappedExecutor = (ExecutorService) beanPostProcessor @@ -284,10 +270,8 @@ public class ExecutorBeanPostProcessorTests { } @Test - public void should_use_jdk_proxy_when_async_task_executor_has_final_methods() - throws Exception { - ExecutorBeanPostProcessor beanPostProcessor = new ExecutorBeanPostProcessor( - this.beanFactory); + public void should_use_jdk_proxy_when_async_task_executor_has_final_methods() throws Exception { + ExecutorBeanPostProcessor beanPostProcessor = new ExecutorBeanPostProcessor(this.beanFactory); AsyncTaskExecutor wrappedExecutor = (AsyncTaskExecutor) beanPostProcessor .postProcessAfterInitialization(new DirectTaskExecutor(), "taskExecutor"); @@ -299,13 +283,11 @@ public class ExecutorBeanPostProcessorTests { @Test public void should_fallback_to_sleuth_impl_when_thread_pool_task_executor_has_final_methods() { - ExecutorBeanPostProcessor postProcessor = new ExecutorBeanPostProcessor( - this.beanFactory); + ExecutorBeanPostProcessor postProcessor = new ExecutorBeanPostProcessor(this.beanFactory); ThreadPoolTaskExecutor threadPoolTaskExecutor = new PoolTaskExecutor(); ThreadPoolTaskExecutor wrappedTaskExecutor = (ThreadPoolTaskExecutor) postProcessor - .postProcessAfterInitialization(threadPoolTaskExecutor, - "threadPoolTaskExecutor"); + .postProcessAfterInitialization(threadPoolTaskExecutor, "threadPoolTaskExecutor"); then(wrappedTaskExecutor).isInstanceOf(LazyTraceThreadPoolTaskExecutor.class); then(AopUtils.isCglibProxy(wrappedTaskExecutor)).isFalse(); @@ -315,31 +297,26 @@ public class ExecutorBeanPostProcessorTests { @Test public void proxy_is_not_needed() throws Exception { - this.sleuthAsyncProperties - .setIgnoredBeans(Collections.singletonList("fooExecutor")); + this.sleuthAsyncProperties.setIgnoredBeans(Collections.singletonList("fooExecutor")); - boolean isProxyNeeded = new ExecutorBeanPostProcessor(this.beanFactory) - .isProxyNeeded("fooExecutor"); + boolean isProxyNeeded = new ExecutorBeanPostProcessor(this.beanFactory).isProxyNeeded("fooExecutor"); then(isProxyNeeded).isFalse(); } @Test public void proxy_is_needed() throws Exception { - boolean isProxyNeeded = new ExecutorBeanPostProcessor(this.beanFactory) - .isProxyNeeded("fooExecutor"); + boolean isProxyNeeded = new ExecutorBeanPostProcessor(this.beanFactory).isProxyNeeded("fooExecutor"); then(isProxyNeeded).isTrue(); } @Test public void should_not_create_proxy() throws Exception { - this.sleuthAsyncProperties - .setIgnoredBeans(Collections.singletonList("fooExecutor")); + this.sleuthAsyncProperties.setIgnoredBeans(Collections.singletonList("fooExecutor")); Object o = new ExecutorBeanPostProcessor(this.beanFactory) - .postProcessAfterInitialization(new ThreadPoolTaskExecutor(), - "fooExecutor"); + .postProcessAfterInitialization(new ThreadPoolTaskExecutor(), "fooExecutor"); then(o).isInstanceOf(ThreadPoolTaskExecutor.class); then(AopUtils.isCglibProxy(o)).isFalse(); @@ -348,8 +325,7 @@ public class ExecutorBeanPostProcessorTests { @Test public void should_throw_real_exception_when_using_proxy() throws Exception { Object o = new ExecutorBeanPostProcessor(this.beanFactory) - .postProcessAfterInitialization(new RejectedExecutionExecutor(), - "fooExecutor"); + .postProcessAfterInitialization(new RejectedExecutionExecutor(), "fooExecutor"); then(o).isInstanceOf(RejectedExecutionExecutor.class); then(AopUtils.isCglibProxy(o)).isTrue(); @@ -357,6 +333,32 @@ public class ExecutorBeanPostProcessorTests { })).isInstanceOf(RejectedExecutionException.class).hasMessage("rejected"); } + // #1463 + @Test + public void should_not_double_instrument_traced_executors() throws Exception { + LazyTraceThreadPoolTaskExecutor lazyTraceThreadPoolTaskExecutor = BDDMockito + .mock(LazyTraceThreadPoolTaskExecutor.class); + Object o = new ExecutorBeanPostProcessor(this.beanFactory) + .postProcessAfterInitialization(lazyTraceThreadPoolTaskExecutor, "executor"); + BDDAssertions.then(o).isSameAs(lazyTraceThreadPoolTaskExecutor); + + TraceableExecutorService traceableExecutorService = BDDMockito.mock(TraceableExecutorService.class); + o = new ExecutorBeanPostProcessor(this.beanFactory).postProcessAfterInitialization(traceableExecutorService, + "executor"); + BDDAssertions.then(o).isSameAs(traceableExecutorService); + + LazyTraceAsyncTaskExecutor lazyTraceAsyncTaskExecutor = BDDMockito.mock(LazyTraceAsyncTaskExecutor.class); + o = new ExecutorBeanPostProcessor(this.beanFactory).postProcessAfterInitialization(lazyTraceAsyncTaskExecutor, + "executor"); + BDDAssertions.then(o).isSameAs(lazyTraceAsyncTaskExecutor); + + LazyTraceExecutor lazyTraceExecutor = BDDMockito.mock(LazyTraceExecutor.class); + o = new ExecutorBeanPostProcessor(this.beanFactory).postProcessAfterInitialization(lazyTraceExecutor, + "executor"); + BDDAssertions.then(o).isSameAs(lazyTraceExecutor); + + } + class Foo implements Executor { @Override