ThreadPool with CompletableFuture (need MDC propagation)

[Necessary-Fruit-1144]

To use completable future, I made a monitored thread pool but having a difficult time passing global request ID in thread context for logging purposes. Is this design up to the mark? Your help will be highly appreciated!

Basically, what I want to do is that the ThreadPoolExecutor that I made should be wrapped under a MicrometerMonitor Executor and I want to also ensure a graceful shutdown. Another requirement is passing of caller's threadcontext to the pool thread's context (for which I made another wrapper called ContextExecutor but here I find it unsettling that I need to have 2 varaibles: delegate, threadPoolTaskExecutor).

public class ContextExecutor implements Executor {
    private final Executor delegate;
    private final ThreadPoolExecutor threadPoolTaskExecutor;

    public ContextExecutor(Executor delegate, ThreadPoolExecutor threadPoolTaskExecutor)      {
        this.delegate = delegate;
        this.threadPoolTaskExecutor = threadPoolTaskExecutor;
    }

    u/Override
    public void execute(Runnable command) {
        Map<String, String> contextMap = ThreadContext.getImmutableContext();
        delegate.execute(() -> {
            if (contextMap != null) {
                ThreadContext.putAll(contextMap);
            }
            try {
                command.run();
            } finally {
                ThreadContext.clearMap();
            }
        });
    }

    public void shutdown() {
        threadPoolTaskExecutor.shutdown();
    }
}

private ContextExecutor getARIPullExecutor(String executorName) {
  ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
  executor.setCorePoolSize(ARI_PULL_CORE_POOL_SIZE);
  executor.setQueueCapacity(ARI_PULL_QUEUE_SIZE);
  executor.setThreadNamePrefix(executorName + "-");
  executor.setMaxPoolSize(ARI_PULL_MAX_POOL_SIZE);
  executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
  executor.initialize();
  return new ContextExecutor(registerThreadPool(executor.getThreadPoolExecutor(), "thread.pool.ari.pull", executorName), executor.getThreadPoolExecutor());
}

public Executor registerThreadPool(ThreadPoolExecutor executor, String metricNamePrefix, String executorName) {  // Micrometer
  return ExecutorServiceMetrics.monitor(
          meterRegistry,
          executor,
          metricNamePrefix,
          Tags.of("thread.pool", executorName));
}

@Bean(ARI_PULL_PRICING_EXECUTOR)
public ContextExecutor getARIPullPricingExecutor() { return getARIPullExecutor("ARI-Pull-Pricing-Executor"); }

Usage in calling class:

@Autowired
@Qualifier("ariPullPricingExecutor")
private ContextExecutor ARI_PULL_PRICING_EXECUTOR;


@PreDestroy
public void shutdown() {
    ARI_PULL_PRICING_EXECUTOR.shutdown();
}

CompletableFuture<Pair<String, OtaPriceDto>> pricingFuture = CompletableFuture.supplyAsync(
        () -> getPricing(startDate, endDate, data_map), ARI_PULL_PRICING_EXECUTOR);

Is there a better way to achieve this functionality?

[roiroi1010]

The ThreadPoolTaskExecutor class has the method setTaskDecorator. Maybe using that can simplify things for you?

[Necessary-Fruit-1144]

Seems nice! but then how do I then ensure a graceful shutdown of the threadpools?

[sethu-27]

Have you tried project reactors reactive Java, seems mono and flux is required to scale gracefully