Propagation a local thread variable from an user of SqsAsyncClient

[sondemar]

Hi, I am trying to propagate a local thread variable from an user of SqsAsyncClient, which returns an asynchronous CompletableFuture.

So far, I have explored two possible configurations of executors. Still, these executors seem to be invoked directly from the thread of Netty’s Event Loop Model, rather than from the user's thread of SqsAsyncClient.

Below, I have attached a simple example demonstrating this scenario:

...
  static ThreadLocal<String> value = new ThreadLocal<>();
  
  void testSqsClient() {
      value.set("visible log in the thread");
      ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
      executor.setTaskDecorator(DecoratedRunnable::new);
      SqsAsyncClient asyncClient = SqsAsyncClient.builder()
              .asyncConfiguration(builder -> builder.advancedOption(FUTURE_COMPLETION_EXECUTOR, executor))
              .httpClientBuilder(NettyNioAsyncHttpClient.builder().eventLoopGroup(SdkEventLoopGroup.builder().threadFactory(new ThreadLocalPropagationThreadFactory()).build()))
              .build();


    CompletableFuture<Void> cf = CompletableFuture.completedFuture("created message: ")
            .thenCombine(asyncClient.sendMessage(builder -> builder.messageBody("test")), (s, messageResponse) -> s + messageResponse.messageId())
            .thenAcceptAsync(s -> log.info("result: {}", s), executor);
    cf.join();
}

static class DecoratedRunnable implements Runnable {
    private final String propagatedValue;
    private final Runnable original;

    public DecoratedRunnable(Runnable original) {
        this.propagatedValue = value.get();
        this.original = original;
    }

    @Override
    public void run() {
        log.info(propagatedValue);
        original.run();
    }
}

static class ThreadLocalPropagationThreadFactory implements ThreadFactory {
    private final ThreadFactory delegate = new ThreadFactoryBuilder()
                .build();

    @Override
    public Thread newThread(Runnable runnable) {
        return delegate.newThread(new DecoratedRunnable(runnable));
    }
}
....

In the above example, the ThreadLocalPropagationThreadFactory is invoked by the event loop model thread instead of the user's thread.

Could you please advise on how to properly propagate the local thread variable in this context (I expect that "visible log in the thread" will be logged in the console log)?

This question is closely linked to the following issue

[github-actions[bot]]

Hello! Reopening this discussion to make it searchable.