How to manually ack messages from rabbitmq using spring integration

Question

[EDIT] Uploading complete configs:

rabbit.xml which dequeues from rabbit

<rabbit:connection-factory id="amqpConnectionFactoryInbound" 
host="${rabbit.host}" port="${rabbit.port}"
username="${rabbit.username}" password="${rabbit.password}" channel-
cache-size="5"
connection-factory="rabbitConnectionFactoryInbound"/>

<beans:bean id="rabbitConnectionFactoryInbound" 
class="com.rabbitmq.client.ConnectionFactory">
<beans:property name="requestedHeartbeat" 
value="60" />
</beans:bean>


<!-- Inbound Adapter to AMQP RabbitMq and write to file -->
<int-amqp:inbound-channel-adapter id="rabbitMQInboundChannelAdapter" 
channel="rabbitInboundMessageChannel"
concurrent-consumers="8" task-
executor="rabbit-executor" connection-
factory="amqpConnectionFactoryInbound"
message-converter="byteArrayToStringConverter" queue-
names="${rabbit.queue}" acknowledge-mode="MANUAL" error-
channel="errorChannelId"
prefetch-count="25" />

<header-enricher input-channel="rabbitInboundMessageChannel" output-
channel="rabbitOutboundboundMessageChannel">
<int:header name="Operation" value="${operation.rabbit}" />
<int:header name="GUID" expression="#{ 
'T(java.util.UUID).randomUUID().toString()' }" />
<int:header name="operationStartTime" expression="#{ 
'T(java.lang.System).currentTimeMillis()' }" />
</header-enricher>

<int:channel id="rabbitOutboundboundMessageChannel">
<int:interceptors>
<int:wire-tap channel="loggerChannel" />
</int:interceptors>
</int:channel>

<task:executor id="rabbit-executor" rejection-policy="CALLER_RUNS" 
pool-size="10-30"
queue-capacity="25" />
</beans:beans>

The message is then sent to router channel: router.xml

<int:header-enricher input-channel="rabbitOutboundboundMessageChannel" 
output-channel="routerChannel">
<int:header name="Operation" value="${operation.router}" 
overwrite="true" />
<int:header name="file_name" expression="headers['GUID'] + '.xml'" />
<int:header name="operationStartTime" expression="#{ 
'T(java.lang.System).currentTimeMillis()' }"
overwrite="true" />
<int:error-channel ref="errorChannelId" />
</int:header-enricher>

<int:recipient-list-router id="rabbitMsgrouter" input-
channel="routerChannel">
<int:recipient channel="fileBackupChannel" selector-expression="new 
String(payload).length()>0" />
<int:recipient channel="transformerChannel" />
</int:recipient-list-router>

<int:channel id="transformerChannel">
<int:interceptors>
<int:wire-tap channel="loggerChannel" />
</int:interceptors>
</int:channel>
<int:channel id="fileBackupChannel"/>
<int:channel id="loggerChannel"/>
</beans>

The message is now sent to persister.xml and transformer.xml. The following is persister.xml and I want to ack if persistence is successful. There are other downstream processes after transformer.xml

<int:header-enricher input-channel="fileBackupChannel" output-
channel="fileSaveChannel">
<int:header name="Operation" value="${operation.filePersister}" 
overwrite="true" />
<int:header name="replyChannel" value="nullChannel" />
<int:header name="operationStartTime" expression="#{ 
'T(java.lang.System).currentTimeMillis()' }" />
<int:error-channel ref="errorChannelId" />
</int:header-enricher>

<int-file:outbound-gateway id="fileBackUpChannelAdapter" 
directory="${file.location}"
request-channel="fileSaveChannel" reply-channel="rabbitAckChannel"/>

<int:service-activator input-channel="rabbitAckChannel" output-
channel="nullChannel" ref="ackRabbit" method="handleRabbitAcks" />

<bean id="ackRabbit" 
class="com.expedia.dataloader.rabbit.RabbitAcknowledgement"/>

<int:channel id="rabbitAckChannel">
<int:interceptors>
<int:wire-tap channel="loggerChannel" />
</int:interceptors>
</int:channel>
<int:channel id="loggerChannel"/>
<int:channel id="fileSaveChannel"/>
</beans>

