То же сообщение повторяется и не отправляется на dlq

Question

Я определил очередь - blah.queue и определил для нее мертвую букву-ключ маршрутизации.Тем не менее, когда у меня есть подозрительное сообщение, которое выходит из строя с исключением, то же самое сообщение помещается в очередь и повторно пытается (в бесконечном цикле).Я ожидаю, что после 3 повторных попыток сообщение должно быть помещено в обмен с ключом маршрутизации мертвой буквы.Но, похоже, этого не происходит.

У меня есть следующие настройки:

    <rabbit:queue name="blah.queue" auto-delete="false" durable="true">
    <rabbit:queue-arguments>
        <entry key="x-dead-letter-exchange" value="" />
        <entry key="x-dead-letter-routing-key" value="blah.queue.dlq.route" />
        <entry key="x-ha-policy" value="all" />
    </rabbit:queue-arguments>
</rabbit:queue>
<rabbit:direct-exchange name="${rabbit.idesk.exchange}">
    <rabbit:bindings>
        <rabbit:binding queue="blah.queue" key="blah.route" />
    </rabbit:bindings>
</rabbit:direct-exchange>

<bean id="myConsumer" class="com.ankit.CustomConsumer" />

<bean id="myConsumerMessageListenerAdapter" class="org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter">
    <constructor-arg ref="myConsumer" />
    <constructor-arg ref="myMessageConverter" />
</bean>

<bean id="myConsumerMessageListenerContainer" class="org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer">
    <property name="queueNames" value="blah.queue" />
    <property name="connectionFactory" ref="queueConnectionFactory" />
    <property name="messageListener" ref="myConsumerMessageListenerAdapter" />
    <property name="errorHandler" ref="loggingErrorHandler" />
    <property name="adviceChain">
        <list>
            <ref bean="retryAdvice" />
        </list>
    </property>
</bean>

<bean id="loggingErrorHandler" class="org.springframework.scheduling.support.TaskUtils.LoggingErrorHandler" />

<bean id="myMessageConverter"
    class="org.springframework.amqp.support.converter.JsonMessageConverter">
    <property name="classMapper">
        <bean class="com.ankit.queue.mapper.NamedClassMapper">
            <constructor-arg
                value="com.ankit.dto.EventDTO" />
        </bean>
    </property>
    <property name="createMessageIds" value="true" />
</bean>

<bean id="retryAdvice" class="org.springframework.amqp.rabbit.config.StatefulRetryOperationsInterceptorFactoryBean">
    <property name="messageRecoverer" ref="rejectAndDontRequeueRecoverer" />
    <property name="retryOperations" ref="retryTemplate" />
</bean>

<bean id="retryTemplate" class="org.springframework.retry.support.RetryTemplate">
    <property name="retryPolicy" ref="simpleRetryPolicy" />
    <property name="backOffPolicy">
        <bean class="org.springframework.retry.backoff.FixedBackOffPolicy">
            <property name="backOffPeriod" value="5000" />
        </bean>
    </property>
</bean>

<bean id="simpleRetryPolicy" class="org.springframework.retry.policy.SimpleRetryPolicy">
    <property name="maxAttempts" value="3" />
</bean>

Answer 1

Разве вы не видите сообщение WARN от восстановителя?

@Override
public void recover(Message message, Throwable cause) {
    if (this.logger.isWarnEnabled()) {
        this.logger.warn("Retries exhausted for message " + message, cause);
    }
    throw new ListenerExecutionFailedException("Retry Policy Exhausted",
                new AmqpRejectAndDontRequeueException(cause), message);
}

Включите ведение журнала DEBUG, чтобы посмотреть поведение повторных попыток.