Best way of creating custom MessageListenerContainer beans

[otbe]

Hi there,

in my current project we want to create some custom listener containers and when looking at the docs its basically as simple as

@Bean
MessageListenerContainer<Object> listenerContainer(SqsAsyncClient sqsAsyncClient) {
    SqsMessageListenerContainer<Object> container = new SqsMessageListenerContainer<>(sqsAsyncClient);
    container.setMessageListener(System.out::println);
    container.setQueueNames("myTestQueue");
    return container;
}

This works just fine but those containers are not the same way created than the default one when you use @SqsListener. One subtle difference is that they don't go through io.awspring.cloud.sqs.config.AbstractEndpoint#setupContainer because we only specify a string as queue name and not an SqsEndpoint.

Why do I care?
I think that we skip this part of the code is the reason I dont get any instrumentation for my listener class in terms of auto instrumentation by the opentelemetry java agent. If I get it right it looks exclusively for a MessagingMessageListenerAdapter here
I already use the latest version of Spring AWS Cloud and the latest agent version (2.18.1) with the fix for open-telemetry/opentelemetry-java-instrumentation#14207. In another project (with same spring aws cloud &agent version) where we just use @SqsListener instrumentation works as expected, so if the SQS message contains tracing information its attached.

I tried to mimic the same behaviour of setting up the container but it's quite hard. Do you have any better idea how can I achieve this? I know its not really a problem of this project but you know it best :D

[otbe]

I now came up with something like

@Bean
MessageListenerContainer<Object> listenerContainer(SqsAsyncClient sqsAsyncClient, MyListener listener) {
    SqsMessageListenerContainer<Object> container = new SqsMessageListenerContainer<>(sqsAsyncClient);
    container.setMessageListener(createListenerAdapter(listener));
    container.setQueueNames("myTestQueue");
    return container;
}

 @SneakyThrows
  private MessagingMessageListenerAdapter<Object> createListenerAdapter(
      @NonNull MyListener listener) {
    final var handlerMethodFactory = createDefaultHandlerMethodFactory();

    return new MessagingMessageListenerAdapter<>(
        handlerMethodFactory.createInvocableHandlerMethod(
            listener,
            listener
                .getClass()
                .getMethod("onMessage", org.springframework.messaging.Message.class)));
  }

  private DefaultMessageHandlerMethodFactory createDefaultHandlerMethodFactory() {
    final var handlerMethodFactory = new DefaultMessageHandlerMethodFactory();
    handlerMethodFactory.setArgumentResolvers(
        List.of(
            new MessageMethodArgumentResolver(
                createDefaultMappingJackson2MessageConverter(objectMapper))));
    handlerMethodFactory.afterPropertiesSet();
    return handlerMethodFactory;
  }

  private MappingJackson2MessageConverter createDefaultMappingJackson2MessageConverter(
      @NonNull ObjectMapper objectMapper) {
    MappingJackson2MessageConverter jacksonMessageConverter = new MappingJackson2MessageConverter();
    jacksonMessageConverter.setSerializedPayloadClass(String.class);
    jacksonMessageConverter.setStrictContentTypeMatch(false);
    jacksonMessageConverter.setObjectMapper(objectMapper);
    return jacksonMessageConverter;
  }

This seems to work. Autoinstrumentation is back and works as expected. Do you see any downside in doing this?
I only use springs Message class as single parameter in my listener (no other resolvers needed in that case)

[tomazfernandes]

Hey @otbe, happy you found a way of making it work.

The Endpoint abstraction represents an endpoint created with the SqsListener annotation and is a vehicle for its configuration, that's the reason why when you create the container directly it skips that part of the code and it takes the configuration directly from the ContainerOptions.

From your solution it seems that the opentelemetry-java is instrumenting the MessagingMessageListenerAdapter or MessageMethodArgumentResolver abstractions. Those are used to dynamically resolve the method invocation in the listener, hence they're really not necessary if you're providing the listener logic manually. It shouldn't be an issue, but it adds some unnecessary overhead and complexity to your configuration.

In my understanding, unless they're also instrumenting async listener methods, which is tricky, it might be better for them to instrument the AsyncComponentAdapters, which now contains a few entry points for opening scopes and handling ThreadLocals.

I'm not sure if they're interested in instrumenting manual container creation, but perhaps you could open an issue in their project, and I'd be happy to collaborate with them in figuring a solution out.

Let me know your thoughts, thanks.

[otbe]

@tomazfernandes thanks for your answer. That makes total sense.
I created open-telemetry/opentelemetry-java-instrumentation#14339.

For now the overhead in configuration is fine for us because it’s in an internal library we then use in our services.