Use @Async and Application Events Deliberately

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

package com.example.async;

import org.slf4j.MDC;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.TaskDecorator;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.Map;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;

@Configuration(proxyBeanMethods = false)
@EnableAsync
public class AsyncThreadPoolConfig {

    @Bean(name = "orderProcessingExecutor")
    public Executor orderProcessingExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10);
        executor.setMaxPoolSize(50);
        executor.setQueueCapacity(500);
        executor.setThreadNamePrefix("order-async-");
        // Rejection policy: caller thread executes task when 500-slot queue is saturated
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.setAwaitTerminationSeconds(30);

        // Propagate SLF4J MDC (traceId, correlationId) from caller to async thread
        executor.setTaskDecorator(new MdcTaskDecorator());
        executor.initialize();
        return executor;
    }

    static class MdcTaskDecorator implements TaskDecorator {
        @Override
        public Runnable decorate(Runnable runnable) {
            Map<String, String> contextMap = MDC.getCopyOfContextMap();
            return () -> {
                try {
                    if (contextMap != null) {
                        MDC.setContextMap(contextMap);
                    }
                    runnable.run();
                } finally {
                    MDC.clear();
                }
            };
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

package com.example.order;

import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.springframework.transaction.event.TransactionPhase;
import org.springframework.transaction.event.TransactionalEventListener;

@Component
public class OrderNotificationListener {

    private final EmailGateway emailGateway;

    public OrderNotificationListener(EmailGateway emailGateway) {
        this.emailGateway = emailGateway;
    }

    // Only fires if the order transaction successfully commits to PostgreSQL!
    @Async("orderProcessingExecutor")
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void onOrderPlaced(OrderPlacedEvent event) {
        emailGateway.sendReceipt(event.customerId(), event.orderId());
    }
}

1
2
3
4

spring:
  threads:
    virtual:
      enabled: true

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

package com.example.async;

import org.junit.jupiter.api.Test;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

import static org.junit.jupiter.api.Assertions.*;

@SpringBootTest
class AsyncExecutorMdcTest {

    @Autowired
    private AsyncOrderService orderService;

    @Test
    void verifiesMdcPropagationAcrossAsyncThreadBoundary() throws Exception {
        String expectedTraceId = "req-trace-789";
        MDC.put("traceId", expectedTraceId);

        try {
            // Service method captures MDC.get("traceId") and Thread.currentThread().getName() inside @Async task
            CompletableFuture<AsyncExecutionResult> future = orderService.processOrderAsync("ORD-202");
            AsyncExecutionResult result = future.get(5, TimeUnit.SECONDS);

            // 1. Verify task executed on dedicated thread pool
            assertTrue(result.threadName().startsWith("order-async-"),
                "Expected execution on custom thread pool, but ran on: " + result.threadName());

            // 2. Verify correlation ID was faithfully propagated by TaskDecorator
            assertEquals(expectedTraceId, result.capturedTraceId(),
                "TaskDecorator must copy MDC context map to the async worker thread");
        } finally {
            MDC.clear();
        }
    }

    @Test
    void deliberateFailure_rawThreadLacksMdcWithoutDecorator() throws Exception {
        MDC.put("traceId", "transient-id");
        try {
            // Executing on an un-decorated raw executor demonstrates the silent telemetry failure
            CompletableFuture<String> rawFuture = CompletableFuture.supplyAsync(() -> MDC.get("traceId"));
            String capturedInWorker = rawFuture.get(5, TimeUnit.SECONDS);

            // Deliberate failure proof: Raw worker thread has an empty ThreadLocal!
            assertNull(capturedInWorker,
                "Without TaskDecorator, ThreadLocal MDC is null on the new thread");
        } finally {
            MDC.clear();
        }
    }
}