apache / apache/eventmesh

RabbitMQ durable error when consuming

Open
#4,588 10 comments 0 reactions 0 assignees View on GitHub
bug Stale
Dominant language
Java
Stars
1.8k
Forks
658
Avg merge
11h 29m
Merged PRs (30d)
52

Description

### Search before asking

- [X] I had searched in the [issues](https://github.com/apache/eventmesh/issues?q=is%3Aissue) and found no similar issues.

### Environment

Mac

### EventMesh version

master

### What happened

I'm the event mesh with rabbitMQ as the store. Producing messages is working fine and it nicely appears in my RabbitMq dashboard. However when I try to consume messages via Apache Event Mesh (HTTP API call), then I'm seeing this error:
`2023-11-30 13:01:35,309 ERROR [EventMesh-Rabbitmq-Consumer-1] RabbitmqConsumerHandler(RabbitmqConsumerHandler.java:71) - [RabbitmqConsumerHandler] thread run happen exception.
com.rabbitmq.client.AlreadyClosedException: channel is already closed due to channel error; protocol method: #method(reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'durable' for exchange 'eventmesh.default' in vhost '/': received 'true' but current is 'false', class-id=40, method-id=10)
at com.rabbitmq.client.impl.AMQChannel.ensureIsOpen(AMQChannel.java:258) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.AMQChannel.rpc(AMQChannel.java:341) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.AMQChannel.privateRpc(AMQChannel.java:282) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc(AMQChannel.java:141) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.ChannelN.basicGet(ChannelN.java:1154) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.recovery.AutorecoveringChannel.basicGet(AutorecoveringChannel.java:434) ~[amqp-client-5.16.0.jar:5.16.0]
at org.apache.eventmesh.storage.rabbitmq.consumer.RabbitmqConsumerHandler.run(RabbitmqConsumerHandler.java:53) [eventmesh-storage-rabbitmq-1.9.0-release.jar:1.9.0-release]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [?:1.8.0_111]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [?:1.8.0_111]
at java.lang.Thread.run(Thread.java:745) [?:1.8.0_111]`

Initially I had the RabbitMQ exchange configured as durable but I changed it to transient. However the problem is still there. Here is the config in RabbitMQ:
![Screenshot 2023-11-30 at 14 07 38](https://github.com/apache/eventmesh/assets/17197669/98871926-fcc9-4762-9dee-a56ce060f700)

And the queue:
![Screenshot 2023-11-30 at 14 07 47](https://github.com/apache/eventmesh/assets/17197669/14c1efa6-8e74-486d-a864-09a3c4e1cb68)

Anyone faced this issue before? I tried with multiple different settings but still no luck.

### How to reproduce

Configure RabbitMQ as store. Consume a message via an HTTP API call.

### Debug logs

2023-11-30 13:01:35,309 ERROR [EventMesh-Rabbitmq-Consumer-1] RabbitmqConsumerHandler(RabbitmqConsumerHandler.java:71) - [RabbitmqConsumerHandler] thread run happen exception.
com.rabbitmq.client.AlreadyClosedException: channel is already closed due to channel error; protocol method: #method(reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'durable' for exchange 'eventmesh.default' in vhost '/': received 'true' but current is 'false', class-id=40, method-id=10)
at com.rabbitmq.client.impl.AMQChannel.ensureIsOpen(AMQChannel.java:258) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.AMQChannel.rpc(AMQChannel.java:341) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.AMQChannel.privateRpc(AMQChannel.java:282) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc(AMQChannel.java:141) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.ChannelN.basicGet(ChannelN.java:1154) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.recovery.AutorecoveringChannel.basicGet(AutorecoveringChannel.java:434) ~[amqp-client-5.16.0.jar:5.16.0]
at org.apache.eventmesh.storage.rabbitmq.consumer.RabbitmqConsumerHandler.run(RabbitmqConsumerHandler.java:53) [eventmesh-storage-rabbitmq-1.9.0-release.jar:1.9.0-release]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [?:1.8.0_111]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [?:1.8.0_111]
at java.lang.Thread.run(Thread.java:745) [?:1.8.0_111]
2023-11-30 13:01:35,309 ERROR [EventMesh-Rabbitmq-Consumer-1] RabbitmqConsumerHandler(RabbitmqConsumerHandler.java:71) - [RabbitmqConsumerHandler] thread run happen exception.
com.rabbitmq.client.AlreadyClosedException: channel is already closed due to channel error; protocol method: #method(reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'durable' for exchange 'eventmesh.default' in vhost '/': received 'true' but current is 'false', class-id=40, method-id=10)
at com.rabbitmq.client.impl.AMQChannel.ensureIsOpen(AMQChannel.java:258) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.AMQChannel.rpc(AMQChannel.java:341) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.AMQChannel.privateRpc(AMQChannel.java:282) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc(AMQChannel.java:141) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.ChannelN.basicGet(ChannelN.java:1154) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.recovery.AutorecoveringChannel.basicGet(AutorecoveringChannel.java:434) ~[amqp-client-5.16.0.jar:5.16.0]
at org.apache.eventmesh.storage.rabbitmq.consumer.RabbitmqConsumerHandler.run(RabbitmqConsumerHandler.java:53) [eventmesh-storage-rabbitmq-1.9.0-release.jar:1.9.0-release]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [?:1.8.0_111]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [?:1.8.0_111]
at java.lang.Thread.run(Thread.java:745) [?:1.8.0_111]
2023-11-30 13:01:35,309 ERROR [EventMesh-Rabbitmq-Consumer-1] RabbitmqConsumerHandler(RabbitmqConsumerHandler.java:71) - [RabbitmqConsumerHandler] thread run happen exception.
com.rabbitmq.client.AlreadyClosedException: channel is already closed due to channel error; protocol method: #method(reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'durable' for exchange 'eventmesh.default' in vhost '/': received 'true' but current is 'false', class-id=40, method-id=10)
at com.rabbitmq.client.impl.AMQChannel.ensureIsOpen(AMQChannel.java:258) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.AMQChannel.rpc(AMQChannel.java:341) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.AMQChannel.privateRpc(AMQChannel.java:282) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc(AMQChannel.java:141) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.ChannelN.basicGet(ChannelN.java:1154) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.recovery.AutorecoveringChannel.basicGet(AutorecoveringChannel.java:434) ~[amqp-client-5.16.0.jar:5.16.0]
at org.apache.eventmesh.storage.rabbitmq.consumer.RabbitmqConsumerHandler.run(RabbitmqConsumerHandler.java:53) [eventmesh-storage-rabbitmq-1.9.0-release.jar:1.9.0-release]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [?:1.8.0_111]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [?:1.8.0_111]
at java.lang.Thread.run(Thread.java:745) [?:1.8.0_111]
2023-11-30 13:01:35,309 ERROR [EventMesh-Rabbitmq-Consumer-1] RabbitmqConsumerHandler(RabbitmqConsumerHandler.java:71) - [RabbitmqConsumerHandler] thread run happen exception.
com.rabbitmq.client.AlreadyClosedException: channel is already closed due to channel error; protocol method: #method(reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'durable' for exchange 'eventmesh.default' in vhost '/': received 'true' but current is 'false', class-id=40, method-id=10)
at com.rabbitmq.client.impl.AMQChannel.ensureIsOpen(AMQChannel.java:258) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.AMQChannel.rpc(AMQChannel.java:341) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.AMQChannel.privateRpc(AMQChannel.java:282) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc(AMQChannel.java:141) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.ChannelN.basicGet(ChannelN.java:1154) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.recovery.AutorecoveringChannel.basicGet(AutorecoveringChannel.java:434) ~[amqp-client-5.16.0.jar:5.16.0]
at org.apache.eventmesh.storage.rabbitmq.consumer.RabbitmqConsumerHandler.run(RabbitmqConsumerHandler.java:53) [eventmesh-storage-rabbitmq-1.9.0-release.jar:1.9.0-release]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [?:1.8.0_111]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [?:1.8.0_111]
at java.lang.Thread.run(Thread.java:745) [?:1.8.0_111]
2023-11-30 13:01:35,309 ERROR [EventMesh-Rabbitmq-Consumer-1] RabbitmqConsumerHandler(RabbitmqConsumerHandler.java:71) - [RabbitmqConsumerHandler] thread run happen exception.
com.rabbitmq.client.AlreadyClosedException: channel is already closed due to channel error; protocol method: #method(reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'durable' for exchange 'eventmesh.default' in vhost '/': received 'true' but current is 'false', class-id=40, method-id=10)
at com.rabbitmq.client.impl.AMQChannel.ensureIsOpen(AMQChannel.java:258) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.AMQChannel.rpc(AMQChannel.java:341) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.AMQChannel.privateRpc(AMQChannel.java:282) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc(AMQChannel.java:141) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.ChannelN.basicGet(ChannelN.java:1154) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.recovery.AutorecoveringChannel.basicGet(AutorecoveringChannel.java:434) ~[amqp-client-5.16.0.jar:5.16.0]
at org.apache.eventmesh.storage.rabbitmq.consumer.RabbitmqConsumerHandler.run(RabbitmqConsumerHandler.java:53) [eventmesh-storage-rabbitmq-1.9.0-release.jar:1.9.0-release]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [?:1.8.0_111]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [?:1.8.0_111]
at java.lang.Thread.run(Thread.java:745) [?:1.8.0_111]
2023-11-30 13:01:35,309 ERROR [EventMesh-Rabbitmq-Consumer-1] RabbitmqConsumerHandler(RabbitmqConsumerHandler.java:71) - [RabbitmqConsumerHandler] thread run happen exception.
com.rabbitmq.client.AlreadyClosedException: channel is already closed due to channel error; protocol method: #method(reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'durable' for exchange 'eventmesh.default' in vhost '/': received 'true' but current is 'false', class-id=40, method-id=10)
at com.rabbitmq.client.impl.AMQChannel.ensureIsOpen(AMQChannel.java:258) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.AMQChannel.rpc(AMQChannel.java:341) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.AMQChannel.privateRpc(AMQChannel.java:282) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc(AMQChannel.java:141) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.ChannelN.basicGet(ChannelN.java:1154) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.recovery.AutorecoveringChannel.basicGet(AutorecoveringChannel.java:434) ~[amqp-client-5.16.0.jar:5.16.0]
at org.apache.eventmesh.storage.rabbitmq.consumer.RabbitmqConsumerHandler.run(RabbitmqConsumerHandler.java:53) [eventmesh-storage-rabbitmq-1.9.0-release.jar:1.9.0-release]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [?:1.8.0_111]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [?:1.8.0_111]
at java.lang.Thread.run(Thread.java:745) [?:1.8.0_111]
2023-11-30 13:01:35,309 ERROR [EventMesh-Rabbitmq-Consumer-1] RabbitmqConsumerHandler(RabbitmqConsumerHandler.java:71) - [RabbitmqConsumerHandler] thread run happen exception.
com.rabbitmq.client.AlreadyClosedException: channel is already closed due to channel error; protocol method: #method(reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'durable' for exchange 'eventmesh.default' in vhost '/': received 'true' but current is 'false', class-id=40, method-id=10)
at com.rabbitmq.client.impl.AMQChannel.ensureIsOpen(AMQChannel.java:258) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.AMQChannel.rpc(AMQChannel.java:341) ~[amqp-client-5.16.0.jar:5.16.0]
at com.rabbitmq.client.impl.AMQChannel.privateRpc(AMQChannel.java:282) ~[amqp-client-5.16.0.jar:5.16.0]

### Are you willing to submit PR?

- [X] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct) *

Contributor guide

Open the contributing guide

Research direction

Start with org.apache.eventmesh.storage.rabbitmq.consumer.RabbitmqConsumerHandler.java, especially the reported lines 53 and 71, and reproduce consumption through the HTTP API with RabbitMQ configured as the store. Trace the exchange declaration and compare its durable setting with the existing eventmesh.default exchange; done means consuming no longer produces the reported PRECONDITION_FAILED error.

Written by the indexing model from the issue text.

Assessment

Tech stack
java, rabbitmq
Domain
backend, distributed-systems
Issue type
Bug
Difficulty
3/5
Estimated time
1-2 days
Activity status
Stale
Clarity
Needs clarification
Newbie friendliness
25/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.