I am using spring integration (with inbound/outbound channel adapters) to transfer messages from
- JMS to AMQP
- AMQP to JMS
- AMQP to AMQP
This works fine.
When the destination queue is full, I would like to stop the process, and sending back the message to the original Queue (rollback message from the channel).
This works fine for JMS => AMQP and AMQP => JMS but KO for AMQP => AMQP.
Following code working fine (JMS => AMQP)
<bean id="myListener" class="org.springframework.jms.listener.DefaultMessageListenerContainer" >
<property name="autoStartup" value="false" />
<property name="connectionFactory" ref="connectionFactoryCaching" />
<property name="destination" ref="jmsQueue" />
<property name="maxMessagesPerTask" value="1" />
<property name="receiveTimeout" value="1" />
<property name="backOff" ref="fixedBackOff" />
<property name="sessionTransacted" value="true"/>
</bean>
<int-jms:message-driven-channel-adapter id="jmsIn" container="myListener" channel="channelJMS_AMQP" error-channel="processChannel1"/>
<rabbit:template id="rabbitTemplate"
connection-factory="rabbitConnectionFactory"
mandatory="true"
channel-transacted="true"
message-converter="simpleMessageConverter"/>
<int-amqp:outbound-channel-adapter channel="channelJMS_AMQP"
routing-key="RK1"
exchange-name="EXC1"
amqp-template="rabbitTemplate"
default-delivery-mode="PERSISTENT"/>
In log file the message is well rollback:
2020-11-19 10:08:57.860 [AMQP Connection XX.XX.XX.XX.130:5672] ERROR o.s.a.r.c.CachingConnectionFactory - Channel shutdown: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - partial tx completion, class-id=90, method-id=20)
2020-11-19 10:08:57.861 [myListener-1] DEBUG o.s.a.r.c.CachingConnectionFactory - Detected closed channel on exception. Re-initializing: AMQChannel(amqp://[email protected]:5672/,1)
2020-11-19 10:08:57.950 [myListener-1] DEBUG o.s.a.r.c.RabbitResourceHolder - Rolling back messages to channel: Cached Rabbit Channel: AMQChannel(amqp://[email protected]:5672/,1), conn: Proxy@68935e42 Dedicated Rabbit Connection: SimpleConnection@577ca937 [delegate=amqp://[email protected]:5672/, localPort= 41314]
I have similar code for AMQP=> JMS that works fine.
But for AMQP => AMQP I have a problem, the message is lost and the process isn't stop:
<bean id="myListener" class="org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer">
<property name="connectionFactory" ref="rabbitConnectionFactory" />
<property name="queueNames" value="MY_QUEUE" />
<property name="recoveryBackOff" ref="fixedBackOffRabbitMQ"/>
<property name="channelTransacted" value="true"></property>
</bean>
<int-amqp:inbound-channel-adapter channel="channelAMQP_AMQP"
id="inboundChannelAdapter"
auto-startup="true" listener-container="myListener" error-channel="processChannel1" />
<rabbit:template id="rabbitTemplate"
connection-factory="rabbitConnectionFactory2"
mandatory="true"
channel-transacted="true"
message-converter="simpleMessageConverter"/>
<int-amqp:outbound-channel-adapter channel="channelAMQP_AMQP"
routing-key="RK1"
exchange-name="EXC1"
amqp-template="rabbitTemplate"
default-delivery-mode="PERSISTENT"/>
Log file:
2020-11-19 10:15:24.260 [AMQP Connection XX.XX.XX.XX.130:5672] ERROR o.s.a.r.c.CachingConnectionFactory - Channel shutdown: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - partial tx completion, class-id=90, method-id=20)
2020-11-19 10:15:24.351 [myListener-1] DEBUG o.s.a.r.c.CachingConnectionFactory - Detected closed channel on exception. Re-initializing: AMQChannel(amqp://[email protected]:5672/,1)
2020-11-19 10:15:24.356 [pool-4-thread-3] DEBUG o.s.a.r.l.BlockingQueueConsumer - Received shutdown signal for consumer tag=amq.ctag--o5d3WkrF0N2aiCHul-qiA
And then restarting a consumer, instead of stopping, and rollbacking the Message.
Then my message is lost:- (
Do you have any recommendations to have the expected operation in this particular case? Should I add a specific treatment?
