Bug 1578967

The fix for BZ #1546236 misses an edge case.

When Qpid is in FLOW_STOPPED state (triggering the errors in step 13 above), once messages are drained and the Qpid state changes back to normal, a new connection is created when a connection already exists. When the Qpid connection state changes to flow_stopped, candlepin does not need to close the connection as messages will again flow when the connection goes back to normal. However, there is a bug in the connection change listener that will re-create the connection when going from flow_stopped to connected that leaves a connection open.


Steps To Reproduce
==================

1. Deploy candlepin with Qpid support.
deploy -gtaq

2. Stop Tomcat once deployed
sudo systemctl stop tomat

3. Disable candlepin's suspend mode /etc/candlepin/candlepin.conf
candlepin.suspend_mode_enabled=false

4. Disable encryption/SSL requirement for Qpidd so we can use the 'drain' tool
require-encryption=no
ssl-require-client-authentication=no

$ sudo systemctl restart qpidd

5. Re-configure the event queue and exchanges
# Delete the event queue and create a new one with configuration to enable flow stop (max-queue-size=16384).
# NOTE:  Run from the base candlepin checkout dir.
sudo qpid-config --ssl-certificate=server/bin/qpid/keys/candlepin.crt --ssl-key server/bin/qpid/keys/candlepin.key -b amqps://localhost:5671 del queue allmsg --force
sudo qpid-config --ssl-certificate=server/bin/qpid/keys/candlepin.crt --ssl-key server/bin/qpid/keys/candlepin.key -b amqps://localhost:5671 add queue allmsg --durable --max-queue-size=16384
sudo qpid-config --ssl-certificate=server/bin/qpid/keys/candlepin.crt --ssl-key server/bin/qpid/keys/candlepin.key -b amqps://localhost:5671 bind event allmsg '#'

6. Clean out any existing hornetQ data
$ sudo rm -rf /var/lib/candlepin/activemq-artemis/*

7. Restart candlepin
sudo systemctl restart tomcat

8. Register a system to this candlepin (I used candlepin's test data)
sudo subscription-manager register -u admin -p admin --org admin

9. List the available subscriptions for this system, and make note of your favorite pool ID
$ sudo subscription-manager list --avail --all

10. List the details of the event queue. You should see that after registration a couple of events were put on the 'allmsg' queue.
$ sudo qpid-stat --ssl-certificate=server/bin/qpid/keys/candlepin.crt --ssl-key server/bin/qpid/keys/candlepin.key -b amqps://localhost:5671 -q allmsg

11. In a separate terminal, tail the candlepin log so that it can be monitored.
$ tail -f /var/log/candlepin/candlepin.log

12. Trigger an endless attach/unattach pattern for your registed system. Sub in the pool_id from above.
# while true; do subscription-manager remove --all; subscription-manager attach --pool=YOUR_FAV_POOL_ID; date; sleep 1; done

13. Watch the log file until you start seeing the following in the log. This can take a few minutes to fill the Qpid queue to its configured max.

2018-05-16 11:27:29,575 [thread=pool-4-thread-1] [=, org=, csid=] INFO  org.candlepin.audit.QpidQmf - Exchange 'event' is flow stopped because of queue org.apache.qpid.broker:queue:allmsg

14. Check to make sure that Artemis' queue starts to store the failed messages. You should see the message counts start to rise.
$ curl -k -u admin:admin https://localhost:8443/candlepin/admin/queues

16. Let the errors continue to occur until you stockpile around 100 or so messages in Artemis. Use 14 to check the count.

17. At this point you can stop the loop from 12.

18. Check the number of connections that are currently open.

$ sudo netstat -anp | grep qpidd | grep -c ESTAB

19. Drain some of the messages from Qpid so that more get sent from Artemis. Run this a couple of times with a short pause in between runs. We need to let the status monitor switch the status.

/usr/share/doc/python2-qpid/examples/api/drain -b amqps://localhost:5671 -c 5 "allmsg"

20. Check the connection counts again and each time you run drain the connection count will increase by 1.


Expected Result:

Each time we drain, the connection count should not increase.