Actuator health check for SQS Listener Container

[jubinjacobthomas]

I am using the SQS messaging features in Spring Cloud AWS (version 3.0.0-M2).
Have created an sqs listener as below

    @SqsListener(value = "${sqs.queue-url}", factory = "testSqsListenerContainerFactory")
    public void getEvent(SQSPayload sqsPayload) throws JsonProcessingException {
        log.info("Received message= {}", sqsPayload);
       // Business logic here
    }

and SqsListenerContainerFactory bean as below

    @Bean
    SqsMessageListenerContainerFactory<Object> testSqsListenerContainerFactory(SqsAsyncClient sqsAsyncClient, ErrorHandler<Object> errorHandler) {
        return SqsMessageListenerContainerFactory
                .builder()
                .configure(options -> options
                        .maxInflightMessagesPerQueue(50)
                        .maxMessagesPerPoll(20)
                        .pollTimeout(Duration.ofSeconds(10)))
                .sqsAsyncClient(sqsAsyncClient)
                .errorHandler(errorHandler)
                .build();
    }

I would like to create a custom actuator HealthIndicator that could detect if the container is running or not . I have seen that SqsMessageListenerContainer class has an isRunning method but i have no idea how to access that object, since the container is created via @SqsListener annotation and factory class. Is there anyway to create this custom HealthIndicator to detect unhealthy services that are supporting the SQS messaging.

Note: My requirement is same as the one mentioned here : spring-attic/spring-cloud-aws#317

[maciejwalkowiak]

cc @tomazfernandes

[tomazfernandes]

Hi @jubinjacobthomas, thanks for looking into this.

You can inject / lookup the MessageListenerContainerRegistry bean, which has methods to retrieve all or an individual container:

spring-cloud-aws/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/MessageListenerContainerRegistry.java

Lines 22 to 52 in 57737cf

	/**
	* Interface for registering and looking up containers at startup and runtime.
	*
	* @author Tomaz Fernandes
	* @since 3.0
	* @see DefaultListenerContainerRegistry
	*/
	public interface MessageListenerContainerRegistry extends SmartLifecycle {
	/**
	* Register a {@link MessageListenerContainer} instance with this registry.
	* @param listenerContainer the instance.
	*/
	void registerListenerContainer(MessageListenerContainer<?> listenerContainer);
	/**
	* Return the {@link MessageListenerContainer} instances registered within this registry.
	* @return the container instances.
	*/
	Collection<MessageListenerContainer<?>> getListenerContainers();
	/**
	* Return the {@link MessageListenerContainer} instance registered within this registry with the provided id, or
	* null if none.
	* @param id the id.
	* @return the container instance.
	*/
	@Nullable
	MessageListenerContainer<?> getContainerById(String id);
	}

If your use case is just making sure all queues are present / accessible at startup, you can set QueueNotFoundStrategy.FAIL in the ContainerOptions. Default is CREATE and we currently don't offer a IGNORE option.

I'm not sure if it's worth it just checking if all containers are running, as currently the only way to stop a container is manually or on application shutdown.

Would it be worth it adding an IGNORE option, and maybe stopping the container on some exceptions (such as QueueDoesNotExistException)? I think it's strange to assume queues can simply not be there on a running app, but I'm open to that if it makes sense.

Please let us know if this works for you!

[tomazfernandes]

Note: The framework registers the Registry bean with all “@SqsListener” containers at startup, so you can simply inject it.

[jubinjacobthomas]

Hi @tomazfernandes, Thanks for your suggestion. The idea here is to create a healthcheck endpoint using actuator which our infrastructure(Kubernetes) can use to restart the app if the sqs listeners get stopped due to any error. Based on your suggestion I came up with this

@Component
public class SqsListenerHealthIndicator implements HealthIndicator {

    @Autowired
    MessageListenerContainerRegistry registry;

    @Override
    public Health health() {
        Collection<MessageListenerContainer<?>> containers =  registry.getListenerContainers();
        for(MessageListenerContainer<?> container: containers) {
            if(!container.isRunning()) {
                return Health.down().build();
            }
        }
        return Health.up().build();
    }

}

[maciejwalkowiak]

Looks good to me!

[tomazfernandes]

Well, as I mentioned, this should never return false unless users stop the container manually, or the application is shutting down.

Currently the only error that can stop the container is InterruptedException, and it doesn’t change the running status to false. AFAIK, since users don’t have access to the container thread, the only way the thread can be interrupted is at application shutdown, so the running flag will have been set to false already.

Maybe I’m missing something?

Anyway, you can give it a try and let us know how it works out!

[jubinjacobthomas]

Thanks. I decided to not use this approach since this method always returns true when app is running. I will instead go for a queuehealthlistener which will check if sqs queue os active or not using sqsAsyncClient.getQueueUrl() method.
Anyways MessageListenerContainerRegistry was new info for me which could be useful in the future.