Kue: (node) warning: possible EventEmitter memory leak detected. 11 job ttl exceeded ack listeners added

[andrewtamura]

(node) warning: possible EventEmitter memory leak detected. 11 job ttl exceeded ack listeners added. Use emitter.setMaxListeners() to increase limit.
Trace
    at Queue.addListener (events.js:239:17)
    at Queue.on (/mnt/app/node_modules/kue/lib/kue.js:130:13)
    at Command.callback (/mnt/app/node_modules/kue/lib/kue.js:234:16)
    at RedisClient.return_reply (/mnt/app/node_modules/redis/index.js:664:25)
    at HiredisReplyParser.reply_parser.send_reply (/mnt/app/node_modules/redis/index.js:332:14)
    at HiredisReplyParser.execute (/mnt/app/node_modules/redis/lib/parsers/hiredis.js:30:18)
    at Socket.<anonymous> (/mnt/app/node_modules/redis/index.js:131:27)
    at emitOne (events.js:77:13)
    at Socket.emit (events.js:169:7)
    at readableAddChunk (_stream_readable.js:146:16)
    at Socket.Readable.push (_stream_readable.js:110:10)
    at TCP.onread (net.js:523:20)

Is this warning just related to the number of jobs that have TTL expiring? We can use emitter.setMaxListeners() to increase this value and prevent the warning, but I'm curious about the root cause.

[behrad]

the listener here https://github.com/Automattic/kue/blob/master/lib/kue.js#L234 should be called only once per process to check jobs ttl, this is strange. Can you send me a code to produce it?

[bartonhammond]

I have same problem:

My Redis client initialization:

  Config.redis =
 redis: {
    host: '127.0.0.1',
    port: 6379
  }
  var connection_string = Config.redis;
  redisClient = Redis.createClient(connection_string);    
...
  module.exports = redisClient;

My Kue

     import redisClient from '../database/redis';
    let queue = Kue.createQueue({
      prefix: 'jq',
      redis: {
        createClientFactory: function(){
          return redisClient; 
        }                
      }
    });    
    Kue.app.listen(3000);
    queue.watchStuckJobs(1000);        
    return queue;
  }

error:

node) warning: possible EventEmitter memory leak detected. 11 error listeners added. Use emitter.setMaxListeners() to increase limit.
Trace
    at RedisClient.addListener (events.js:239:17)
    at Object.exports.createClient (/Users/barton/projects/stargazers/openshift-stargazers/node_modules/kue/lib/redis.js:64:12)
    at Worker.getJob (/Users/barton/projects/stargazers/openshift-stargazers/node_modules/kue/lib/queue/worker.js:267:70)
    at Worker.start (/Users/barton/projects/stargazers/openshift-stargazers/node_modules/kue/lib/queue/worker.js:82:8)
    at Queue.process (/Users/barton/projects/stargazers/openshift-stargazers/node_modules/kue/lib/kue.js:332:41)
    at GetMapData.init (/Users/barton/projects/stargazers/openshift-stargazers/build/newlib/jobs/mapdata/GetMapData.js:36:18)
    at _queue (/Users/barton/projects/stargazers/openshift-stargazers/build/newlib/Queue.js:112:39)
    at Object.queue (/Users/barton/projects/stargazers/openshift-stargazers/build/newlib/Queue.js:141:12)
    at Object.<anonymous> (/Users/barton/projects/stargazers/openshift-stargazers/build/routes/jobs/handlers.js:36:43)
    at Module._compile (module.js:435:26)
Server is running: http://Bartons-MBP.centurylink.net:5000
events.js:141
      throw er; // Unhandled 'error' event
      ^

Error: ERR only (P)SUBSCRIBE / (P)UNSUBSCRIBE / QUIT allowed in this context
    at JavascriptReplyParser._parseResult (/Users/barton/projects/stargazers/openshift-stargazers/node_modules/redis/lib/parsers/javascript.js:43:16)
    at JavascriptReplyParser.try_parsing (/Users/barton/projects/stargazers/openshift-stargazers/node_modules/redis/lib/parsers/javascript.js:114:21)
    at JavascriptReplyParser.run (/Users/barton/projects/stargazers/openshift-stargazers/node_modules/redis/lib/parsers/javascript.js:135:22)
    at JavascriptReplyParser.execute (/Users/barton/projects/stargazers/openshift-stargazers/node_modules/redis/lib/parsers/javascript.js:107:10)
    at Socket.<anonymous> (/Users/barton/projects/stargazers/openshift-stargazers/node_modules/redis/index.js:131:27)
    at emitOne (events.js:77:13)
    at Socket.emit (events.js:169:7)
    at readableAddChunk (_stream_readable.js:146:16)
    at Socket.Readable.push (_stream_readable.js:110:10)
    at TCP.onread (net.js:523:20)

