Publisher Confirms

Channel channel = connection.createChannel();

// Put channel in confirm mode — every publish gets a sequence number
channel.confirmSelect();

// Async confirm listener — fires on broker ack or nack
channel.addConfirmListener(
    (deliveryTag, multiple) -> {
        // ACK: broker persisted the message
        log.info("Confirmed delivery tag {}", deliveryTag);
    },
    (deliveryTag, multiple) -> {
        // NACK: broker could NOT persist — resend or alert
        log.error("Broker nacked delivery tag {} — message lost!", deliveryTag);
        retryPublish(deliveryTag);
    }
);

// Publish a persistent message
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
    .deliveryMode(2)   // persistent
    .build();

channel.basicPublish("orders", "order.created", props, messageBody);

// Synchronous wait — blocks until all pending confirms arrive (lower throughput)
boolean allConfirmed = channel.waitForConfirms(5000); // 5s timeout
if (!allConfirmed) {
    throw new RuntimeException("Some messages not confirmed by broker");
}

# application.yml
spring:
  rabbitmq:
    publisher-confirm-type: correlated   # enables publisher confirms
    publisher-returns: true              # enables mandatory routing returns

@Configuration
public class RabbitConfig {
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory cf) {
        RabbitTemplate template = new RabbitTemplate(cf);
        template.setMandatory(true);  // returns unroutable messages

        // Called when broker ACKs or NACKs the message
        template.setConfirmCallback((correlationData, ack, cause) -> {
            String messageId = correlationData != null ? correlationData.getId() : "?";
            if (ack) {
                log.info("Message {} confirmed by broker", messageId);
            } else {
                log.error("Message {} NOT confirmed: {}", messageId, cause);
                // Re-queue for retry / alert
            }
        });

        // Called when a message is returned (not routed to any queue)
        template.setReturnsCallback(returned -> {
            log.warn("Message returned — not routed: {} → {} {}",
                returned.getRoutingKey(), returned.getReplyCode(), returned.getReplyText());
        });

        return template;
    }
}

// Publisher — pass CorrelationData for traceability
@Service
@RequiredArgsConstructor
public class OrderPublisher {
    private final RabbitTemplate rabbitTemplate;

    public void publish(Order order) {
        CorrelationData cd = new CorrelationData(order.getId().toString());
        rabbitTemplate.convertAndSend("orders", "order.created", order, cd);
    }
}

// AVOID: channel transactions — very slow
channel.txSelect();
try {
    channel.basicPublish(...);
    channel.basicPublish(...);
    channel.txCommit();  // synchronous, blocks, ~250x slower than confirms
} catch (Exception e) {
    channel.txRollback();
}

// PREFER: publisher confirms — async, fast, same safety guarantee
channel.confirmSelect();
channel.addConfirmListener(ackCallback, nackCallback);
channel.basicPublish(...);
channel.basicPublish(...);
// No blocking — confirm callbacks fire asynchronously