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 443a82041..392014a73 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 @@ -104,15 +104,13 @@ class ExecutorBeanPostProcessor implements BeanPostProcessor { } private Object wrapExecutor(Object bean) { - Method execute = ReflectionUtils.findMethod(bean.getClass(), "execute", - Runnable.class); - boolean methodFinal = Modifier.isFinal(execute.getModifiers()); + Executor executor = (Executor) bean; + boolean methodFinal = anyFinalMethods(executor, Executor.class); 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)); + new ExecutorMethodInterceptor<>(executor, this.beanFactory)); } catch (AopConfigException ex) { if (cglibProxy) { @@ -122,46 +120,33 @@ class ExecutorBeanPostProcessor implements BeanPostProcessor { ex); } return createProxy(bean, false, - new ExecutorMethodInterceptor(executor, this.beanFactory)); + 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; + boolean classFinal = Modifier.isFinal(bean.getClass().getModifiers()); + boolean methodsFinal = anyFinalMethods(executor, ThreadPoolTaskExecutor.class); + boolean cglibProxy = !classFinal && !methodsFinal; return createThreadPoolTaskExecutorProxy(bean, cglibProxy, executor); } - private ProxyFactoryBean wrapThreadPoolTaskScheduler(Object bean) { - boolean classFinal = Modifier.isFinal(bean.getClass().getModifiers()); - boolean cglibProxy = !classFinal; - ThreadPoolTaskScheduler executor = (ThreadPoolTaskScheduler) bean; - return proxyFactoryBean(bean, cglibProxy, executor, - createThreadPoolTaskSchedulerProxy(executor)); - } - - private ProxyFactoryBean wrapScheduledThreadPoolExecutor(Object bean) { - boolean classFinal = Modifier.isFinal(bean.getClass().getModifiers()); - boolean cglibProxy = !classFinal; - ScheduledThreadPoolExecutor executor = (ScheduledThreadPoolExecutor) bean; - return proxyFactoryBean(bean, cglibProxy, executor, - createScheduledThreadPoolExecutorProxy(executor)); - } - private Object wrapExecutorService(Object bean) { - boolean classFinal = Modifier.isFinal(bean.getClass().getModifiers()); - boolean cglibProxy = !classFinal; ExecutorService executor = (ExecutorService) bean; + boolean classFinal = Modifier.isFinal(bean.getClass().getModifiers()); + boolean methodFinal = anyFinalMethods(executor, ExecutorService.class); + boolean cglibProxy = !classFinal && !methodFinal; return createExecutorServiceProxy(bean, cglibProxy, executor); } private Object wrapAsyncTaskExecutor(Object bean) { - boolean classFinal = Modifier.isFinal(bean.getClass().getModifiers()); - boolean cglibProxy = !classFinal; AsyncTaskExecutor executor = (AsyncTaskExecutor) bean; + boolean classFinal = Modifier.isFinal(bean.getClass().getModifiers()); + boolean methodsFinal = anyFinalMethods(executor, AsyncTaskExecutor.class); + boolean cglibProxy = !classFinal && !methodsFinal; return createAsyncTaskExecutorProxy(bean, cglibProxy, executor); } @@ -172,15 +157,18 @@ class ExecutorBeanPostProcessor implements BeanPostProcessor { Object createThreadPoolTaskExecutorProxy(Object bean, boolean cglibProxy, ThreadPoolTaskExecutor executor) { + if (!cglibProxy) { + return new LazyTraceThreadPoolTaskExecutor(this.beanFactory, executor); + } return getProxiedObject(bean, cglibProxy, executor, () -> new LazyTraceThreadPoolTaskExecutor(this.beanFactory, executor)); } - Supplier createThreadPoolTaskSchedulerProxy(ThreadPoolTaskScheduler executor) { + Supplier createThreadPoolTaskSchedulerProxy(ThreadPoolTaskScheduler executor) { return () -> new LazyTraceThreadPoolTaskScheduler(this.beanFactory, executor); } - Supplier createScheduledThreadPoolExecutorProxy( + Supplier createScheduledThreadPoolExecutorProxy( ScheduledThreadPoolExecutor executor) { return () -> new LazyTraceScheduledThreadPoolExecutor(executor.getCorePoolSize(), executor.getThreadFactory(), executor.getRejectedExecutionHandler(), @@ -247,8 +235,8 @@ class ExecutorBeanPostProcessor implements BeanPostProcessor { factory.addAdvice( new ExecutorMethodInterceptor(executor, this.beanFactory) { @Override - T executor(BeanFactory beanFactory, T executor) { - return (T) supplier.get(); + Executor executor(BeanFactory beanFactory, Executor executor) { + return supplier.get(); } }); factory.setTarget(bean); @@ -276,6 +264,16 @@ class ExecutorBeanPostProcessor implements BeanPostProcessor { return this.sleuthAsyncProperties; } + private static boolean anyFinalMethods(T object, Class iface) { + for (Method method : ReflectionUtils.getDeclaredMethods(iface)) { + Method m = ReflectionUtils.findMethod(object.getClass(), method.getName(), method.getParameterTypes()); + if (m != null && Modifier.isFinal(m.getModifiers())) { + return true; + } + } + return false; + } + } /** @@ -318,8 +316,7 @@ class ExecutorMethodInterceptor implements MethodInterceptor method.getParameterTypes()); } - T executor(BeanFactory beanFactory, T executor) { + T executor(BeanFactory beanFactory, T executor) { return (T) new LazyTraceExecutor(beanFactory, 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 cc6a4c5ff..e2dd9dd2a 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 @@ -29,6 +29,7 @@ import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; import brave.Tracing; import org.aopalliance.aop.Advice; @@ -43,9 +44,12 @@ import org.mockito.junit.MockitoJUnitRunner; import org.springframework.aop.framework.AopConfigException; import org.springframework.aop.framework.ProxyFactoryBean; +import org.springframework.aop.support.AopUtils; import org.springframework.beans.factory.BeanFactory; +import org.springframework.core.task.AsyncTaskExecutor; +import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; -import org.springframework.util.ClassUtils; +import org.springframework.security.concurrent.DelegatingSecurityContextExecutorService; import static org.assertj.core.api.BDDAssertions.then; import static org.assertj.core.api.BDDAssertions.thenThrownBy; @@ -82,7 +86,7 @@ public class ExecutorBeanPostProcessorTests { .postProcessAfterInitialization(new Foo(), "foo"); then(o).isInstanceOf(Foo.class); - then(ClassUtils.isCglibProxy(o)).isTrue(); + then(AopUtils.isCglibProxy(o)).isTrue(); } @Test @@ -123,7 +127,7 @@ public class ExecutorBeanPostProcessorTests { .postProcessAfterInitialization(new FooThreadPoolTaskExecutor(), "foo"); then(o).isInstanceOf(FooThreadPoolTaskExecutor.class); - then(ClassUtils.isCglibProxy(o)).isTrue(); + then(AopUtils.isCglibProxy(o)).isTrue(); } @Test @@ -245,6 +249,60 @@ public class ExecutorBeanPostProcessorTests { }; } + @Test + public void should_use_jdk_proxy_when_executor_has_final_methods() { + ExecutorBeanPostProcessor beanPostProcessor = new ExecutorBeanPostProcessor(this.beanFactory); + Executor executor = Runnable::run; + Executor wrappedExecutor = (Executor) beanPostProcessor.postProcessAfterInitialization( + executor, "executor"); + + then(AopUtils.isJdkDynamicProxy(wrappedExecutor)).isTrue(); + then(AopUtils.isCglibProxy(wrappedExecutor)).isFalse(); + + AtomicBoolean wasCalled = new AtomicBoolean(false); + wrappedExecutor.execute(() -> { wasCalled.set(true); }); + then(wasCalled).isTrue(); + } + + @Test + 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.postProcessAfterInitialization( + executorService, "executorService"); + + then(AopUtils.isJdkDynamicProxy(wrappedExecutor)).isTrue(); + then(AopUtils.isCglibProxy(wrappedExecutor)).isFalse(); + then(wrappedExecutor.submit(() -> "done").get()).isEqualTo("done"); + wrappedExecutor.shutdownNow(); + } + + @Test + 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"); + + then(AopUtils.isJdkDynamicProxy(wrappedExecutor)).isTrue(); + then(AopUtils.isCglibProxy(wrappedExecutor)).isFalse(); + then(wrappedExecutor.submit(() -> "done").get()).isEqualTo("done"); + } + + @Test + public void should_fallback_to_sleuth_impl_when_thread_pool_task_executor_has_final_methods() { + ExecutorBeanPostProcessor postProcessor = new ExecutorBeanPostProcessor(this.beanFactory); + ThreadPoolTaskExecutor threadPoolTaskExecutor = new PoolTaskExecutor(); + + ThreadPoolTaskExecutor wrappedTaskExecutor = (ThreadPoolTaskExecutor) postProcessor + .postProcessAfterInitialization(threadPoolTaskExecutor, "threadPoolTaskExecutor"); + + then(wrappedTaskExecutor).isInstanceOf(LazyTraceThreadPoolTaskExecutor.class); + then(AopUtils.isCglibProxy(wrappedTaskExecutor)).isFalse(); + then(AopUtils.isJdkDynamicProxy(wrappedTaskExecutor)).isFalse(); + threadPoolTaskExecutor.shutdown(); + } + @Test public void proxy_is_not_needed() throws Exception { this.sleuthAsyncProperties @@ -274,7 +332,7 @@ public class ExecutorBeanPostProcessorTests { "fooExecutor"); then(o).isInstanceOf(ThreadPoolTaskExecutor.class); - then(ClassUtils.isCglibProxy(o)).isFalse(); + then(AopUtils.isCglibProxy(o)).isFalse(); } @Test @@ -284,7 +342,7 @@ public class ExecutorBeanPostProcessorTests { "fooExecutor"); then(o).isInstanceOf(RejectedExecutionExecutor.class); - then(ClassUtils.isCglibProxy(o)).isTrue(); + then(AopUtils.isCglibProxy(o)).isTrue(); thenThrownBy(() -> ((RejectedExecutionExecutor) o).execute(() -> { })).isInstanceOf(RejectedExecutionException.class).hasMessage("rejected"); } @@ -311,4 +369,22 @@ public class ExecutorBeanPostProcessorTests { } + static class DirectTaskExecutor extends SimpleAsyncTaskExecutor { + @Override + public final Future submit(Callable callable) { + return super.submit(callable); + } + + @Override + protected void doExecute(Runnable task) { + task.run(); + } + } + + static class PoolTaskExecutor extends ThreadPoolTaskExecutor { + @Override + public final void execute(Runnable task, long startTimeout) { + super.execute(task, startTimeout); + } + } }