I'm having trouble manually acking payloads from rabbitmq.

This is my work flow:

1. Get message from rabbit using inbound-channel-adapter:

<int-amqp:inbound-channel-adapter id="rabbitMQInboundChannelAdapter" 
channel="rabbitInboundMessageChannel"
concurrent-consumers="${rabbit.concurrentConsumers}" task-
executor="rabbit-executor" connection-
factory="amqpConnectionFactoryInbound"
message-converter="byteArrayToStringConverter" queue-
names="${rabbit.queue}" acknowledge-mode="MANUAL" error-
channel="errorChannelId"
prefetch-count="${rabbit.prefetchCount}" />

2. Persist message to disk using outbound-gateway:

<int-file:outbound-gateway id="fileBackUpChannelAdapter" 
directory="${file.location}"
request-channel="fileSaveChannel" reply-channel="loggerChannel" />

3. ack from rabbit when persister (step 2) succeeds.

for step (3), i wrote the following code:

public class RabbitAcknowledgement {
public void handleRabbitAcks(Message<?> message) throws IOException {
com.rabbitmq.client.Channel channel = (Channel) 
message.getHeaders().get("amqp_channel");
long deliveryTag = (long) message.getHeaders().get("amqp_deliveryTag");
channel.basicAck(deliveryTag, false);
}

which I'm calling from spring via:

<int:service-activator input-
channel="rabbitOutboundboundMessageChannel" output-
channel="routerChannel" ref="ackRabbit" method="handleRabbitAcks" />

This doesn't work and the the rabbit payloads in my queue are not acked.

My questions are:

Do I need MANUAL ack in this scenario?
What am I doing wrong?

The payloads remain on rabbit and are not consumed by the consumers. — Praveen Kumar
Also, in my current setup, I believe I ack 1 message per thread. Can I somehow increase the rate (I.e acknowledge in batches?) not critical to feature but was wondering if I make the system more efficient somehow — Praveen Kumar
You generally don't need manual acks; with AUTO, the container will ack the message if the persistence is successful and nack it if an exception is thrown. It's unusual to need to use manual acks for a simple scenario like this. However, it should work. Are you running the ack on the container thread? Or are you handing off to another thread using a queue or executor channel? Can you edit the question to show your complete configuration (including channels). Also turn on DEBUG logging to see if it provides any more information. — Gary Russell
Thanks Gary. How would AUTO work if there are other downstream processes running on the same thread? For eg. after persistence, i want to transform the payload and write to solr. currently the ACK happens only when both transformation and solr succeeds causing a lot of load on the rabbit server. All these processes have their own configs (rabbit.xml, persist.xml, transform.xml, solr.xml etc.). — Praveen Kumar

Gary Russell Gary Russell · Accepted Answer · 2017-06-21T15:22:43

It should work fine; I just ran a quick test and it works for me...

@SpringBootApplication
public class So44666444Application implements CommandLineRunner {

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

    @Autowired
    private RabbitTemplate template;

    private final CountDownLatch latch = new CountDownLatch(1);

    @Override
    public void run(String... args) throws Exception {
        this.template.convertAndSend("foo", "bar");
        latch.await();
    }

    @Bean
    public AmqpInboundChannelAdapter adapter(ConnectionFactory cf) {
        AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(listenerContainer(cf));
        adapter.setOutputChannelName("ack");
        return adapter;
    }

    @Bean
    public AbstractMessageListenerContainer listenerContainer(ConnectionFactory cf) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(cf);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setQueueNames("foo");
        return container;
    }

    @ServiceActivator(inputChannel = "ack")
    public void ack(@Header(AmqpHeaders.CHANNEL) Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) Long tag)
            throws IOException {
        System.out.println("Acking: " + tag);
        channel.basicAck(tag, false);
        latch.countDown();
    }

}

If I set a breakpoint on the basicAck, I see the message as unacked on the console; stepping over to the next line and the message is removed.

How to manually ack messages from rabbitmq using spring integration

1 Answers