republishing message into RabbitMQ queue creates an infinite loop

Question

I have a RabbitMQ queue to hold unprocessed messages. I happy path, I will read message from the queue, process it, and removes the message in the queue. But if certain criteria are met while processing I have to republish the message to the queue again. I am using a pollable channel adapter to fetch the message. since I want to fetch all the available messages in that queue while polling I have set the maxMessagesPerPoll to -1. This causes the code to go in an infinite loop. after republishing the message into the queue, the inbound polled adapter picks it up immediately. How can I prevent this situation?

Is there any way to delay the message delivery or can we restrict the message processing once per message in single polling of the InboundPolledAdapter. What will be the best approach?

The inboundPolledAdapter is,

@Bean
public IntegrationFlow inboundIntegrationFlowPaymentRetry() {
    return IntegrationFlows
            .from(Amqp.inboundPolledAdapter(connectionFactory, RetryQueue),
                    e -> e.poller(Pollers.fixedDelay(20_000).maxMessagesPerPoll(-1)).autoStartup(true))
            .handle(message -> {
                channelRequestFromQueue()
                        .send(MessageBuilder.withPayload(message.getPayload()).copyHeaders(message.getHeaders())
                                .setHeader(IntegrationConstants.QUEUED_MESSAGE, message).build());
            }).get();
}

For the first posting of first message to the queue by,

@Bean
Binding bindingRetryQueue() {
    return BindingBuilder.bind(queueRetry()).to(exchangeRetry())
            .with(ProcessQueuedMessageService.RETRY_ROUTING_KEY);
}

@Bean
TopicExchange exchangeRetry() {
    return new TopicExchange(ProcessQueuedMessageService.RETRY_EXCHANGE);
}

@Bean
Queue queueRetry() {
    return new Queue(RetryQueue, false);
}

@Bean
@ServiceActivator(inputChannel = "channelAmqpOutbound")
public AmqpOutboundEndpoint outboundAmqp(AmqpTemplate amqpTemplate) {
    final AmqpOutboundEndpoint outbound = new AmqpOutboundEndpoint(amqpTemplate);
    outbound.setRoutingKey(RetryQueue);
    return outbound;
}

Republishing message by,

StaticMessageHeaderAccessor.getAcknowledgmentCallback(requeueMessage).acknowledge(Status.REQUEUE);

Artem Bilan Artem Bilan · Accepted Answer · 2021-10-01T13:21:09

Is there any way to delay the message delivery

See Delayed Exchange feature in Rabbit MQ and its API in Spring AMQP: https://docs.spring.io/spring-amqp/docs/current/reference/html/#delayed-message-exchange

restrict the message processing once per message

For this scenario you can take a look into Idempotent Receiver pattern and its implementation in Spring Integration: https://docs.spring.io/spring-integration/docs/current/reference/html/messaging-endpoints.html#idempotent-receiver.

The redelivered message is going to have an AmqpHeaders.REDELIVERED header. See more in docs: https://www.rabbitmq.com/reliability.html#consumer-side

republishing message into RabbitMQ queue creates an infinite loop

1 Answers