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

classic Classic list List threaded Threaded
1 message Options
Reply | Threaded
Open this post in threaded view
|

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

Shang Yuanchun (Jira)
ChaojianZhang created FLINK-17204:
-------------------------------------

             Summary: The RMQSource and RMQSink of the RabbitMQ connector have inconsistent default value of durable ​​when declaring the queue.
                 Key: FLINK-17204
                 URL: https://issues.apache.org/jira/browse/FLINK-17204
             Project: Flink
          Issue Type: Improvement
          Components: Connectors/ RabbitMQ
    Affects Versions: 1.10.0
            Reporter: ChaojianZhang


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:

 
{code:java}
...
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
...{code}
 

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.



--
This message was sent by Atlassian Jira
(v8.3.4#803005)