[behrad]

You should set createClientFactory to a function:

var connection_string = Config.redis;
  redisClient = function(){ return Redis.createClient(connection_string); }
...
  module.exports = redisClient;

[bartonhammond]

@behrad - This is confusing to me.

You want me to change the definition of the 'src/database/redis.js' to export a function? That messes everything else as the code expects an instance of redis. And the creation of the redis client depends upon the environment I'm running in.

Sorry - but I'm not understanding. Here's my redis.js

/**

var connection_string = Config.redis;
if(process.env.OPENSHIFT_REDIS_DB_HOST){
  //The redis env variables on openshift
  connection_string.host = process.env.OPENSHIFT_REDIS_DB_HOST;
  connection_string.port = process.env.OPENSHIFT_REDIS_DB_PORT;

  //connect to Redis
  redisClient = Redis.createClient(connection_string);

  //have to authenticate
  redisClient.auth( process.env.OPENSHIFT_REDIS_DB_PASSWORD);

  //https://strongloop.com/strongblog/robust-node-applications-error-handling/
  redisClient.on('error', function (er) {
    console.trace('Here I am'); 
    console.error(er.stack); 
  });

} else if (process.env.REDIS_PORT_FORWARD) {
  connection_string.port = process.env.REDIS_PORT_FORWARD;
  //connect to Redis
  redisClient = Redis.createClient(connection_string);

  //have to authenticate
  redisClient.auth(process.env.REDIS_AUTH);
} else {
  //running locally - make sure you've started redis server
  redisClient =  Redis.createClient(connection_string);
}

module.exports = redisClient;

[behrad]

Kue needs a connection factory method, not a connection instance, since it should be able to create it's needed connection instances (e.g. separate connections for BLPOP, subscriptions, and operations is needed). You can simply pass your redis options to createQueue, But if you are using a custom made module for redis client implementation (Kue supports both node_redis and ioredis), then createRedisClient option is available for you.

[bartonhammond]

Feeling real confused/stupid about now...

With my original Kue implementation, shown below, I had ECONNRESET errors for which I created issue for: https://github.com/NodeRedis/node_redis/issues/980.

This comment suggested I was using different client instances: https://github.com/NodeRedis/node_redis/issues/980#issuecomment-183055976

Are you using different client instances?

So I'm still confused - how do I get Kue and my code using the same instance?

  if(process.env.OPENSHIFT_REDIS_DB_HOST){
    let queue = Kue.createQueue({
      prefix: 'jq',
      redis: {
        port: process.env.OPENSHIFT_REDIS_DB_PORT,  
        host: process.env.OPENSHIFT_REDIS_DB_HOST,
        auth:  process.env.OPENSHIFT_REDIS_DB_PASSWORD
      }
    });
   .....
    return queue;
  } 

[andrewtamura]

I am using Kue's bundled redis driver. Here is our createQueue function. We're on 0.10.5 connecting to a Redis instance hosted by compose.io

var queue = kue.createQueue({
  prefix: 'send_queue',
  jobEvents: false,
  redis: {
    port: process.env.REDIS_PORT,
    host: process.env.REDIS_HOST,
    auth: process.env.REDIS_PASSWORD
  }
});

I haven't seen his bug since reporting it here.

[bartonhammond]

I finally got the Kue initialization working (my knickers were wrapped around my ankles...)

 if(process.env.OPENSHIFT_REDIS_DB_HOST){
    let queue = Kue.createQueue({
      prefix: 'jq',

      redis: {
        createClientFactory: function(){
          let client = Redis.createClient({
            port: process.env.OPENSHIFT_REDIS_DB_PORT,  
            host: process.env.OPENSHIFT_REDIS_DB_HOST
          });
          client.auth(process.env.OPENSHIFT_REDIS_DB_PASSWORD);
          return client;
          }
        }
    });

So running I again see with running multiple jobs concurrently:

160211/194700.354, [response], http://ex-std-node496.prod.rhcloud.com:8080: get / {} 200 (20ms) 
events.js:141
      throw er; // Unhandled 'error' event
      ^
Error: Redis connection to 56a388737628e1252200004b-bartonhammond.rhcloud.com:50681 failed - read ECONNRESET
    at exports._errnoException (util.js:874:11)
    at TCP.onread (net.js:544:26)
DEBUG: Program node build/server.js exited with code 1
DEBUG: Starting child process with 'node build/server.js'
Server is running: http://ex-std-node496.prod.rhcloud.com:8080
Trace: Here I am
    at RedisClient.<anonymous> (/var/lib/openshift/56a387302d5271a0390000e9/app-root/runtime/repo/build/database/redis.js:56:13)
    at emitOne (events.js:77:13)
    at RedisClient.emit (events.js:169:7)
    at RedisClient.on_error (/var/lib/openshift/56a387302d5271a0390000e9/app-root/runtime/repo/node_modules/redis/index.js:216:10)
    at Socket.<anonymous> (/var/lib/openshift/56a387302d5271a0390000e9/app-root/runtime/repo/node_modules/redis/index.js:135:14)
    at emitOne (events.js:77:13)
    at Socket.emit (events.js:169:7)
    at emitErrorNT (net.js:1253:8)
    at doNTCallback2 (node.js:441:9)
    at process._tickCallback (node.js:355:17)
Error: Redis connection to 56a388737628e1252200004b-bartonhammond.rhcloud.com:50681 failed - connect ECONNREFUSED 172.16.2.86:50681
    at Object.exports._errnoException (util.js:874:11)
    at exports._exceptionWithHostPort (util.js:897:20)
    at TCPConnectWrap.afterConnect [as oncomplete] (net.js:1063:14)
events.js:141
      throw er; // Unhandled 'error' event

[behrad]

you are getting connection refused from your redis server:

172.16.2.86:50681

[shrikrishnaholla]

@behrad I faced the issue mentioned in the first comment and upon debugging, it appears to be a case where worker-specific event listening logic is attached to a global queue-level event namespace

https://github.com/Automattic/kue/blob/master/lib/queue/worker.js#L67

  if (self.ttlExceededCb)
    self.queue.removeListener('job ttl exceeded', self.ttlExceededCb);

  self.ttlExceededCb = function(id) {
    if( self.job && self.job.id && self.job.id === id ) {
      self.failed( self.job, { error: true, message: 'TTL exceeded' }, fn );
      events.emit(id, 'ttl exceeded ack');
    }
  }

  /*
   listen if current job ttl received,
   so that this worker can fail current stuck job and continue,
   in case user's process callback is stuck and done is not called in time
   */
  this.queue.on( 'job ttl exceeded', self.ttlExceededCb);

even though there is a check to ensure only one copy of event listener is added, the event-listener itself is worker-scope, so the number of event listeners added to queue's job ttl exceeded event will be equal to the number of workers, which can easily go above 11. Is there some way by which whatever ttlExceededCb does on a per-worker basis can be done with just one event listener?

[sirajabbas]

I am also facing the same problem mentioned in first comment.

The problem seems to exist only when i write the second process method

queueJobs.process('email_job', 10, function (job, done) {
//process the job here

});

queueJobs.process('notification_job', 10, function (job, done) {
//process the job here

});

Is it a good practice to only have one process method???

[ghost]

I'm seeing the same issue when I have 10+ subscriptions to different job types...

Does this mean that one should have a separate queue instead of different job types?

[swordfish444]

We are experiencing a similiar issue where jobs are enqueued into two different queues (e.g. eventQueue and notificationQueue). We are using the _delay_ when inserting a job into a queue if that helps diagnose this.

Let me know if you want me to provide the log output for this and I'll do so ASAP. This is a critical error to look into as its occurring on our production servers which is processing millions of events and notifications daily.

[airbinder]

We are also experiencing similar issue. I don't know what caused this. I have tried using the most minimal setup, but the warning still showed up.

The only thing I can find is, the warning show up when using kue.process method. I have test it by delaying the code that's calling process method, I also tried commenting it out and no warning.

Whereas on my old application, with an exact setup, everything run smoothly without warning.

Sorry can't be much of help. Just want to let people know about this error.

[tony-gutierrez]

We were getting the

warning: possible EventEmitter memory leak detected. 11 job ttl exceeded ack listeners added. Use emitter.setMaxListeners() to increase limit

error as well, and realized we had set the processing limit to 15. lowering that eliminated the errors.

[Davounet]

I'm running into the same warning when the total amount of jobs (their concurrency value actually) exceeds 10.
It doesn't seem to be anything bad, only a warning, but it's quite annoying seeing this poping up.

[amitsaxena]

I am seeing the same warning as well when I set the concurrency to 20:

queue.process('email', 20, function (job, done) {
    console.log('Processing job: ', job);
    WorkerService.process(job.data, done);
  });

Kue version: 0.11.6

[noveogroup-amorgunov]

I have a same error as @amitsaxena after restarting server (by pm2):

2|WORKER   | (node:32137) MaxListenersExceededWarning: Possible EventEmitter memory leak detected. 11 job ttl exceeded listeners added. Use emitter.setMaxListeners() to increase limit

[danieltanfh95]

I find that lowering the total concurrency amount to <11 will remove the warning, which is annoying when you have multiple job processes on the same queue with different concurrency limits.

[alanosman]

@Shadowys - how do you set that total concurrency value you mention?

[danieltanfh95]

Its the sum of the currency limit set on each process. I can't set this value. So far I have three queue processes, with 8, 5 , 5 limits respectively.

[popey456963]

@Shadowys The warning isn't an issue in this use case. You're expecting over ten event listeners, so it's not a memory leak.

You can do some prototype hacks to avoid the error:

// 0 = infinite listeners
require('events').EventEmitter.prototype._maxListeners = 0;

[danieltanfh95]

@popey456963 Yes it isn't a memory leak issue. It's just an annoying warning.

[rikez]

Hello Folks,

We have faced the same issue with the maxListeners of EventEmitter.

The workaround we found was to setMaxListeners at instance level, using queue.setMaxListeners(number).

Do you guys have any suggestion?

[pcegarra]

Same error here,

[shiny]

got the same error, any advise?

[bahatron]

I'm also getting (node:38) MaxListenersExceededWarning: Possible EventEmitter memory leak detected. 11 job ttl exceeded listeners added. Use emitter.setMaxListeners() to increase limit

my Kue implementation:

const kue = require('kue')
const $raven = require('./Raven')
const $jobs = require('../config/jobs')

const queue = kue.createQueue({
    // supress job level events
    jobEvents: false,
    // config redis connection
    redis: {
        host: process.env.REDIS_HOST || '127.0.0.1',
        port: process.env.REDIS_PORT || 6379
    }
})

queue.on('error', function (err) {
    console.log('Kue error \n', err)
    $raven.captureException(err)
})

// register jobs
Object.keys($jobs).forEach(name => {
    queue.process(name, $jobs[name])
})

module.exports = queue

a job declaration example

    'supplier-notify-created': async function (job, done) {
        try {
            console.log('notifying supplier was created...')
            let { message } = job.data

            await $axios({
                method: 'POST',
                url: $api.crmSupplierWebhook(),
                data: {
                    message
                }
            })

            console.log('successfully notified supplier was created')

            done()
        } catch (err) {
            console.log('error notifying supplier created', err)
            $raven.captureException(err)
            done(err)
        }
    }

job registration

const supplierNotifyUpdated = function (req, res) {
    let { message, user, supplierUuid } = req.body

    if (!message || !user || !supplierUuid) {
        throw new BadRequestError('Missing Parameters')
    }

    $kue.create('supplier-notify-updated', { message, user, supplierUuid })
        .priority('critical')
        .attempts(3)
        .removeOnComplete(true)
        .save()

    console.log('created supplier update notification job')
    res.json('ok')
}

whole project here: https://github.com/AddToEvent/JobProcessor

Kue version: 0.11.6

I'm using docker images to launch my app

[yangkun2197]

I got the same error, any advise?????

[shiny]

@yangkun2197 queue.setMaxListeners(concurrency + 1)

[manuelxaguilar]

@behrad I faced the issue mentioned in the first comment and upon debugging, it appears to be a case where worker-specific event listening logic is attached to a global queue-level event namespace

https://github.com/Automattic/kue/blob/master/lib/queue/worker.js#L67

  if (self.ttlExceededCb)
    self.queue.removeListener('job ttl exceeded', self.ttlExceededCb);

  self.ttlExceededCb = function(id) {
    if( self.job && self.job.id && self.job.id === id ) {
      self.failed( self.job, { error: true, message: 'TTL exceeded' }, fn );
      events.emit(id, 'ttl exceeded ack');
    }
  }

  /*
   listen if current job ttl received,
   so that this worker can fail current stuck job and continue,
   in case user's process callback is stuck and done is not called in time
   */
  this.queue.on( 'job ttl exceeded', self.ttlExceededCb);

even though there is a check to ensure only one copy of event listener is added, the event-listener itself is worker-scope, so the number of event listeners added to queue's job ttl exceeded event will be equal to the number of workers, which can easily go above 11. Is there some way by which whatever ttlExceededCb does on a per-worker basis can be done with just one event listener?

@shrikrishnaholla Did you ever find how to remove the message? I just looked at this same thing and I think you're right.