[Event Hubs] MaxListenersExceededWarning when stopping a subscription

[the-ress]

• Package Name: @azure/event-hubs
• Package Version: 5.11.4
• Operating system: Windows
• nodejs
  • version: v20.11.1
• browser
  • name/version:
• typescript
  • version:
• Is the bug related to documentation in
  • README.md
  • source code documentation
  • SDK API docs on https://docs.microsoft.com

Describe the bug
Node reports MaxListenersExceededWarning when stopping a subscription with more than 10 partitions.

To Reproduce
Steps to reproduce the behavior:

1. Install @azure/event-hubs, @azure/eventhubs-checkpointstore-blob, @azure/storage-blob, and ws
2. Save the javascript code below as repro.js
3. Run node repro.js

repro.js contents

const { EventHubConsumerClient } = require('@azure/event-hubs');
const {
  BlobCheckpointStore,
} = require('@azure/eventhubs-checkpointstore-blob');
const { ContainerClient } = require('@azure/storage-blob');
const { WebSocket } = require('ws');

const connectionString = '[REDACTED]';
const checkpointStorageConnectionString = '[REDACTED]';
const consumerGroup = '[REDACTED]';
const checkpointStorageContainerName = '[REDACTED]';

const containerClient = new ContainerClient(
  checkpointStorageConnectionString,
  checkpointStorageContainerName,
);
const checkpointStore = new BlobCheckpointStore(containerClient);

const client = new EventHubConsumerClient(
  consumerGroup,
  connectionString,
  checkpointStore,
  {
    webSocketOptions: {
      webSocket: WebSocket,
    },
    loadBalancingOptions: {
      strategy: 'greedy',
      partitionOwnershipExpirationIntervalInMs: 5000,
      updateIntervalInMs: 1000,
    },
  },
);

const subscription = client.subscribe(
  {
    processInitialize: async (context) => {
      console.log(`Starting to read from partition ${context.partitionId}`);
    },
    processEvents: async (events, context) => {
      console.log(
        `Received ${events.length} for partition ${context.partitionId}`,
      );
      await context.updateCheckpoint(events[events.length - 1]);
    },
    processError: async (err, context) => {
      console.log(
        `Error when processing partition ${context.partitionId}:`,
        err,
      );
    },
    processClose: async (reason, context) => {
      console.log(
        `Stopped reading from partition ${context.partitionId}, reason: ${reason}`,
      );
    },
  },
  {
    maxBatchSize: 10,
    skipParsingBodyAsJson: true,
  },
);

setTimeout(() => {
  console.log('Stopping...')
  subscription
    .close()
    .then(() => client.close())
    .catch(console.log);
}, 10000);

Expected behavior
No node warnings.

Actual behavior
Node reports a warning:

MaxListenersExceededWarning: Possible EventEmitter memory leak detected. 11 disconnected listeners added to [Connection]. Use emitter.setMaxListeners() to increase limit

Screenshots

Console output

❯ node --trace-warnings repro.js
Starting to read from partition 0
Starting to read from partition 1
Starting to read from partition 2
Starting to read from partition 3
Starting to read from partition 4
Starting to read from partition 5
Starting to read from partition 6
Starting to read from partition 7
Starting to read from partition 8
Starting to read from partition 9
Starting to read from partition 10
Starting to read from partition 11
Starting to read from partition 12
Starting to read from partition 13
Starting to read from partition 14
Starting to read from partition 15
Starting to read from partition 16
Starting to read from partition 17
Starting to read from partition 18
Starting to read from partition 19
Starting to read from partition 20
Stopping...
(node:23636) MaxListenersExceededWarning: Possible EventEmitter memory leak detected. 11 disconnected listeners added to [Connection]. Use emitter.setMaxListeners() to increase limit
    at _addListener (node:events:592:17)
    at Connection.addListener (node:events:610:10)
    at Connection.once (node:events:654:8)
    at C:\Code\azure-max-listeners-repro\node_modules\rhea-promise\dist\lib\link.js:247:43
    at new Promise (<anonymous>)
    at Receiver.<anonymous> (C:\Code\azure-max-listeners-repro\node_modules\rhea-promise\dist\lib\link.js:191:34)
    at Generator.next (<anonymous>)
    at C:\Code\azure-max-listeners-repro\node_modules\tslib\tslib.js:169:75
    at new Promise (<anonymous>)
Stopped reading from partition 19, reason: Shutdown
Stopped reading from partition 20, reason: Shutdown
Stopped reading from partition 0, reason: Shutdown
Stopped reading from partition 1, reason: Shutdown
Stopped reading from partition 2, reason: Shutdown
Stopped reading from partition 3, reason: Shutdown
Stopped reading from partition 4, reason: Shutdown
Stopped reading from partition 5, reason: Shutdown
Stopped reading from partition 6, reason: Shutdown
Stopped reading from partition 7, reason: Shutdown
Stopped reading from partition 8, reason: Shutdown
Stopped reading from partition 9, reason: Shutdown
Stopped reading from partition 10, reason: Shutdown
Stopped reading from partition 11, reason: Shutdown
Stopped reading from partition 12, reason: Shutdown
Stopped reading from partition 13, reason: Shutdown
Stopped reading from partition 14, reason: Shutdown
Stopped reading from partition 15, reason: Shutdown
Stopped reading from partition 16, reason: Shutdown
Stopped reading from partition 17, reason: Shutdown
Stopped reading from partition 18, reason: Shutdown

Additional context

The warning stems from rhea-promise subscribing to the disconnected event for every partition at the same time: https://github.com/amqp/rhea-promise/blob/e37362a3df7ce9661ff6b7f20980a093596963b1/lib/link.ts#L296

Related PR (in draft since 2021):
https://github.com/amqp/rhea-promise/pull/78/files

[github-actions]

Thanks for the feedback! We are routing this to the appropriate team for follow-up. cc @kasun04.

[deyaaeldeen]

Hi @the-ress,

Thanks for opening this report! I did some debugging today and I traced the issue to the rhea library. The problem is that the Connection event emitter installs one listener per partition/topic and currently Event Hubs has a limit of up to 1024 partitions and Service Bus has a limit of up to 10k topics. I created this commit raising the limit to 10k deyaaeldeen/rhea@493b34b but I would like to discuss it internally with the team first.

Now that we determined this problem is in rhea, I'll go ahead and close this issue.

[deyaaeldeen]

I opened amqp/rhea#418.