[FLINK-17204] The RMQSource and RMQSink of the RabbitMQ connector have inconsistent default value of durable ​​when declaring the queue.

• Type: Improvement
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.10.0
• Fix Version/s: 1.11.0
• Component/s: Connectors/ RabbitMQ
• Labels: pull-request-available
• Assignee: Austin Cawley-Edwards
• Reporter: ChaojianZhang
• Created: 17/Apr/20 06:12
• Updated: 09/May/20 07:05
• Resolved: 09/May/20 07:05

Description

When the RabbitMQ queue is created and the value of durable is set to true. When I use the data in the RabbitMQ queue as the source and sink it into another queue of RabbitMQ after Flink processing, the program reports an exception, the main exception information is as follows:

 

...
Caused by: com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'durable' for queue 'rabbitmq_connectors_sink' in vhost '/': received 'false' but current is 'true', class-id=50, method-id=10)Caused by: com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'durable' for queue 'rabbitmq_connectors_sink' in vhost '/': received 'false' but current is 'true', class-id=50, method-id=10) at com.rabbitmq.utility.ValueOrException.getValue(ValueOrException.java:66) at com.rabbitmq.utility.BlockingValueOrException.uninterruptibleGetValue(BlockingValueOrException.java:36) at com.rabbitmq.client.impl.AMQChannel$BlockingRpcContinuation.getReply(AMQChannel.java:443) at com.rabbitmq.client.impl.AMQChannel.privateRpc(AMQChannel.java:263) at com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc(AMQChannel.java:136) ... 15 more
...

 

I checked the source code of RMQSource and RMQSink and found that the setupQueue() method of these two classes set the durable value in queueDeclare() is inconsistent, I think they should be consistent to be reasonable.

If possible, I want to fix it.

Issue Links

• links to: GitHub Pull Request #12001