ActiveMQ Stomp Python client listener creating extraneous consumers

We're building a queuing system in Python that involves a series of microservices that subscribe to a queue. After each microservice processes a message, it creates the next message in the queue as the process moves along in the workflow.

All microservices are using the python stomp package version stomp==7.0.0. The microservices all use the same wrapper code in a custom package we've built for the purpose of connecting and subscribing.

The microservices are all running in a docker swarm, and each microservice has multiple replicas, in this case 4 replicas per microservice.

Issue:

We would expect one instance of a microservice to create one listener (consumer) only. So we would expect that one microservice with 4 replicas to have 4 consumers only.

However, what we're seeing is the number of consumers increasing and decreasing intermittently as messages are being processed in the queue.

So one microservice with 4 replicas starts with 4 consumers as expected, but then increases to 5 and then more and more as time goes on, in some cases 10 consumers or more. Also in some cases consumers are dropping off entirely, going down to 3 and then eventually down to 0 and then the entire pipeline seems to get stalled.

The extraneous consumers seem to cause performance and stability issues in the pipeline.

Notes

We set the container hostname as the client_id for each connection to track which consumers are being created on each host. The extra consumers have the same client_ids as the ones already connected. This is probably a key fact, since by all accounts that shouldn’t be possible.

Questions

• Are our expectations correct that one instance of a microservice would have one consumer only?
• What could be causing the extraneous consumers to be created?
• What could be causing consumers to be lost?

Code

_client_id = f"mqutils_listener_{os.getenv('HOSTNAME')}"

def initialize_mqlistener():
    if _failback_host and _failback_port:
        print('failback host and port in use')
        conn = stomp.Connection([(_host, _port),(_failback_host, _failback_port)], heartbeats=(40000, 40000), keepalive=True)
    else:
        print('failback host and port not in use')
        conn = stomp.Connection([(_host, _port)], heartbeats=(40000, 40000), keepalive=True)
    conn.set_listener('', MqListener(conn))
    connect_and_subscribe(conn)
    # http_clients://github.com/jasonrbriggs/stomp.py/issues/206
    while True:
        time.sleep(2)
        if not conn.is_connected():
            print('Disconnected in loop, reconnecting')
            connect_and_subscribe(conn)


def connect_and_subscribe(conn, queue=_queue, sub_id=_sub_id):
    print(_hostname_prefix + "************************ MQUTILS MQLISTENER - CONNECT_AND_SUBSCRIBE *******************************")
    global _reconnect_attempts
    _reconnect_attempts = _reconnect_attempts + 1
    if _reconnect_attempts <= _max_attempts:
        # TODO: Retry timer with exponential backoff
        time.sleep(1)
        try:
            if not os.getenv('MQ_DISABLE_SSL'):
                if _failback_host and _failback_port:
                    conn.set_ssl([(_host, _port),(_failback_host, _failback_port)])
                else:
                    conn.set_ssl([(_host, _port)])
            if not conn.is_connected():
                conn.connect(_user, _password, headers={'client-id': _client_id}, wait=True)
                print(f'{_hostname_prefix}connect_and_subscribe connecting {queue} to with connection id {sub_id} reconnect attempts: {_reconnect_attempts}', flush=True)
            else:
                print(f'{_hostname_prefix}connect_and_subscibe already connected {queue} to with connection id {sub_id} reconnect attempts {_reconnect_attempts}', flush=True)
        except Exception as e:
            print(_hostname_prefix + 'Exception on disconnect. reconnecting...')
            print(_hostname_prefix + traceback.format_exc())
            connect_and_subscribe(conn)
        else:
            conn.subscribe(destination=queue, id=sub_id, ack='client-individual')
            _reconnect_attempts = 0
    else:
        print('{}Maximum reconnect attempts reached for this connection. reconnect attempts: {}'.format(_hostname_prefix, _reconnect_attempts), flush=True)

Logs

received an error "javax.jms.InvalidClientIDException: Broker: mps-qa - Client: mqutils_listener_7b69983dd500 already connected from tcp://redacted

2022-09-07 14:44:04,030 INFO exited: queuelistener (terminated by SIGABRT; not expected)
2022-09-07 14:44:04,049 INFO spawned: 'queuelistener' with pid 6238
2022-09-07 14:44:05,618 INFO success: queuelistener entered RUNNING state, process has stayed up for > than 1 seconds (startsecs)
2022-09-07 14:44:15,334 INFO exited: queuelistener (terminated by SIGABRT; not expected)
2022-09-07 14:44:16,367 INFO spawned: 'queuelistener' with pid 6329
2022-09-07 14:44:17,941 INFO success: queuelistener entered RUNNING state, process has stayed up for > than 1 seconds (startsecs)
2022-09-07 14:44:20,143 INFO exited: queuelistener (terminated by SIGABRT; not expected)
2022-09-07 14:44:20,164 INFO spawned: 'queuelistener' with pid 6348
2022-09-07 14:44:21,734 INFO success: queuelistener entered RUNNING state, process has stayed up for > than 1 seconds (startsecs)
2022-09-07 14:44:27,280 INFO exited: queuelistener (terminated by SIGABRT; not expected)
2022-09-07 14:44:28,313 INFO spawned: 'queuelistener' with pid 6386