This is my listener code:
@MessageExceptionHandler(MessageConversionException.class)
@SqsListener(value = "${cloud.aws.sqs.url}",deletionPolicy = SqsMessageDeletionPolicy.ON_SUCCESS)
public void receiveEvent(User user){
handleRequest(user);
}
I want to delete a message if a MessageConversionException occurs but the message is being read again and again.
Is there any way to delete the message from the SQS queue in case we are unable to deserialize from the object received?
Looking at the official documentation, you will need to use SqsMessageDeletionPolicy.ALWAYS.
Using ON_SUCCESS will only delete the message if everything was successful in receiving and processing a message.
Defines the policy that must be used for the deletion of SQS messages once they were processed. The deletion policy can be set individually on every listener method using the @SqsListener annotation. The default policy is NO_REDRIVE because it is the safest way to avoid poison messages and have a safe way to avoid the loss of messages (i.e. using a dead letter queue).
The following deletion policies are available:
ALWAYS: Always deletes message in case of success (no exception thrown) or failure (exception thrown) during message processing by the listener method.
NEVER: Never deletes message automatically. The receiving listener method must acknowledge each message manually by using the acknowledgment parameter.
NO_REDRIVE: Deletes message if no redrive policy is defined.
ON_SUCCESS: Deletes message when successfully executed by the listener method. If an exception is thrown by the listener method, the message will not be deleted.
If you want to manuall acknowedge that the message must be deleted, then your other option is NEVER
--- If you wanted to manually acknowledge the deletion ----
@SqsListener(value = Queue.PORTAL_TEST, deletionPolicy = SqsMessageDeletionPolicy.NEVER)
@SendTo(value = Queue.PORTAL_TEST_B)
public UserDto receive(@Payload UserDto userDto
, @Headers Map<String, String> headers
, @Header(name = "ReceiptHandle") String receiptHandle
, Acknowledgment acknowledgment) throws Exception {
log.info("threadId : {}, currentTime : {}", Thread.currentThread().getId(), System.currentTimeMillis());
log.info("message : {}", userDto.toString());
//do something with user
acknowledgment.acknowledge();
return userDto;
}
--- Edited ----
Please separate your exception method from the listener:
@SqsListener(value = Queue.PORTAL_TEST, deletionPolicy = SqsMessageDeletionPolicy.NEVER)
public UserDto receive(@Payload UserDto userDto
, @Headers Map<String, String> headers
, @Header(name = "ReceiptHandle") String receiptHandle
, Acknowledgment acknowledgment) throws Exception {
}
@MessageExceptionHandler(MessageConversionException.class)
public void exceptionHandler(MessageConversionException e) {
LOG.info("Failed to deserialize that pojo", e);
}
you can use NO_REDRIVE policy - always deletes the message on normal completion, but when an exception is thrown it is only deleted if there is no DLQ and redrive policy set up for the queue. If the DLQ and redrive policy are set up, the message is not deleted.