[Messenger] Issues after migrating to AWS RabbitMQ #48241
Comments
|
Seems to be a known issue php-enqueue/enqueue-dev#1162 and the fix seems to be the heartbeat you already mentioned. |
|
@maxbeckers as I undershood heartbeats are implemented only during consumption, not publishing in amqp php extension:
|
|
We have the same issue but on the receiver not publishing. @xorik for publishing, we wrote a middleware that retries sending messages. I think this is how Symfony should solve it. For receiving, https://github.com/php-enqueue/enqueue-dev/blob/master/docs/transport/amqp_lib.md#long-running-task-and-heartbeat-and-timeouts shows how to fix it for there, but we haven't looked into how to do this for messenger. I'm not sure if a Symfony dev has any idea here? |
|
Hello there. Did anyone found a solution to this problem ? I have tried manual reconnect after |
|
Same issue using Symfony 5.4 |
|
I have managed to solve this problem with Amqproxy . Now i can publish message after idle timeout (350 sec). As i understand amqpoxy is the one who deal with the AWS managed RabbitMQ now in this setup, but also allows php symfony worker to be connected seamlessly without any timeout or heartbeat. |
|
Hey, thanks for your report! |
|
Just a quick reminder to make a comment on this. If I don't hear anything I'll close this. |
|
I think this issue should not be closed automatically. |
…lishing a message. (jwage) This PR was squashed before being merged into the 6.4 branch. Discussion ---------- [Messenger] [Amqp] Handle AMQPConnectionException when publishing a message. | Q | A | ------------- | --- | Branch? | 6.4 | Bug fix? | yes | New feature? | no | Deprecations? | no | Issues | Fix #36538 Fix #48241 | License | MIT If you have a message handler that dispatches messages to another queue, you can encounter `AMQPConnectionException` with the message "Library error: a SSL error occurred" or "a socket error occurred" depending on if you are using tls or not or if you are running behind a load balancer or not. You can manually reproduce this issue by dispatching a message where the handler then dispatches another message to a different queue, then go to rabbitmq admin and close the connection manually, then dispatch another message and when the message handler goes to dispatch the other message, you will get this exception: ``` a socket error occurred #0 /vagrant/vendor/symfony/amqp-messenger/Transport/AmqpTransport.php(60): Symfony\Component\Messenger\Bridge\Amqp\Transport\AmqpSender->send() #1 /vagrant/vendor/symfony/messenger/Middleware/SendMessageMiddleware.php(62): Symfony\Component\Messenger\Bridge\Amqp\Transport\AmqpTransport->send() #2 /vagrant/vendor/symfony/messenger/Middleware/FailedMessageProcessingMiddleware.php(34): Symfony\Component\Messenger\Middleware\SendMessageMiddleware->handle() #3 /vagrant/vendor/symfony/messenger/Middleware/DispatchAfterCurrentBusMiddleware.php(61): Symfony\Component\Messenger\Middleware\FailedMessageProcessingMiddleware->handle() #4 /vagrant/vendor/symfony/messenger/Middleware/RejectRedeliveredMessageMiddleware.php(41): Symfony\Component\Messenger\Middleware\DispatchAfterCurrentBusMiddleware->handle() #5 /vagrant/vendor/symfony/messenger/Middleware/AddBusNameStampMiddleware.php(37): Symfony\Component\Messenger\Middleware\RejectRedeliveredMessageMiddleware->handle() #6 /vagrant/vendor/symfony/messenger/Middleware/TraceableMiddleware.php(40): Symfony\Component\Messenger\Middleware\AddBusNameStampMiddleware->handle() #7 /vagrant/vendor/symfony/messenger/MessageBus.php(70): Symfony\Component\Messenger\Middleware\TraceableMiddleware->handle() #8 /vagrant/vendor/symfony/messenger/TraceableMessageBus.php(38): Symfony\Component\Messenger\MessageBus->dispatch() #9 /vagrant/src/Messenger/MessageBus.php(37): Symfony\Component\Messenger\TraceableMessageBus->dispatch() #10 /vagrant/vendor/symfony/mailer/Mailer.php(66): App\Messenger\MessageBus->dispatch() #11 /vagrant/src/Mailer/Mailer.php(83): Symfony\Component\Mailer\Mailer->send() #12 /vagrant/src/Mailer/Mailer.php(96): App\Mailer\Mailer->send() #13 /vagrant/src/MessageHandler/Trading/StrategySubscriptionMessageHandler.php(118): App\Mailer\Mailer->sendEmail() #14 /vagrant/src/MessageHandler/Trading/StrategySubscriptionMessageHandler.php(72): App\MessageHandler\Trading\StrategySubscriptionMessageHandler->handle() #15 /vagrant/vendor/symfony/messenger/Middleware/HandleMessageMiddleware.php(152): App\MessageHandler\Trading\StrategySubscriptionMessageHandler->__invoke() #16 /vagrant/vendor/symfony/messenger/Middleware/HandleMessageMiddleware.php(91): Symfony\Component\Messenger\Middleware\HandleMessageMiddleware->callHandler() #17 /vagrant/vendor/symfony/messenger/Middleware/SendMessageMiddleware.php(71): Symfony\Component\Messenger\Middleware\HandleMessageMiddleware->handle() #18 /vagrant/vendor/symfony/messenger/Middleware/FailedMessageProcessingMiddleware.php(34): Symfony\Component\Messenger\Middleware\SendMessageMiddleware->handle() #19 /vagrant/vendor/symfony/messenger/Middleware/DispatchAfterCurrentBusMiddleware.php(68): Symfony\Component\Messenger\Middleware\FailedMessageProcessingMiddleware->handle() #20 /vagrant/vendor/symfony/messenger/Middleware/RejectRedeliveredMessageMiddleware.php(41): Symfony\Component\Messenger\Middleware\DispatchAfterCurrentBusMiddleware->handle() #21 /vagrant/vendor/symfony/messenger/Middleware/AddBusNameStampMiddleware.php(37): Symfony\Component\Messenger\Middleware\RejectRedeliveredMessageMiddleware->handle() #22 /vagrant/vendor/symfony/messenger/Middleware/TraceableMiddleware.php(40): Symfony\Component\Messenger\Middleware\AddBusNameStampMiddleware->handle() #23 /vagrant/vendor/symfony/messenger/MessageBus.php(70): Symfony\Component\Messenger\Middleware\TraceableMiddleware->handle() #24 /vagrant/vendor/symfony/messenger/TraceableMessageBus.php(38): Symfony\Component\Messenger\MessageBus->dispatch() #25 /vagrant/vendor/symfony/messenger/RoutableMessageBus.php(54): Symfony\Component\Messenger\TraceableMessageBus->dispatch() #26 /vagrant/vendor/symfony/messenger/Worker.php(162): Symfony\Component\Messenger\RoutableMessageBus->dispatch() #27 /vagrant/vendor/symfony/messenger/Worker.php(109): Symfony\Component\Messenger\Worker->handleMessage() #28 /vagrant/vendor/symfony/messenger/Command/ConsumeMessagesCommand.php(238): Symfony\Component\Messenger\Worker->run() #29 /vagrant/vendor/symfony/console/Command/Command.php(326): Symfony\Component\Messenger\Command\ConsumeMessagesCommand->execute() #30 /vagrant/vendor/symfony/console/Application.php(1096): Symfony\Component\Console\Command\Command->run() #31 /vagrant/vendor/symfony/framework-bundle/Console/Application.php(126): Symfony\Component\Console\Application->doRunCommand() #32 /vagrant/vendor/symfony/console/Application.php(324): Symfony\Bundle\FrameworkBundle\Console\Application->doRunCommand() #33 /vagrant/vendor/symfony/framework-bundle/Console/Application.php(80): Symfony\Component\Console\Application->doRun() #34 /vagrant/vendor/symfony/console/Application.php(175): Symfony\Bundle\FrameworkBundle\Console\Application->doRun() #35 /vagrant/vendor/symfony/runtime/Runner/Symfony/ConsoleApplicationRunner.php(49): Symfony\Component\Console\Application->run() #36 /vagrant/vendor/autoload_runtime.php(29): Symfony\Component\Runtime\Runner\Symfony\ConsoleApplicationRunner->run() #37 /vagrant/bin/console(11): require_once('...') #38 {main} ``` TODO: - [x] Add test for retry logic when publishing messages Commits ------- f123370 [Messenger] [Amqp] Handle AMQPConnectionException when publishing a message.
|
Wohoo. So @jwage made a fix for this. Please try it out and report issues/success. |
|
@Nyholm I think catching AMQPConnectionException in https://github.com/symfony/symfony/pull/54167/files#diff-e73cdfa592b9e8cd24170c11468e4a3c26d69832c3e7d1322db693ba171df859R585 isn't enough. Please have a look at php-amqp/php-amqp#571 this SSL socket error is being reported as more generic AMQPException. Could the try-catch block be extended to cover also this case? Not catching all AMQP exceptions of course, but only those recoverable, like the socket/ssl error :) |
|
Hello, I face the same problem but while receiving messages i think when it come to acknowledge a message. Is the issue still existing in symfony 6.4? is there any workaround or solution. In Connection.php line 432: [AMQPException] Exception trace: |
|
@alikira Yes it can happen on ackowledging a message if previously acked message(s) took too long to process (especially if you are using AWS, where MQ connection timeout is hardcoded to 350 seconds). The solution is to reconnect with the AMQP broker (by using reconnect() or preconnect() from amqp lib). This is partially done in #54167 but needs to be extended to not only catch |
Symfony version(s) affected
6.0
Description
We are switching from normal RabbitMQ server, to AWS-managed cluster.
After that periodically we have this error in log:
It happens on line
$exchange->publish(...)in\Symfony\Component\Messenger\Bridge\Amqp\Transport\Connection::publishOnExchangeThe problem is that after some time, the connection becomes "dead", but method
Connection::clearWhenDisconnecteddoesn't detect is as disconnected.It happened after 350 seconds, because they have some idle timeout.
I've provided a small script, that reproduces the issue: https://gist.github.com/xorik/24065aab0db094c99e746b277466fcba
Here's output:
I can give you credentials to a test server, if you want (PM https://twitter.com/xorik_dev/ or give me your contact).
I've tried to make persistent connection, adding heartbeat, but it won't help.
This is probably connected with #36538, but suggested fix doesn't help.
How to reproduce
Possible Solution
clearWhenDisconnected)Additional Context
This is clearly not a Symfony issue, but I think Messenger should handle this issue, since this AWS RabbitMQ is quite popular, and there is no easy way to workaround it by developers, except creating a new transport and copy+pasting or extending bunch of Symfony classes.
The text was updated successfully, but these errors were encountered: