MaxListenersExceededWarning: Possible EventEmitter memory leak detected.

[dogada]

I started to see in logs following warning, but it's unclear what mat cause it. Do you have any ideas?

        at Writable.writable._write (/home/node/app/packer/node_modules/mqtt/lib/client.js:302:5)
        2019-12-27 20:34:20
        at work (/home/node/app/packer/node_modules/mqtt/lib/client.js:292:12)
        2019-12-27 20:34:20
        at MqttClient._handlePacket (/home/node/app/packer/node_modules/mqtt/lib/client.js:336:12)
        2019-12-27 20:34:20
        at MqttClient._handlePublish (/home/node/app/packer/node_modules/mqtt/lib/client.js:987:12)
        2019-12-27 20:34:20
        at MqttClient.emit (events.js:189:13)
        2019-12-27 20:34:20
        at MqttClient.onMessage (/home/node/app/packer/pipes/eventsToRecords.js:68:16)
        2019-12-27 20:34:20
        at AsyncClient.publish (/home/node/app/packer/node_modules/async-mqtt/index.js:25:12)
        2019-12-27 20:34:20
        at new Promise (<anonymous>)
        2019-12-27 20:34:20
        at Promise (/home/node/app/packer/node_modules/async-mqtt/index.js:26:31)
        2019-12-27 20:34:20
        at MqttClient.publish (/home/node/app/packer/node_modules/mqtt/lib/client.js:431:12)
        2019-12-27 20:34:20
        at MqttClient._sendPacket (/home/node/app/packer/node_modules/mqtt/lib/client.js:841:7)
        2019-12-27 20:34:20
        at sendPacket (/home/node/app/packer/node_modules/mqtt/lib/client.js:40:19)
        2019-12-27 20:34:20
        at TLSSocket.once (events.js:292:8)
        2019-12-27 20:34:20
        at TLSSocket.Readable.on (_stream_readable.js:822:35)
        2019-12-27 20:34:20
        at TLSSocket.addListener (events.js:263:10)
        2019-12-27 20:34:20
        at _addListener (events.js:247:17)
        2019-12-27 20:34:20
        (node:27) MaxListenersExceededWarning: Possible EventEmitter memory leak detected. 1001 drain listeners added. Use emitter.setMaxListeners() to increase limit

[RangerMauve]

Thanks for the report!

I'm not sure what could be causing this off the top of my head.

Would you mind posting a reproducible test case?

[dogada]

There is no reproducible test case ATM. We have an application that uses 5-7 MQTT topics and sends data between then with some processing. So a node application opens 5-7 MQTT connections to a hub, receives messages using client.on('message', (topic, message) => client.publish(nextTopic, process(message))), process and send to a different topic/queue. At the end messages are saved to the Influx. It works fine until we started to process approx 300 MQTT packets/seconds moving them between above 5-7 topics. Things become very unstable.

Now we refactored application and after receiving same 300 packets per second packets we send to intermediary topics much less data (approx. 10% of incoming packets) and problem disappeared for now but its root cause is still unclear.

[RangerMauve]

Weird. I'm not sure if this'd be an issue with async-mqtt specifically or if it's MQTT.js in general. It might be a good idea to check the issues there to see if anyone has had the same problem.

I'll leave this open for now, but I don't have bandwidth to investigate without a reproducible test case. 😁

[fijiwebdesign]

MaxListenersExceededWarning I believe is a node.js events warning when you attach 10+ listeners to an event.

https://nodejs.org/api/events.html#events_emitter_setmaxlisteners_n

MQTT.js MqttClient extends events. EventEmitter
https://github.com/mqttjs/MQTT.js/blob/master/lib/client.js#L246

So a solution is to set maxListeners to infinity if you are sure there is no memory leak.

MqttClient. setMaxListeners(0)

IMO this should only be set in end-user code. Not in the library as it is there to catch possible memory leaks.

[dogada]

@fijiwebdesign I already tried to tweak listeners count a week ago and added logging like this:

function logClientState (id, client) {
  const startTime = Date.now()
  let sentPackets = 0
  let receivedPackets = 0
  client.on('packetsend', () => {
    sentPackets++
  })
  client.on('packetreceive', () => {
    receivedPackets++
  })

  return setInterval(() => {
    debug(
      'mqttClientState',
      id,
      JSON.stringify({
        listenerCount: client.listenerCount(),
        maxListeners: client.getMaxListeners(),
        listeners: client.listeners().length,
        rawListeners: client.rawListeners().length,
        sentPackets,
        receivedPackets,
        age: Math.round((Date.now() - startTime) / 1000),
        lastMessageId: client.getLastMessageId()
      })
    )
  }, 60 * 1000)
}

It properly reports number of send and received packets but listenerCount and rawListeners().length are zero all time.

ATM we refactored our data processing, added MQTT packets filtering and decreased traffic from 300 packets/second to approx 40 packets/second. With this decreased load same MQTT code without outages, so I think the problem is related somehow to the incoming data rate.

I will try to prepare isolated test case for this problem later.

[GDnsk]

Hello, first of all thank you for your amazing work.

This problem it is also happening with me.

My script is simple, i do a query on a DB and then for every record retrieved i publish it.

My script:

    let collectionObject = db.collection(config.mongodb.collection)
    let queryResult = await collectionObject.find().limit(2000).toArray()
    console.log('Mongo Collection Query');
    asyncClient.setMaxListeners(0)
    await Promise.all(queryResult.map(async (doc) => {
        let payload = `${doc.message};${doc.datetime};`
        try {
            await asyncClient.publish(doc.topic, payload)
        } catch (err) {
            console.error(err);
            process.exit(1)
        }
        finally {
            await collectionObject.deleteOne({ _id: doc._id })
        }
    }))

As you can see, i even tried using setMaxListeners(0), but it didn't work.

When i run this script it with --trace-warnings i get:

My only work around by now was limiting the query on DB.

I hope that this will help you guys to solve the problem, thank you,

[RangerMauve]

@GuilhermeDaniluski It could be that somewhere internally in the MQTT.js library it might be adding a listener when you publish to listen on when the data has been written. Since you're doing that 2000 times at once in your queryResult.map call, it'd quickly add more listeners that the stream would expect.

I don't think this is a memory leak that you should be worrying about or anything, though.

If you want to control the event listener list of internal streams, that'd probably be an issue for the MQTT.js repo that this library depends on.