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