AmqpOutboundEndpoint publisher confirms timeout?

[bojanv55]

I have workflow where PUBLISHER CONFIRM is waited for and then on receiving of this message, I manually ACK messages on other queue.

If I put this config for AmqpOutboundEndpoint:

AmqpOutboundEndpoint outboundEndpoint = new AmqpOutboundEndpoint(ackTemplate());
outboundEndpoint.setConfirmNackChannel(nackFromPublisherConfirmsChannel())

and ackTemplate uses connectionFactory with publisher-confirms: true, if one of ACK (PUB. CONF) messages from server to client is dropped, I never ACK message on other queue.

If inside AmqpOutboundEndpoint in the method send instead of this code

((RabbitTemplate) this.amqpTemplate).send(exchangeName, routingKey, amqpMessage, correlationData);

I use this:

RabbitTemplate rt = (RabbitTemplate)this.amqpTemplate;

			rt.invoke(t -> {
				t.send(exchangeName, routingKey, amqpMessage, correlationData);
				t.waitForConfirmsOrDie(15_000);
				return true;
			});

everything works as expected - ACK is timed out, and we forwad message to nackFromPublisherConfirmsChannel.

[garyrussell]

It's not clear what you are asking; in one case you are looking for NACKs and the other you don't care about ACK Vs. NACK.

Dropped records are not NACKed, they are ACKed. See the rabbitmq documentation.

If you need to detect failure to route a message to a queue, you need returns, not confirms.

[bojanv55]

OK, let me explain more. The code that I'm testing is at https://github.com/bojanv55/spring-integration-aggregate-ack.

I have 2 queues q1 and q2 that are read, aggregated and this aggregated message is delivered to ex1 exchange. q1 and q2 are Manual ACK. When I send message to ex1, I'm expecting one of : [ACK, NACK]. In case of ACK, I am manually ACK-ing message on q1 or q2. In case of NACK, I am manually requeueing message on q1 or q2 (NACK + requeue).

I made my test env with: spring client -> toxiproxy -> rabbitmq, and in toxiproxy I coded toxic to drop one of PUBLISHER CONFIRMS from rabbitmq -> spring client. In case this is dropped, spring client never received PUBLISHER CONFIRMS and cannot ACK/NACK message on q1 or q2. When I add this timeout, it works as expected for me.

I hope it is more clear now?

What I wanted to achieve with this is that if spring client process dies, messages in q1 and q2 will be NACKED + requeued. If there is some problem in downstream (ex1 delivery), messages will be returned to q1 and q2 and process is retried.

[bojanv55]

I just noticed that I have to add this:

ackTemplate.setMandatory(true);

in order to have rabbitmq RETURNS in case of failure to route message (I just found out that after RETURN we would also have ACK or NACK message from broker for that message. So ACK/NACK from broker is always sent no matter if RETURNS is there or not).

I also guess that here are possible scenarios for reply from rabbit when sending message to rabbitmq exchange (in my case ex1), when I set publisher-confirms: true and publisher-returns: true on connection and also when I set ackTemplate.setMandatory(true);:

1. exhange ex1 exists and there is a queue bound to ex1 with needed key
   • message is delivered to queue and rabbitmq PUBLISHER_CONFIRM message
   • message is delivered to queue and rabbitmq PUB_CONF, but somebody in between drops the message -> spring client is going to hang forever waiting for response from rabbitmq
2. exchange ex1 does not exist
   • message is sent to ex1, but since no exchange, setConfirmNackChannel is invoked
3. exchange ex1 exists, but not bound to any queue (not routable)
   • message is sent to ex1, but since cannot be routed, setReturnChannel is invoked AND ACK IS ALSO INVOKED (setConfirmAckChannel)

[artembilan]

Looks like this is related to this SO: https://stackoverflow.com/questions/54713121/spring-integration-publisher-confirms-with-timeout

[garyrussell]

@bojanv55 In Spring AMQP version 2.1 (Spring Integration 5.1), We added a Future<?> and returned message to the CorrelationData to assist with this kind of thing. If you are using an older version, you can subclass CorrelationData (and you'd have to handle setting the future and returned message in your code).

This, together with a scheduled task can detect missing acks...

@SpringBootApplication
@EnableScheduling
public class Igh2755Application {

	public static void main(String[] args) {
		SpringApplication.run(Igh2755Application.class, args);
	}

	private final BlockingQueue<CorrelationData> futures = new LinkedBlockingQueue<>();

	@Bean
	public ApplicationRunner runner(RabbitTemplate template) {
		return args -> {
			SuccessCallback<? super Confirm> successCallback = confirm -> {
				System.out.println((confirm.isAck() ? "A" : "Na") + "ck received");
			};
			FailureCallback failureCallback = throwable -> {
				System.out.println(throwable.getMessage());
			};

			// Good - ack
			CorrelationData correlationData = new CorrelationData("good");
			correlationData.getFuture().addCallback(successCallback, failureCallback);
			this.futures.put(correlationData);
			template.convertAndSend("", "foo", "data", correlationData);

			// Missing exchange nack, no return
			correlationData = new CorrelationData("missing exchange");
			correlationData.getFuture().addCallback(successCallback, failureCallback);
			this.futures.put(correlationData);
			template.convertAndSend("missing exchange", "foo", "data", correlationData);

			// Missing queue ack, with return
			correlationData = new CorrelationData("missing queue");
			correlationData.getFuture().addCallback(successCallback, failureCallback);
			this.futures.put(correlationData);
			template.convertAndSend("", "missing queue", "data", correlationData);
		};
	}

	@Scheduled(fixedDelay = 5_000)
	public void checkForMissingAcks() {
		System.out.println("Checking pending acks");
		CorrelationData correlationData = this.futures.poll();
		while (correlationData != null) {
			try {
				if (correlationData.getFuture().get(10, TimeUnit.SECONDS).isAck()) {
					if (correlationData.getReturnedMessage() == null) {
						System.out.println("Ack received OK for " + correlationData.getId());
					}
					else {
						System.out.println("Message returned for " + correlationData.getId());
					}
				}
				else {
					System.out.println("Nack received for " + correlationData.getId());
				}
			}
			catch (InterruptedException e) {
				Thread.currentThread().interrupt();
				System.out.println("Interrupted");
			}
			catch (ExecutionException e) {
				System.out.println("Failed to get an ack " + e.getCause().getMessage());
			}
			catch (TimeoutException e) {
				System.out.println("Timed out waiting for ack for " + correlationData.getId());
			}
			correlationData = this.futures.poll();
		}
		System.out.println("No pending acks, exiting");
	}

}

Checking pending acks
Ack received OK for good
Nack received for missing exchange
Message returned for missing queue
No pending acks, exiting

With Spring Integration there is a confirmCorrelationExpression which can be used to create the CorrelationData instance.

[bojanv55]

Thanks for your answer on Stackoverflow.

Just to write here also - are there any plans to include publisher confirms timeout in framework? I expected this to be not that common in production, but since last week I deployed new code, it seems that it timeouts once in 7 days (2 messages were having pub. confirm timeout) - so it is not that uncommon as I thought.

[garyrussell]

I've marked it as an enhancement.

[garyrussell]

Duplicate of #2797