diff --git a/.github/workflows/maven.yml b/.github/workflows/maven.yml index 32d179893..8b5ff26b1 100644 --- a/.github/workflows/maven.yml +++ b/.github/workflows/maven.yml @@ -1,6 +1,3 @@ -# This workflow will build a Java project with Maven -# For more information see: https://help.github.com/actions/language-and-framework-guides/building-and-testing-java-with-maven - name: Build on: @@ -13,19 +10,23 @@ jobs: build: runs-on: ubuntu-latest + strategy: + matrix: + java: ["8", "11", "16"] steps: - - uses: actions/checkout@v2 - - name: Set up JDK 1.8 - uses: actions/setup-java@v1 - with: - java-version: 1.8 - - name: Cache local Maven repository - uses: actions/cache@v2 - with: - path: ~/.m2/repository - key: ${{ runner.os }}-maven-${{ hashFiles('**/pom.xml') }} - restore-keys: | - ${{ runner.os }}-maven- - - name: Build with Maven - run: ./mvnw clean install -B -U \ No newline at end of file + - uses: actions/checkout@v2 + - name: Setup java + uses: actions/setup-java@v2 + with: + distribution: 'zulu' + java-version: ${{ matrix.java }} + - name: Cache local Maven repository + uses: actions/cache@v2 + with: + path: ~/.m2/repository + key: ${{ runner.os }}-maven-${{ hashFiles('**/pom.xml') }} + restore-keys: | + ${{ runner.os }}-maven- + - name: Build with Maven + run: ./mvnw clean install -B -U diff --git a/spring-cloud-sleuth-autoconfigure/src/test/java/org/springframework/cloud/sleuth/autoconfig/brave/instrument/web/client/WebClientTests.java b/spring-cloud-sleuth-autoconfigure/src/test/java/org/springframework/cloud/sleuth/autoconfig/brave/instrument/web/client/WebClientTests.java index 665832d98..d874d7544 100644 --- a/spring-cloud-sleuth-autoconfigure/src/test/java/org/springframework/cloud/sleuth/autoconfig/brave/instrument/web/client/WebClientTests.java +++ b/spring-cloud-sleuth-autoconfigure/src/test/java/org/springframework/cloud/sleuth/autoconfig/brave/instrument/web/client/WebClientTests.java @@ -19,6 +19,7 @@ package org.springframework.cloud.sleuth.autoconfig.brave.instrument.web.client; import java.util.HashMap; import java.util.Map; import java.util.concurrent.Future; +import java.util.stream.Collectors; import brave.Span; import brave.Tracer; @@ -104,7 +105,7 @@ public class WebClientTests { then(this.tracer.currentSpan()).isNull(); then(this.spans).isNotEmpty().extracting("traceId", String.class).containsOnly(span.context().traceIdString()); - then(this.spans).extracting("kind.name").contains("CLIENT"); + then(this.spans.spans().stream().map(s -> s.kind().name()).collect(Collectors.toList())).contains("CLIENT"); } @Test @@ -140,7 +141,7 @@ public class WebClientTests { then(this.tracer.currentSpan()).isNull(); then(this.spans).isNotEmpty().extracting("traceId", String.class).containsOnly(span.context().traceIdString()); - then(this.spans).extracting("kind.name").contains("CLIENT"); + then(this.spans.spans().stream().map(s -> s.kind().name()).collect(Collectors.toList())).contains("CLIENT"); } @Configuration(proxyBeanMethods = false) diff --git a/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/async/LazyTraceScheduledThreadPoolExecutor.java b/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/async/LazyTraceScheduledThreadPoolExecutor.java index c75331036..b8d534f65 100644 --- a/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/async/LazyTraceScheduledThreadPoolExecutor.java +++ b/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/async/LazyTraceScheduledThreadPoolExecutor.java @@ -66,8 +66,6 @@ class LazyTraceScheduledThreadPoolExecutor extends ScheduledThreadPoolExecutor { private final Method decorateTaskCallable; - private final Method finalize; - private final Method beforeExecute; private final Method afterExecute; @@ -88,32 +86,66 @@ class LazyTraceScheduledThreadPoolExecutor extends ScheduledThreadPoolExecutor { this.beanFactory = beanFactory; this.delegate = delegate; this.beanName = beanName; - this.decorateTaskRunnable = ReflectionUtils.findMethod(ScheduledThreadPoolExecutor.class, "decorateTask", + Method decorateTaskRunnable = ReflectionUtils.findMethod(ScheduledThreadPoolExecutor.class, "decorateTask", Runnable.class, RunnableScheduledFuture.class); - makeAccessibleIfNotNull(this.decorateTaskRunnable); - this.decorateTaskCallable = ReflectionUtils.findMethod(ScheduledThreadPoolExecutor.class, "decorateTask", + this.decorateTaskRunnable = makeAccessibleIfNotNullAndOverridden(decorateTaskRunnable); + Method decorateTaskCallable = ReflectionUtils.findMethod(ScheduledThreadPoolExecutor.class, "decorateTask", Callable.class, RunnableScheduledFuture.class); - makeAccessibleIfNotNull(this.decorateTaskCallable); - this.finalize = ReflectionUtils.findMethod(ScheduledThreadPoolExecutor.class, "finalize", null); - makeAccessibleIfNotNull(this.finalize); - this.beforeExecute = ReflectionUtils.findMethod(ScheduledThreadPoolExecutor.class, "beforeExecute", null); - makeAccessibleIfNotNull(this.beforeExecute); - this.afterExecute = ReflectionUtils.findMethod(ScheduledThreadPoolExecutor.class, "afterExecute", null); - makeAccessibleIfNotNull(this.afterExecute); - this.terminated = ReflectionUtils.findMethod(ScheduledThreadPoolExecutor.class, "terminated", null); - makeAccessibleIfNotNull(this.terminated); - this.newTaskForRunnable = ReflectionUtils.findMethod(ScheduledThreadPoolExecutor.class, "newTaskFor", + this.decorateTaskCallable = makeAccessibleIfNotNullAndOverridden(decorateTaskCallable); + Method beforeExecute = ReflectionUtils.findMethod(ScheduledThreadPoolExecutor.class, "beforeExecute", null); + this.beforeExecute = makeAccessibleIfNotNullAndOverridden(beforeExecute); + Method afterExecute = ReflectionUtils.findMethod(ScheduledThreadPoolExecutor.class, "afterExecute", null); + this.afterExecute = makeAccessibleIfNotNullAndOverridden(afterExecute); + Method terminated = ReflectionUtils.findMethod(ScheduledThreadPoolExecutor.class, "terminated", null); + this.terminated = makeAccessibleIfNotNullAndOverridden(terminated); + Method newTaskForRunnable = ReflectionUtils.findMethod(ScheduledThreadPoolExecutor.class, "newTaskFor", Runnable.class, Object.class); - makeAccessibleIfNotNull(this.newTaskForRunnable); - this.newTaskForCallable = ReflectionUtils.findMethod(ScheduledThreadPoolExecutor.class, "newTaskFor", + this.newTaskForRunnable = makeAccessibleIfNotNullAndOverridden(newTaskForRunnable); + Method newTaskForCallable = ReflectionUtils.findMethod(ScheduledThreadPoolExecutor.class, "newTaskFor", Callable.class); - makeAccessibleIfNotNull(this.newTaskForCallable); + this.newTaskForCallable = makeAccessibleIfNotNullAndOverridden(newTaskForCallable); } - private void makeAccessibleIfNotNull(Method method) { + private Method makeAccessibleIfNotNullAndOverridden(Method method) { if (method != null) { - ReflectionUtils.makeAccessible(method); + if (isMethodOverridden(method)) { + try { + ReflectionUtils.makeAccessible(method); + return method; + } + catch (Throwable ex) { + if (anyCauseIsInaccessibleObjectException(ex)) { + throw new IllegalStateException("The executor [" + this.delegate.getClass() + + "] has overridden a method with name [" + method.getName() + + "] and the object is inaccessible. You have to run your JVM with [--add-opens] switch to allow such access. Example: [--add-opens java.base/java.util.concurrent=ALL-UNNAMED].", + ex); + } + throw ex; + } + } } + return null; + } + + private boolean anyCauseIsInaccessibleObjectException(Throwable t) { + Throwable parent = t; + Throwable cause = t.getCause(); + while (cause != null && cause != parent) { + if (cause.getClass().toString().contains("InaccessibleObjectException")) { + return true; + } + parent = cause; + cause = parent.getCause(); + } + return false; + } + + boolean isMethodOverridden(Method originalMethod) { + Method delegateMethod = ReflectionUtils.findMethod(this.delegate.getClass(), originalMethod.getName()); + if (delegateMethod == null) { + return false; + } + return !delegateMethod.equals(originalMethod); } LazyTraceScheduledThreadPoolExecutor(int corePoolSize, ThreadFactory threadFactory, @@ -123,26 +155,24 @@ class LazyTraceScheduledThreadPoolExecutor extends ScheduledThreadPoolExecutor { this.beanFactory = beanFactory; this.delegate = delegate; this.beanName = beanName; - this.decorateTaskRunnable = ReflectionUtils.findMethod(ScheduledThreadPoolExecutor.class, "decorateTask", + Method decorateTaskRunnable = ReflectionUtils.findMethod(ScheduledThreadPoolExecutor.class, "decorateTask", Runnable.class, RunnableScheduledFuture.class); - makeAccessibleIfNotNull(this.decorateTaskRunnable); - this.decorateTaskCallable = ReflectionUtils.findMethod(ScheduledThreadPoolExecutor.class, "decorateTask", + this.decorateTaskRunnable = makeAccessibleIfNotNullAndOverridden(decorateTaskRunnable); + Method decorateTaskCallable = ReflectionUtils.findMethod(ScheduledThreadPoolExecutor.class, "decorateTask", Callable.class, RunnableScheduledFuture.class); - makeAccessibleIfNotNull(this.decorateTaskCallable); - this.finalize = ReflectionUtils.findMethod(ScheduledThreadPoolExecutor.class, "finalize", null); - makeAccessibleIfNotNull(this.finalize); - this.beforeExecute = ReflectionUtils.findMethod(ScheduledThreadPoolExecutor.class, "beforeExecute", null); - makeAccessibleIfNotNull(this.beforeExecute); - this.afterExecute = ReflectionUtils.findMethod(ScheduledThreadPoolExecutor.class, "afterExecute", null); - makeAccessibleIfNotNull(this.afterExecute); - this.terminated = ReflectionUtils.findMethod(ScheduledThreadPoolExecutor.class, "terminated"); - makeAccessibleIfNotNull(this.terminated); - this.newTaskForRunnable = ReflectionUtils.findMethod(ScheduledThreadPoolExecutor.class, "newTaskFor", + this.decorateTaskCallable = makeAccessibleIfNotNullAndOverridden(decorateTaskCallable); + Method beforeExecute = ReflectionUtils.findMethod(ScheduledThreadPoolExecutor.class, "beforeExecute", null); + this.beforeExecute = makeAccessibleIfNotNullAndOverridden(beforeExecute); + Method afterExecute = ReflectionUtils.findMethod(ScheduledThreadPoolExecutor.class, "afterExecute", null); + this.afterExecute = makeAccessibleIfNotNullAndOverridden(afterExecute); + Method terminated = ReflectionUtils.findMethod(ScheduledThreadPoolExecutor.class, "terminated", null); + this.terminated = makeAccessibleIfNotNullAndOverridden(terminated); + Method newTaskForRunnable = ReflectionUtils.findMethod(ScheduledThreadPoolExecutor.class, "newTaskFor", Runnable.class, Object.class); - makeAccessibleIfNotNull(this.newTaskForRunnable); - this.newTaskForCallable = ReflectionUtils.findMethod(ScheduledThreadPoolExecutor.class, "newTaskFor", + this.newTaskForRunnable = makeAccessibleIfNotNullAndOverridden(newTaskForRunnable); + Method newTaskForCallable = ReflectionUtils.findMethod(ScheduledThreadPoolExecutor.class, "newTaskFor", Callable.class); - makeAccessibleIfNotNull(this.newTaskForCallable); + this.newTaskForCallable = makeAccessibleIfNotNullAndOverridden(newTaskForCallable); } private Runnable traceRunnableWhenContextReady(Runnable delegate) { @@ -166,6 +196,9 @@ class LazyTraceScheduledThreadPoolExecutor extends ScheduledThreadPoolExecutor { @Override @SuppressWarnings("unchecked") public RunnableScheduledFuture decorateTask(Runnable runnable, RunnableScheduledFuture task) { + if (this.decorateTaskRunnable == null) { + return super.decorateTask(traceRunnableWhenContextReady(runnable), task); + } return (RunnableScheduledFuture) ReflectionUtils.invokeMethod(this.decorateTaskRunnable, this.delegate, traceRunnableWhenContextReady(runnable), task); } @@ -173,6 +206,9 @@ class LazyTraceScheduledThreadPoolExecutor extends ScheduledThreadPoolExecutor { @Override @SuppressWarnings("unchecked") public RunnableScheduledFuture decorateTask(Callable callable, RunnableScheduledFuture task) { + if (this.decorateTaskCallable == null) { + return super.decorateTask(traceCallableWhenContextReady(callable), task); + } return (RunnableScheduledFuture) ReflectionUtils.invokeMethod(this.decorateTaskCallable, this.delegate, traceCallableWhenContextReady(callable), task); } @@ -358,7 +394,7 @@ class LazyTraceScheduledThreadPoolExecutor extends ScheduledThreadPoolExecutor { @Override public boolean remove(Runnable task) { - return this.delegate.remove(task); + return this.delegate.remove(traceRunnableWhenContextReady(task)); } @Override @@ -398,22 +434,37 @@ class LazyTraceScheduledThreadPoolExecutor extends ScheduledThreadPoolExecutor { @Override public void beforeExecute(Thread t, Runnable r) { + if (this.beforeExecute == null) { + super.beforeExecute(t, traceRunnableWhenContextReady(r)); + return; + } ReflectionUtils.invokeMethod(this.beforeExecute, this.delegate, t, traceRunnableWhenContextReady(r)); } @Override public void afterExecute(Runnable r, Throwable t) { + if (this.afterExecute == null) { + super.afterExecute(traceRunnableWhenContextReady(r), t); + return; + } ReflectionUtils.invokeMethod(this.afterExecute, this.delegate, traceRunnableWhenContextReady(r), t); } @Override public void terminated() { + if (this.terminated == null) { + super.terminated(); + return; + } ReflectionUtils.invokeMethod(this.terminated, this.delegate); } @Override @SuppressWarnings("unchecked") public RunnableFuture newTaskFor(Runnable runnable, T value) { + if (this.newTaskForRunnable == null) { + return super.newTaskFor(traceRunnableWhenContextReady(runnable), value); + } return (RunnableFuture) ReflectionUtils.invokeMethod(this.newTaskForRunnable, this.delegate, traceRunnableWhenContextReady(runnable), value); } @@ -421,6 +472,9 @@ class LazyTraceScheduledThreadPoolExecutor extends ScheduledThreadPoolExecutor { @Override @SuppressWarnings("unchecked") public RunnableFuture newTaskFor(Callable callable) { + if (this.newTaskForRunnable == null) { + return super.newTaskFor(traceCallableWhenContextReady(callable)); + } return (RunnableFuture) ReflectionUtils.invokeMethod(this.newTaskForCallable, this.delegate, traceCallableWhenContextReady(callable)); } diff --git a/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/async/LazyTraceThreadPoolTaskExecutor.java b/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/async/LazyTraceThreadPoolTaskExecutor.java index fe41d2698..0a5ad92ee 100644 --- a/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/async/LazyTraceThreadPoolTaskExecutor.java +++ b/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/async/LazyTraceThreadPoolTaskExecutor.java @@ -70,38 +70,32 @@ public class LazyTraceThreadPoolTaskExecutor extends ThreadPoolTaskExecutor { @Override public void execute(Runnable task) { - this.delegate.execute(ContextUtil.isContextUnusable(this.beanFactory) ? task - : new TraceRunnable(tracing(), spanNamer(), task, this.beanName)); + this.delegate.execute(wrap(task)); } @Override public void execute(Runnable task, long startTimeout) { - this.delegate.execute(ContextUtil.isContextUnusable(this.beanFactory) ? task - : new TraceRunnable(tracing(), spanNamer(), task, this.beanName), startTimeout); + this.delegate.execute(wrap(task), startTimeout); } @Override public Future submit(Runnable task) { - return this.delegate.submit(ContextUtil.isContextUnusable(this.beanFactory) ? task - : new TraceRunnable(tracing(), spanNamer(), task, this.beanName)); + return this.delegate.submit(wrap(task)); } @Override public Future submit(Callable task) { - return this.delegate.submit(ContextUtil.isContextUnusable(this.beanFactory) ? task - : new TraceCallable<>(tracing(), spanNamer(), task, this.beanName)); + return this.delegate.submit(wrap(task)); } @Override public ListenableFuture submitListenable(Runnable task) { - return this.delegate.submitListenable(ContextUtil.isContextUnusable(this.beanFactory) ? task - : new TraceRunnable(tracing(), spanNamer(), task, this.beanName)); + return this.delegate.submitListenable(wrap(task)); } @Override public ListenableFuture submitListenable(Callable task) { - return this.delegate.submitListenable(ContextUtil.isContextUnusable(this.beanFactory) ? task - : new TraceCallable<>(tracing(), spanNamer(), task, this.beanName)); + return this.delegate.submitListenable(wrap(task)); } @Override @@ -174,7 +168,23 @@ public class LazyTraceThreadPoolTaskExecutor extends ThreadPoolTaskExecutor { @Override public Thread newThread(Runnable runnable) { - return this.delegate.newThread(runnable); + return this.delegate.newThread(wrap(runnable)); + } + + private Runnable wrap(Runnable runnable) { + if (runnable instanceof TraceRunnable) { + return runnable; + } + return ContextUtil.isContextUnusable(this.beanFactory) ? runnable + : new TraceRunnable(tracer(), spanNamer(), runnable, this.beanName); + } + + private Callable wrap(Callable callable) { + if (callable instanceof TraceCallable) { + return callable; + } + return ContextUtil.isContextUnusable(this.beanFactory) ? callable + : new TraceCallable<>(tracer(), spanNamer(), callable, this.beanName); } @Override @@ -224,7 +234,7 @@ public class LazyTraceThreadPoolTaskExecutor extends ThreadPoolTaskExecutor { @Override public Thread createThread(Runnable runnable) { - return this.delegate.createThread(runnable); + return this.delegate.createThread(wrap(runnable)); } @Override @@ -272,7 +282,7 @@ public class LazyTraceThreadPoolTaskExecutor extends ThreadPoolTaskExecutor { this.delegate.setTaskDecorator(taskDecorator); } - private Tracer tracing() { + private Tracer tracer() { if (this.tracer == null) { this.tracer = this.beanFactory.getBean(Tracer.class); } diff --git a/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/async/LazyTraceThreadPoolTaskScheduler.java b/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/async/LazyTraceThreadPoolTaskScheduler.java index e82266921..136ef830a 100644 --- a/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/async/LazyTraceThreadPoolTaskScheduler.java +++ b/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/async/LazyTraceThreadPoolTaskScheduler.java @@ -28,6 +28,7 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.ThreadFactory; +import java.util.concurrent.ThreadPoolExecutor; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -106,6 +107,9 @@ class LazyTraceThreadPoolTaskScheduler extends ThreadPoolTaskScheduler { if (ContextUtil.isContextUnusable(this.beanFactory)) { return delegate; } + if (delegate instanceof TraceRunnable) { + return delegate; + } return new TraceRunnable(tracing(), spanNamer(), delegate, this.beanName); } @@ -113,6 +117,9 @@ class LazyTraceThreadPoolTaskScheduler extends ThreadPoolTaskScheduler { if (ContextUtil.isContextUnusable(this.beanFactory)) { return delegate; } + if (delegate instanceof TraceCallable) { + return delegate; + } return new TraceCallable<>(tracing(), spanNamer(), delegate, this.beanName); } @@ -135,15 +142,25 @@ class LazyTraceThreadPoolTaskScheduler extends ThreadPoolTaskScheduler { public ExecutorService initializeExecutor(ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) { ExecutorService executorService = (ExecutorService) ReflectionUtils.invokeMethod(this.initializeExecutor, - this.delegate, traceThreadFactory(threadFactory), rejectedExecutionHandler); + this.delegate, traceThreadFactory(threadFactory), + traceRejectedExecutionHandler(rejectedExecutionHandler)); if (executorService instanceof TraceableScheduledExecutorService) { return executorService; } return new TraceableExecutorService(this.beanFactory, executorService, this.beanName); } + private RejectedExecutionHandler traceRejectedExecutionHandler(RejectedExecutionHandler rejectedExecutionHandler) { + return new RejectedExecutionHandler() { + @Override + public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { + rejectedExecutionHandler.rejectedExecution(traceRunnableWhenContextReady(r), executor); + } + }; + } + private ThreadFactory traceThreadFactory(ThreadFactory threadFactory) { - return r -> threadFactory.newThread(new TraceRunnable(tracing(), spanNamer(), r, this.beanName)); + return r -> threadFactory.newThread(traceRunnableWhenContextReady(r)); } @Override @@ -151,7 +168,7 @@ class LazyTraceThreadPoolTaskScheduler extends ThreadPoolTaskScheduler { RejectedExecutionHandler rejectedExecutionHandler) { ScheduledExecutorService executorService = (ScheduledExecutorService) ReflectionUtils.invokeMethod( this.createExecutor, this.delegate, poolSize, traceThreadFactory(threadFactory), - rejectedExecutionHandler); + traceRejectedExecutionHandler(rejectedExecutionHandler)); if (executorService instanceof TraceableScheduledExecutorService) { return executorService; } @@ -313,7 +330,7 @@ class LazyTraceThreadPoolTaskScheduler extends ThreadPoolTaskScheduler { @Override public Thread newThread(Runnable runnable) { - return this.delegate.newThread(runnable); + return this.delegate.newThread(traceRunnableWhenContextReady(runnable)); } @Override @@ -359,7 +376,7 @@ class LazyTraceThreadPoolTaskScheduler extends ThreadPoolTaskScheduler { @Override public Thread createThread(Runnable runnable) { - return this.delegate.createThread(runnable); + return this.delegate.createThread(traceRunnableWhenContextReady(runnable)); } @Override diff --git a/spring-cloud-sleuth-instrumentation/src/test/java/org/springframework/cloud/sleuth/instrument/async/ExecutorInstrumentorTests.java b/spring-cloud-sleuth-instrumentation/src/test/java/org/springframework/cloud/sleuth/instrument/async/ExecutorInstrumentorTests.java index cfc61bd3b..dfed086bb 100644 --- a/spring-cloud-sleuth-instrumentation/src/test/java/org/springframework/cloud/sleuth/instrument/async/ExecutorInstrumentorTests.java +++ b/spring-cloud-sleuth-instrumentation/src/test/java/org/springframework/cloud/sleuth/instrument/async/ExecutorInstrumentorTests.java @@ -36,6 +36,8 @@ import org.aopalliance.aop.Advice; import org.assertj.core.api.BDDAssertions; import org.awaitility.Awaitility; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.EnabledForJreRange; +import org.junit.jupiter.api.condition.JRE; import org.junit.jupiter.api.extension.ExtendWith; import org.mockito.BDDMockito; import org.mockito.Mock; @@ -294,6 +296,7 @@ public class ExecutorInstrumentorTests { } @Test + @EnabledForJreRange(min = JRE.JAVA_8, max = JRE.JAVA_15) public void should_use_cglib_proxy_when_an_executor_has_a_final_package_protected_method() { ExecutorInstrumentor beanPostProcessor = new ExecutorInstrumentor(Collections::emptyList, beanFactory); ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(10); @@ -307,6 +310,21 @@ public class ExecutorInstrumentorTests { Awaitility.await().untilAsserted(() -> BDDAssertions.then(wasCalled).isTrue()); } + @Test + @EnabledForJreRange(min = JRE.JAVA_16) + public void should_use_jdk_proxy_when_an_executor_has_a_final_package_protected_method() { + ExecutorInstrumentor beanPostProcessor = new ExecutorInstrumentor(Collections::emptyList, beanFactory); + ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(10); + ScheduledThreadPoolExecutor wrappedExecutor = (ScheduledThreadPoolExecutor) beanPostProcessor + .instrument(executor, "executor"); + + BDDAssertions.then(AopUtils.isCglibProxy(wrappedExecutor)).isFalse(); + + AtomicBoolean wasCalled = new AtomicBoolean(false); + wrappedExecutor.execute(() -> wasCalled.set(true)); + Awaitility.await().untilAsserted(() -> BDDAssertions.then(wasCalled).isTrue()); + } + @Test public void should_use_jdk_proxy_when_executor_service_has_final_methods() throws Exception { ExecutorInstrumentor beanPostProcessor = new ExecutorInstrumentor(Collections::emptyList, beanFactory); diff --git a/spring-cloud-sleuth-instrumentation/src/test/java/org/springframework/cloud/sleuth/instrument/async/LazyTraceScheduledThreadPoolExecutorTests.java b/spring-cloud-sleuth-instrumentation/src/test/java/org/springframework/cloud/sleuth/instrument/async/LazyTraceScheduledThreadPoolExecutorTests.java index 0ace63e7f..8dc98bc85 100644 --- a/spring-cloud-sleuth-instrumentation/src/test/java/org/springframework/cloud/sleuth/instrument/async/LazyTraceScheduledThreadPoolExecutorTests.java +++ b/spring-cloud-sleuth-instrumentation/src/test/java/org/springframework/cloud/sleuth/instrument/async/LazyTraceScheduledThreadPoolExecutorTests.java @@ -16,6 +16,7 @@ package org.springframework.cloud.sleuth.instrument.async; +import java.lang.reflect.Method; import java.util.Arrays; import java.util.Collection; import java.util.List; @@ -34,8 +35,11 @@ import java.util.concurrent.atomic.AtomicBoolean; import org.assertj.core.api.BDDAssertions; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.EnabledForJreRange; +import org.junit.jupiter.api.condition.JRE; import org.junit.jupiter.api.extension.ExtendWith; import org.mockito.ArgumentCaptor; +import org.mockito.BDDMockito; import org.mockito.Captor; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; @@ -92,6 +96,14 @@ public class LazyTraceScheduledThreadPoolExecutorTests { boolean isContextUnusable() { return false; } + + @Override + boolean isMethodOverridden(Method originalMethod) { + if (JRE.currentVersion().ordinal() >= JRE.JAVA_16.ordinal()) { + return false; + } + return true; + } }); } @@ -107,13 +119,22 @@ public class LazyTraceScheduledThreadPoolExecutorTests { }; BeanFactory beanFactory = mock(BeanFactory.class); - new LazyTraceScheduledThreadPoolExecutor(10, beanFactory, executor, null).finalize(); + new LazyTraceScheduledThreadPoolExecutor(10, beanFactory, executor, null) { + @Override + boolean isMethodOverridden(Method originalMethod) { + if (JRE.currentVersion().ordinal() >= JRE.JAVA_16.ordinal()) { + return false; + } + return true; + } + }.finalize(); BDDAssertions.then(wasCalled).isFalse(); BDDAssertions.then(executor.isShutdown()).isFalse(); } @Test + @EnabledForJreRange(min = JRE.JAVA_8, max = JRE.JAVA_15) public void should_delegate_decorateTask_with_runnable() { final Runnable runnable = mock(Runnable.class); final RunnableScheduledFuture value = mock(RunnableScheduledFuture.class); @@ -128,6 +149,7 @@ public class LazyTraceScheduledThreadPoolExecutorTests { } @Test + @EnabledForJreRange(min = JRE.JAVA_8, max = JRE.JAVA_15) public void should_delegate_decorateTask_with_callable() { final Callable callable = mock(Callable.class); final RunnableScheduledFuture value = mock(RunnableScheduledFuture.class); @@ -511,7 +533,7 @@ public class LazyTraceScheduledThreadPoolExecutorTests { executor.remove(expected); - verify(delegate).remove(expected); + verify(delegate).remove(BDDMockito.isA(TraceRunnable.class)); } @Test @@ -582,6 +604,7 @@ public class LazyTraceScheduledThreadPoolExecutorTests { } @Test + @EnabledForJreRange(min = JRE.JAVA_8, max = JRE.JAVA_15) public void should_delegate_beforeExecute() { final Thread thread = mock(Thread.class); final Runnable expected = mock(Runnable.class); @@ -593,6 +616,7 @@ public class LazyTraceScheduledThreadPoolExecutorTests { } @Test + @EnabledForJreRange(min = JRE.JAVA_8, max = JRE.JAVA_15) public void should_delegate_afterExecute() { final Throwable throwable = mock(Throwable.class); final Runnable expected = mock(Runnable.class); @@ -604,6 +628,7 @@ public class LazyTraceScheduledThreadPoolExecutorTests { } @Test + @EnabledForJreRange(min = JRE.JAVA_8, max = JRE.JAVA_15) public void should_delegate_terminated() { executor.terminated(); @@ -611,6 +636,7 @@ public class LazyTraceScheduledThreadPoolExecutorTests { } @Test + @EnabledForJreRange(min = JRE.JAVA_8, max = JRE.JAVA_15) public void should_delegate_newTaskForRunnable() { final Runnable runnable = mock(Runnable.class); final String expected = "testing"; @@ -625,6 +651,7 @@ public class LazyTraceScheduledThreadPoolExecutorTests { } @Test + @EnabledForJreRange(min = JRE.JAVA_8, max = JRE.JAVA_15) public void should_delegate_newTaskForCallable() { final Callable callable = mock(Callable.class); final RunnableFuture expected = mock(RunnableFuture.class); diff --git a/tests/brave/spring-cloud-sleuth-instrumentation-async-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/async/TraceAsyncIntegrationTests.java b/tests/brave/spring-cloud-sleuth-instrumentation-async-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/async/TraceAsyncIntegrationTests.java index 22781b012..0f5afefd6 100644 --- a/tests/brave/spring-cloud-sleuth-instrumentation-async-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/async/TraceAsyncIntegrationTests.java +++ b/tests/brave/spring-cloud-sleuth-instrumentation-async-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/async/TraceAsyncIntegrationTests.java @@ -78,9 +78,8 @@ public class TraceAsyncIntegrationTests { assertThat(span.traceId()).isEqualTo(context.traceIdString()); } finally { - parent.finish(); + parent.abandon(); } - } @Test @@ -100,7 +99,7 @@ public class TraceAsyncIntegrationTests { assertThat(span.traceId()).isEqualTo(context.traceIdString()); } finally { - parent.finish(); + parent.abandon(); } } @@ -108,10 +107,8 @@ public class TraceAsyncIntegrationTests { // We don't want that one. MutableSpan takeDesirableSpan(String name) { MutableSpan span1 = spans.takeLocalSpan(); - MutableSpan span2 = spans.takeLocalSpan(); - log.info("Two last spans [" + span2 + "] and [" + span1 + "]"); - MutableSpan span = span1 != null && name.equals(span1.name()) ? span1 - : span2 != null && name.equals(span2.name()) ? span2 : null; + log.info("Span [" + span1 + "] found"); + MutableSpan span = span1 != null && name.equals(span1.name()) ? span1 : null; assertThat(span).as("No span with name <> was found", name).isNotNull(); return span; } diff --git a/tests/brave/spring-cloud-sleuth-instrumentation-async-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/async/TraceScheduledThreadPoolExecutorAnotherConstructorTests.java b/tests/brave/spring-cloud-sleuth-instrumentation-async-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/async/TraceScheduledThreadPoolExecutorAnotherConstructorTests.java new file mode 100644 index 000000000..6fe688f58 --- /dev/null +++ b/tests/brave/spring-cloud-sleuth-instrumentation-async-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/async/TraceScheduledThreadPoolExecutorAnotherConstructorTests.java @@ -0,0 +1,35 @@ +/* + * Copyright 2013-2021 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.sleuth.brave.instrument.async; + +import org.springframework.cloud.sleuth.brave.BraveTestTracing; +import org.springframework.cloud.sleuth.test.TestTracingAware; + +public class TraceScheduledThreadPoolExecutorAnotherConstructorTests extends + org.springframework.cloud.sleuth.instrument.async.TraceScheduledThreadPoolExecutorAnotherConstructorTests { + + BraveTestTracing testTracing; + + @Override + public TestTracingAware tracerTest() { + if (this.testTracing == null) { + this.testTracing = new BraveTestTracing(); + } + return this.testTracing; + } + +} diff --git a/tests/brave/spring-cloud-sleuth-instrumentation-async-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/async/TraceScheduledThreadPoolExecutorTests.java b/tests/brave/spring-cloud-sleuth-instrumentation-async-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/async/TraceScheduledThreadPoolExecutorTests.java new file mode 100644 index 000000000..397e963fe --- /dev/null +++ b/tests/brave/spring-cloud-sleuth-instrumentation-async-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/async/TraceScheduledThreadPoolExecutorTests.java @@ -0,0 +1,35 @@ +/* + * Copyright 2013-2021 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.sleuth.brave.instrument.async; + +import org.springframework.cloud.sleuth.brave.BraveTestTracing; +import org.springframework.cloud.sleuth.test.TestTracingAware; + +public class TraceScheduledThreadPoolExecutorTests + extends org.springframework.cloud.sleuth.instrument.async.TraceScheduledThreadPoolExecutorTests { + + BraveTestTracing testTracing; + + @Override + public TestTracingAware tracerTest() { + if (this.testTracing == null) { + this.testTracing = new BraveTestTracing(); + } + return this.testTracing; + } + +} diff --git a/tests/brave/spring-cloud-sleuth-instrumentation-async-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/async/TraceThreadPoolTaskExecutorTests.java b/tests/brave/spring-cloud-sleuth-instrumentation-async-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/async/TraceThreadPoolTaskExecutorTests.java new file mode 100644 index 000000000..2f2df4800 --- /dev/null +++ b/tests/brave/spring-cloud-sleuth-instrumentation-async-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/async/TraceThreadPoolTaskExecutorTests.java @@ -0,0 +1,35 @@ +/* + * Copyright 2013-2021 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.sleuth.brave.instrument.async; + +import org.springframework.cloud.sleuth.brave.BraveTestTracing; +import org.springframework.cloud.sleuth.test.TestTracingAware; + +public class TraceThreadPoolTaskExecutorTests + extends org.springframework.cloud.sleuth.instrument.async.TraceThreadPoolTaskExecutorTests { + + BraveTestTracing testTracing; + + @Override + public TestTracingAware tracerTest() { + if (this.testTracing == null) { + this.testTracing = new BraveTestTracing(); + } + return this.testTracing; + } + +} diff --git a/tests/brave/spring-cloud-sleuth-instrumentation-async-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/async/TraceThreadPoolTaskSchedulerTests.java b/tests/brave/spring-cloud-sleuth-instrumentation-async-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/async/TraceThreadPoolTaskSchedulerTests.java new file mode 100644 index 000000000..4d990fdfa --- /dev/null +++ b/tests/brave/spring-cloud-sleuth-instrumentation-async-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/async/TraceThreadPoolTaskSchedulerTests.java @@ -0,0 +1,35 @@ +/* + * Copyright 2013-2021 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.sleuth.brave.instrument.async; + +import org.springframework.cloud.sleuth.brave.BraveTestTracing; +import org.springframework.cloud.sleuth.test.TestTracingAware; + +public class TraceThreadPoolTaskSchedulerTests + extends org.springframework.cloud.sleuth.instrument.async.TraceThreadPoolTaskSchedulerTests { + + BraveTestTracing testTracing; + + @Override + public TestTracingAware tracerTest() { + if (this.testTracing == null) { + this.testTracing = new BraveTestTracing(); + } + return this.testTracing; + } + +} diff --git a/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/async/LazyTraceThreadPoolTaskSchedulerTests.java b/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/async/LazyTraceThreadPoolTaskSchedulerTests.java index c75511d6b..3aa01c258 100644 --- a/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/async/LazyTraceThreadPoolTaskSchedulerTests.java +++ b/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/async/LazyTraceThreadPoolTaskSchedulerTests.java @@ -343,7 +343,7 @@ public abstract class LazyTraceThreadPoolTaskSchedulerTests implements TestTraci }; this.executor.newThread(runnable); - BDDMockito.then(this.delegate).should().newThread(runnable); + BDDMockito.then(this.delegate).should().newThread(BDDMockito.isA(TraceRunnable.class)); } @Test @@ -409,7 +409,7 @@ public abstract class LazyTraceThreadPoolTaskSchedulerTests implements TestTraci }; this.executor.createThread(r); - BDDMockito.then(this.delegate).should().createThread(r); + BDDMockito.then(this.delegate).should().createThread(BDDMockito.isA(TraceRunnable.class)); } @Test diff --git a/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/async/TraceScheduledThreadPoolExecutorAnotherConstructorTests.java b/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/async/TraceScheduledThreadPoolExecutorAnotherConstructorTests.java new file mode 100644 index 000000000..bd659a8f9 --- /dev/null +++ b/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/async/TraceScheduledThreadPoolExecutorAnotherConstructorTests.java @@ -0,0 +1,44 @@ +/* + * Copyright 2013-2021 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.sleuth.instrument.async; + +import java.util.concurrent.RejectedExecutionHandler; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.ThreadPoolExecutor; + +/** + * @author Marcin Grzejszczak + */ +public abstract class TraceScheduledThreadPoolExecutorAnotherConstructorTests + extends TraceScheduledThreadPoolExecutorTests { + + @Override + protected LazyTraceScheduledThreadPoolExecutor executor() { + return new LazyTraceScheduledThreadPoolExecutor(1, new ThreadFactory() { + @Override + public Thread newThread(Runnable r) { + return new Thread(r); + } + }, new RejectedExecutionHandler() { + @Override + public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { + + } + }, beanFactory, delegate, "foo"); + } + +} diff --git a/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/async/TraceScheduledThreadPoolExecutorTests.java b/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/async/TraceScheduledThreadPoolExecutorTests.java new file mode 100644 index 000000000..989ed5a90 --- /dev/null +++ b/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/async/TraceScheduledThreadPoolExecutorTests.java @@ -0,0 +1,355 @@ +/* + * Copyright 2013-2021 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.sleuth.instrument.async; + +import java.util.concurrent.Callable; +import java.util.concurrent.Delayed; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.RunnableScheduledFuture; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.awaitility.Awaitility; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.BDDMockito; + +import org.springframework.beans.factory.BeanFactory; +import org.springframework.cloud.sleuth.Span; +import org.springframework.cloud.sleuth.SpanNamer; +import org.springframework.cloud.sleuth.Tracer; +import org.springframework.cloud.sleuth.internal.DefaultSpanNamer; +import org.springframework.cloud.sleuth.internal.SleuthContextListenerAccessor; +import org.springframework.cloud.sleuth.test.TestTracingAwareSupplier; + +import static org.assertj.core.api.BDDAssertions.then; + +/** + * @author Marcin Grzejszczak + */ +public abstract class TraceScheduledThreadPoolExecutorTests implements TestTracingAwareSupplier { + + ScheduledThreadPoolExecutor delegate = new ScheduledThreadPoolExecutor(1); + + BeanFactory beanFactory = beanFactory(); + + LazyTraceScheduledThreadPoolExecutor traceThreadPoolTaskExecutor = executor(); + + protected LazyTraceScheduledThreadPoolExecutor executor() { + return new LazyTraceScheduledThreadPoolExecutor(1, this.beanFactory, this.delegate, "foo"); + } + + @BeforeEach + void setup() { + SleuthContextListenerAccessor.set(this.beanFactory, true); + } + + @AfterEach + void clear() { + this.delegate.shutdown(); + } + + private BeanFactory beanFactory() { + BeanFactory beanFactory = BDDMockito.mock(BeanFactory.class); + BDDMockito.given(beanFactory.getBean(Tracer.class)).willReturn(tracerTest().tracing().tracer()); + BDDMockito.given(beanFactory.getBean(SpanNamer.class)).willReturn(new DefaultSpanNamer()); + return beanFactory; + } + + @Test + public void should_schedule_trace_runnable() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + this.traceThreadPoolTaskExecutor.schedule(aRunnable(executed, span), 1, TimeUnit.MILLISECONDS).get(1, + TimeUnit.SECONDS); + } + finally { + span.end(); + } + + then(executed.get()).isTrue(); + } + + @Test + public void should_decorate_task_trace_runnable() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + Runnable runnable = aRunnable(executed, span); + this.traceThreadPoolTaskExecutor.decorateTask(runnable, runnableScheduledFuture(runnable)).run(); + } + finally { + span.end(); + } + + Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> { + then(executed.get()).isTrue(); + }); + } + + @Test + public void should_decorate_task_trace_callable() throws Exception { + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + RunnableScheduledFuture fromCallable; + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + Callable callable = aCallable(span); + fromCallable = this.traceThreadPoolTaskExecutor.decorateTask(callable, runnableScheduledFuture(callable)); + fromCallable.run(); + } + finally { + span.end(); + } + + Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> { + then(fromCallable.get(10, TimeUnit.MILLISECONDS)).isNotNull(); + }); + } + + private RunnableScheduledFuture runnableScheduledFuture(Callable run) { + return new RunnableScheduledFuture() { + + private Span result; + + @Override + public boolean isPeriodic() { + return false; + } + + @Override + public long getDelay(TimeUnit unit) { + return 0; + } + + @Override + public int compareTo(Delayed o) { + return 0; + } + + @Override + public void run() { + try { + this.result = run.call(); + } + catch (Exception exception) { + } + } + + @Override + public boolean cancel(boolean mayInterruptIfRunning) { + return false; + } + + @Override + public boolean isCancelled() { + return false; + } + + @Override + public boolean isDone() { + return false; + } + + @Override + public Span get() throws InterruptedException, ExecutionException { + return this.result; + } + + @Override + public Span get(long timeout, TimeUnit unit) + throws InterruptedException, ExecutionException, TimeoutException { + return this.result; + } + }; + } + + private RunnableScheduledFuture runnableScheduledFuture(Runnable run) { + return new RunnableScheduledFuture() { + + @Override + public boolean isPeriodic() { + return false; + } + + @Override + public long getDelay(TimeUnit unit) { + return 0; + } + + @Override + public int compareTo(Delayed o) { + return 0; + } + + @Override + public void run() { + run.run(); + } + + @Override + public boolean cancel(boolean mayInterruptIfRunning) { + return false; + } + + @Override + public boolean isCancelled() { + return false; + } + + @Override + public boolean isDone() { + return false; + } + + @Override + public Object get() throws InterruptedException, ExecutionException { + return null; + } + + @Override + public Object get(long timeout, TimeUnit unit) + throws InterruptedException, ExecutionException, TimeoutException { + return null; + } + }; + } + + @Test + public void should_schedule_trace_callable() throws Exception { + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + Span spanFromCallable; + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + spanFromCallable = this.traceThreadPoolTaskExecutor.schedule(aCallable(span), 1, TimeUnit.MILLISECONDS) + .get(1, TimeUnit.SECONDS); + } + finally { + span.end(); + } + + then(spanFromCallable).isNotNull(); + } + + @Test + public void should_schedule_at_fixed_rate_trace_runnable() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + this.traceThreadPoolTaskExecutor.scheduleAtFixedRate(aRunnable(executed, span), 1L, 1L, + TimeUnit.MILLISECONDS); + } + finally { + span.end(); + } + + Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> { + then(executed.get()).isTrue(); + }); + } + + @Test + public void should_schedule_with_fixed_delay_trace_runnable() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + this.traceThreadPoolTaskExecutor.scheduleWithFixedDelay(aRunnable(executed, span), 1L, 1L, + TimeUnit.MILLISECONDS); + } + finally { + span.end(); + } + + Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> { + then(executed.get()).isTrue(); + }); + } + + @Test + public void should_execute_a_trace_runnable() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + this.traceThreadPoolTaskExecutor.execute(aRunnable(executed, span)); + } + finally { + span.end(); + } + + Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> { + then(executed.get()).isTrue(); + }); + } + + @Test + public void should_submit_trace_callable() throws Exception { + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + Span spanFromListenable; + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + spanFromListenable = this.traceThreadPoolTaskExecutor.submit(aCallable(span)).get(1, TimeUnit.SECONDS); + } + finally { + span.end(); + } + + then(spanFromListenable).isNotNull(); + } + + @Test + public void should_submit_trace_runnable() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + this.traceThreadPoolTaskExecutor.submit(aRunnable(executed, span)).get(1, TimeUnit.SECONDS); + } + finally { + span.end(); + } + + Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> { + then(executed.get()).isTrue(); + }); + } + + Runnable aRunnable(AtomicBoolean executed, Span currentSpan) { + return () -> { + Span span = tracerTest().tracing().tracer().currentSpan(); + then(span).isNotNull(); + then(span.context().traceId()).isEqualTo(currentSpan.context().traceId()); + executed.set(true); + }; + } + + Callable aCallable(Span currentSpan) { + return () -> { + Span span = tracerTest().tracing().tracer().currentSpan(); + then(span.context().traceId()).isEqualTo(currentSpan.context().traceId()); + return span; + }; + } + +} diff --git a/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/async/TraceThreadPoolTaskExecutorTests.java b/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/async/TraceThreadPoolTaskExecutorTests.java new file mode 100644 index 000000000..45539d12b --- /dev/null +++ b/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/async/TraceThreadPoolTaskExecutorTests.java @@ -0,0 +1,234 @@ +/* + * Copyright 2013-2021 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.sleuth.instrument.async; + +import java.util.concurrent.Callable; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.assertj.core.api.BDDAssertions; +import org.awaitility.Awaitility; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.BDDMockito; + +import org.springframework.beans.factory.BeanFactory; +import org.springframework.cloud.sleuth.Span; +import org.springframework.cloud.sleuth.SpanNamer; +import org.springframework.cloud.sleuth.Tracer; +import org.springframework.cloud.sleuth.internal.DefaultSpanNamer; +import org.springframework.cloud.sleuth.internal.SleuthContextListenerAccessor; +import org.springframework.cloud.sleuth.test.TestTracingAwareSupplier; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; + +/** + * @author Marcin Grzejszczak + */ +public abstract class TraceThreadPoolTaskExecutorTests implements TestTracingAwareSupplier { + + ThreadPoolTaskExecutor delegate = new ThreadPoolTaskExecutor(); + + BeanFactory beanFactory = beanFactory(); + + LazyTraceThreadPoolTaskExecutor traceThreadPoolTaskExecutor = new LazyTraceThreadPoolTaskExecutor(this.beanFactory, + this.delegate); + + @BeforeEach + void setup() { + this.delegate.initialize(); + SleuthContextListenerAccessor.set(this.beanFactory, true); + } + + @AfterEach + void clear() { + this.delegate.shutdown(); + } + + private BeanFactory beanFactory() { + BeanFactory beanFactory = BDDMockito.mock(BeanFactory.class); + BDDMockito.given(beanFactory.getBean(Tracer.class)).willReturn(tracerTest().tracing().tracer()); + BDDMockito.given(beanFactory.getBean(SpanNamer.class)).willReturn(new DefaultSpanNamer()); + return beanFactory; + } + + @Test + public void should_create_thread_trace_runnable() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + this.traceThreadPoolTaskExecutor.createThread(aRunnable(executed, span)).start(); + } + finally { + span.end(); + } + + Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> { + BDDAssertions.then(executed.get()).isTrue(); + }); + } + + @Test + public void should_submit_listenable_trace_runnable() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + this.traceThreadPoolTaskExecutor.submitListenable(aRunnable(executed, span)).get(); + } + finally { + span.end(); + } + + BDDAssertions.then(executed.get()).isTrue(); + } + + @Test + public void should_submit_listenable_trace_callable() throws Exception { + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + Span spanFromListenable; + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + spanFromListenable = this.traceThreadPoolTaskExecutor.submitListenable(aCallable(span)).get(1, + TimeUnit.SECONDS); + } + finally { + span.end(); + } + + BDDAssertions.then(spanFromListenable).isNotNull(); + } + + @Test + public void should_execute_a_trace_runnable() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + this.traceThreadPoolTaskExecutor.execute(aRunnable(executed, span)); + } + finally { + span.end(); + } + + Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> { + BDDAssertions.then(executed.get()).isTrue(); + }); + } + + @Test + public void should_execute_with_timeout_a_trace_runnable() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + this.traceThreadPoolTaskExecutor.execute(aRunnable(executed, span), 1L); + } + finally { + span.end(); + } + + Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> { + BDDAssertions.then(executed.get()).isTrue(); + }); + } + + @Test + public void should_submit_trace_callable() throws Exception { + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + Span spanFromListenable; + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + spanFromListenable = this.traceThreadPoolTaskExecutor.submit(aCallable(span)).get(1, TimeUnit.SECONDS); + } + finally { + span.end(); + } + + BDDAssertions.then(spanFromListenable).isNotNull(); + } + + @Test + public void should_submit_trace_runnable() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + this.traceThreadPoolTaskExecutor.submit(aRunnable(executed, span)).get(1, TimeUnit.SECONDS); + } + finally { + span.end(); + } + + Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> { + BDDAssertions.then(executed.get()).isTrue(); + }); + } + + @Test + public void should_submit_trace_runnable_via_new_thread() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + this.traceThreadPoolTaskExecutor.newThread(aRunnable(executed, span)).start(); + } + finally { + span.end(); + } + + Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> { + BDDAssertions.then(executed.get()).isTrue(); + }); + } + + @Test + public void should_submit_trace_runnable_via_create_thread() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + this.traceThreadPoolTaskExecutor.createThread(aRunnable(executed, span)).start(); + } + finally { + span.end(); + } + + Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> { + BDDAssertions.then(executed.get()).isTrue(); + }); + } + + Runnable aRunnable(AtomicBoolean executed, Span currentSpan) { + return () -> { + Span span = tracerTest().tracing().tracer().currentSpan(); + BDDAssertions.then(span).isNotNull(); + BDDAssertions.then(span.context().traceId()).isEqualTo(currentSpan.context().traceId()); + executed.set(true); + }; + } + + Callable aCallable(Span currentSpan) { + return () -> { + Span span = tracerTest().tracing().tracer().currentSpan(); + BDDAssertions.then(span.context().traceId()).isEqualTo(currentSpan.context().traceId()); + return span; + }; + } + +} diff --git a/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/async/TraceThreadPoolTaskSchedulerTests.java b/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/async/TraceThreadPoolTaskSchedulerTests.java new file mode 100644 index 000000000..5c2716f76 --- /dev/null +++ b/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/async/TraceThreadPoolTaskSchedulerTests.java @@ -0,0 +1,527 @@ +/* + * Copyright 2013-2021 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.sleuth.instrument.async; + +import java.sql.Date; +import java.time.Duration; +import java.time.Instant; +import java.util.concurrent.Callable; +import java.util.concurrent.RejectedExecutionHandler; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.awaitility.Awaitility; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.BDDMockito; + +import org.springframework.beans.factory.BeanFactory; +import org.springframework.cloud.sleuth.Span; +import org.springframework.cloud.sleuth.SpanNamer; +import org.springframework.cloud.sleuth.Tracer; +import org.springframework.cloud.sleuth.internal.DefaultSpanNamer; +import org.springframework.cloud.sleuth.internal.SleuthContextListenerAccessor; +import org.springframework.cloud.sleuth.test.TestTracingAwareSupplier; +import org.springframework.scheduling.Trigger; +import org.springframework.scheduling.TriggerContext; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; + +import static org.assertj.core.api.BDDAssertions.then; + +/** + * @author Marcin Grzejszczak + */ +public abstract class TraceThreadPoolTaskSchedulerTests implements TestTracingAwareSupplier { + + ThreadPoolTaskScheduler delegate = new ThreadPoolTaskScheduler(); + + BeanFactory beanFactory = beanFactory(); + + LazyTraceThreadPoolTaskScheduler traceThreadPoolTaskExecutor = new LazyTraceThreadPoolTaskScheduler( + this.beanFactory, this.delegate, "foo"); + + @BeforeEach + void setup() { + this.delegate.initialize(); + SleuthContextListenerAccessor.set(this.beanFactory, true); + } + + @AfterEach + void clear() { + this.delegate.shutdown(); + } + + private BeanFactory beanFactory() { + BeanFactory beanFactory = BDDMockito.mock(BeanFactory.class); + BDDMockito.given(beanFactory.getBean(Tracer.class)).willReturn(tracerTest().tracing().tracer()); + BDDMockito.given(beanFactory.getBean(SpanNamer.class)).willReturn(new DefaultSpanNamer()); + return beanFactory; + } + + @Test + public void should_initialize_wrapped_executor() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + this.traceThreadPoolTaskExecutor.initializeExecutor(new ThreadFactory() { + @Override + public Thread newThread(Runnable r) { + return new Thread(r); + } + }, new RejectedExecutionHandler() { + @Override + public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { + + } + }).submit(aRunnable(executed, span)).get(1, TimeUnit.SECONDS); + } + finally { + span.end(); + } + + Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> { + then(executed.get()).isTrue(); + }); + } + + @Test + public void should_create_wrapped_executor() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + this.traceThreadPoolTaskExecutor.createExecutor(1, new ThreadFactory() { + @Override + public Thread newThread(Runnable r) { + return new Thread(r); + } + }, new RejectedExecutionHandler() { + @Override + public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { + + } + }).submit(aRunnable(executed, span)).get(1, TimeUnit.SECONDS); + } + finally { + span.end(); + } + + Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> { + then(executed.get()).isTrue(); + }); + } + + @Test + public void should_get_scheduled_wrapped_executor() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + this.traceThreadPoolTaskExecutor.getScheduledExecutor().submit(aRunnable(executed, span)).get(1, + TimeUnit.SECONDS); + } + finally { + span.end(); + } + + Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> { + then(executed.get()).isTrue(); + }); + } + + @Test + public void should_get_scheduled_thread_pool_wrapped_executor() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + this.traceThreadPoolTaskExecutor.getScheduledThreadPoolExecutor().submit(aRunnable(executed, span)).get(1, + TimeUnit.SECONDS); + } + finally { + span.end(); + } + + Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> { + then(executed.get()).isTrue(); + }); + } + + @Test + public void should_create_thread_trace_runnable() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + this.traceThreadPoolTaskExecutor.createThread(aRunnable(executed, span)).start(); + } + finally { + span.end(); + } + + Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> { + then(executed.get()).isTrue(); + }); + } + + @Test + public void should_schedule_trace_runnable() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + this.traceThreadPoolTaskExecutor.schedule(aRunnable(executed, span), Instant.now()).get(1, + TimeUnit.SECONDS); + } + finally { + span.end(); + } + + then(executed.get()).isTrue(); + } + + @Test + public void should_schedule_trace_runnable_with_start_time() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + this.traceThreadPoolTaskExecutor.schedule(aRunnable(executed, span), Date.from(Instant.now())).get(1, + TimeUnit.SECONDS); + } + finally { + span.end(); + } + + then(executed.get()).isTrue(); + } + + @Test + public void should_schedule_trace_runnable_with_trigger() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + this.traceThreadPoolTaskExecutor.schedule(aRunnable(executed, span), new Trigger() { + @Override + public java.util.Date nextExecutionTime(TriggerContext triggerContext) { + return java.util.Date.from(Instant.now()); + } + }).get(1, TimeUnit.SECONDS); + } + finally { + span.end(); + } + + then(executed.get()).isTrue(); + } + + @Test + public void should_schedule_at_fixed_rate_trace_runnable_with_date() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + this.traceThreadPoolTaskExecutor.scheduleAtFixedRate(aRunnable(executed, span), Date.from(Instant.now()), + 1L); + } + finally { + span.end(); + } + + Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> { + then(executed.get()).isTrue(); + }); + } + + @Test + public void should_schedule_at_fixed_rate_trace_runnable() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + this.traceThreadPoolTaskExecutor.scheduleAtFixedRate(aRunnable(executed, span), 1L); + } + finally { + span.end(); + } + + Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> { + then(executed.get()).isTrue(); + }); + } + + @Test + public void should_schedule_at_fixed_rate_trace_runnable_with_instant() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + this.traceThreadPoolTaskExecutor.scheduleAtFixedRate(aRunnable(executed, span), Instant.now(), + Duration.ofMillis(10)); + } + finally { + span.end(); + } + + Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> { + then(executed.get()).isTrue(); + }); + } + + @Test + public void should_schedule_at_fixed_rate_trace_runnable_with_duration() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + this.traceThreadPoolTaskExecutor.scheduleAtFixedRate(aRunnable(executed, span), Duration.ofMillis(10)); + } + finally { + span.end(); + } + + Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> { + then(executed.get()).isTrue(); + }); + } + + @Test + public void should_schedule_with_fixed_delay_trace_runnable_with_date() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + this.traceThreadPoolTaskExecutor.scheduleWithFixedDelay(aRunnable(executed, span), Date.from(Instant.now()), + 1L); + } + finally { + span.end(); + } + + Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> { + then(executed.get()).isTrue(); + }); + } + + @Test + public void should_schedule_with_fixed_delay_trace_runnable() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + this.traceThreadPoolTaskExecutor.scheduleWithFixedDelay(aRunnable(executed, span), 1L); + } + finally { + span.end(); + } + + Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> { + then(executed.get()).isTrue(); + }); + } + + @Test + public void should_schedule_with_fixed_delay_trace_runnable_with_instant() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + this.traceThreadPoolTaskExecutor.scheduleWithFixedDelay(aRunnable(executed, span), Instant.now(), + Duration.ofMillis(10)); + } + finally { + span.end(); + } + + Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> { + then(executed.get()).isTrue(); + }); + } + + @Test + public void should_schedule_with_fixed_delay_trace_runnable_with_duration() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + this.traceThreadPoolTaskExecutor.scheduleWithFixedDelay(aRunnable(executed, span), Duration.ofMillis(10)); + } + finally { + span.end(); + } + + Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> { + then(executed.get()).isTrue(); + }); + } + + @Test + public void should_submit_listenable_trace_runnable() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + this.traceThreadPoolTaskExecutor.submitListenable(aRunnable(executed, span)).get(1, TimeUnit.SECONDS); + } + finally { + span.end(); + } + + Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> { + then(executed.get()).isTrue(); + }); + } + + @Test + public void should_submit_listenable_trace_callable() throws Exception { + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + Span spanFromListenable; + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + spanFromListenable = this.traceThreadPoolTaskExecutor.submitListenable(aCallable(span)).get(1, + TimeUnit.SECONDS); + } + finally { + span.end(); + } + + then(spanFromListenable).isNotNull(); + } + + @Test + public void should_execute_a_trace_runnable() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + this.traceThreadPoolTaskExecutor.execute(aRunnable(executed, span)); + } + finally { + span.end(); + } + + Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> { + then(executed.get()).isTrue(); + }); + } + + @Test + public void should_execute_with_timeout_a_trace_runnable() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + this.traceThreadPoolTaskExecutor.execute(aRunnable(executed, span), 1L); + } + finally { + span.end(); + } + + Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> { + then(executed.get()).isTrue(); + }); + } + + @Test + public void should_submit_trace_callable() throws Exception { + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + Span spanFromListenable; + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + spanFromListenable = this.traceThreadPoolTaskExecutor.submit(aCallable(span)).get(1, TimeUnit.SECONDS); + } + finally { + span.end(); + } + + then(spanFromListenable).isNotNull(); + } + + @Test + public void should_submit_trace_runnable() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + this.traceThreadPoolTaskExecutor.submit(aRunnable(executed, span)).get(1, TimeUnit.SECONDS); + } + finally { + span.end(); + } + + Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> { + then(executed.get()).isTrue(); + }); + } + + @Test + public void should_submit_trace_runnable_via_new_thread() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + this.traceThreadPoolTaskExecutor.newThread(aRunnable(executed, span)).start(); + } + finally { + span.end(); + } + + Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> { + then(executed.get()).isTrue(); + }); + } + + @Test + public void should_submit_trace_runnable_via_create_thread() throws Exception { + AtomicBoolean executed = new AtomicBoolean(); + Span span = tracerTest().tracing().tracer().nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = tracerTest().tracing().tracer().withSpan(span.start())) { + this.traceThreadPoolTaskExecutor.createThread(aRunnable(executed, span)).start(); + } + finally { + span.end(); + } + + Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> { + then(executed.get()).isTrue(); + }); + } + + Runnable aRunnable(AtomicBoolean executed, Span currentSpan) { + return () -> { + Span span = tracerTest().tracing().tracer().currentSpan(); + then(span).isNotNull(); + then(span.context().traceId()).isEqualTo(currentSpan.context().traceId()); + executed.set(true); + }; + } + + Callable aCallable(Span currentSpan) { + return () -> { + Span span = tracerTest().tracing().tracer().currentSpan(); + then(span.context().traceId()).isEqualTo(currentSpan.context().traceId()); + return span; + }; + } + +} diff --git a/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/integration/sampled/WebClientTests.java b/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/integration/sampled/WebClientTests.java index bebe8d0fd..89fad7d8f 100644 --- a/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/integration/sampled/WebClientTests.java +++ b/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/integration/sampled/WebClientTests.java @@ -356,7 +356,8 @@ public abstract class WebClientTests { } thenThereIsNoCurrentSpan(); then(this.customizer.isExecuted()).isTrue(); - then(this.spans).extracting("kind.name").contains("CLIENT"); + then(this.spans.reportedSpans().stream().filter(s -> s.getKind() != null).map(s -> s.getKind().name()) + .collect(Collectors.toList())).contains("CLIENT"); } @Test diff --git a/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/async/SleuthContextListenerAccessor.java b/tests/common/src/main/java/org/springframework/cloud/sleuth/internal/SleuthContextListenerAccessor.java similarity index 100% rename from tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/async/SleuthContextListenerAccessor.java rename to tests/common/src/main/java/org/springframework/cloud/sleuth/internal/SleuthContextListenerAccessor.java