Infinite messages resending loop after reconnecting to RabbitMQ

[sgrigorev]

Description
I have several thousands of mosquitto clients that are connected to EMQX. EMQX is configured as a bridge and it forwards messages to RabbitMQ. The issue happens when RabbitMQ server is restarted. RabbitMQ management UI displays very high incoming messages rate and there are a lot of error messages in EMQX logs:

2020-08-10 14:50:56.174 [error] [Bridge] Can't be found from the inflight:1836
2020-08-10 14:50:56.174 [error] [Bridge] Can't be found from the inflight:1837
2020-08-10 14:50:56.174 [error] [Bridge] Can't be found from the inflight:1838
2020-08-10 14:50:56.174 [error] [Bridge] Can't be found from the inflight:1839
2020-08-10 14:50:56.174 [error] [Bridge] Can't be found from the inflight:1840
2020-08-10 14:50:56.174 [error] [Bridge] Can't be found from the inflight:1841
2020-08-10 14:50:56.174 [error] [Bridge] Can't be found from the inflight:1842
2020-08-10 14:50:56.175 [error] [Bridge] Can't be found from the inflight:1843

I checked the incoming messages and there were a lot of duplicates. It seemed like EMQX was trying to resend all messages from memory.
Note: the issue also may happen when all mosquitto clients connect to EMQX (I first noticed the problem in this case).

Steps to reproduce
I implemented infrastructure to reproduce the issue locally inside docker containers. You may find it here (readme contains reproduction steps) https://github.com/sgrigorev/emqx-stress-test.

The main steps are following:

1. Configure EMQX to connect to RabbitMQ and start it
2. Start several mosquitto instances that will periodically send messages to EMQX. I think the issue may be reproduced even with 1 connected mosquitto.
3. Restart RabbitMQ
4. Check EMQX logs

Environment
EMQX v4.1.2
RabbitMQ 3.8.5
Mosquitto 1.6.9

[louisburton]

Believe this is a duplicate of #81 scheduled for v4.2.0