Skip to content

Producers or consumers may be closed after the client exits out of scope #399

Description

@RobertIndie

Here is the reproduction code:

const Pulsar = require('pulsar-client');

(async () => {
  // Create a client
  const client = new Pulsar.Client({
    serviceUrl: 'pulsar://localhost:6650'
  });

  // Create a producer
  const producer = await client.createProducer({
    topic: 'persistent://public/default/my-topic',
  });
  
  const sendRecords = async () => {
    // Send a message
    await producer.send({
        data: Buffer.from("hello")
    });

    console.log("sent hello")
    setTimeout(()=>sendRecords(), 1000)
  }

  
  await sendRecords();
})();

The output would be like:

➜  node node index.js
sent hello
sent hello
sent hello
sent hello
sent hello
sent hello
sent hello
node:internal/process/promises:288
            triggerUncaughtException(err, true /* fromPromise */);
            ^

[Error: Failed to send message: AlreadyClosed]

Node.js v18.19.0

After the client exits the outer function's scope, it will eventually be garbage collected. This closes the producers and causes the AlreadyClosed issue.

A workaround is to pass the client ref to the sendRecords function:

  const sendRecords = async (client) => {
    // Send a message
    await producer.send({
        data: Buffer.from("hello")
    });

    console.log("sent hello")
    setTimeout(()=>sendRecords(client), 1000)
  }

  
  await sendRecords(client);

And it works.

A better approach is to keep a reference to the client inside the producer or consumer. This way, as long as we hold a reference to the producer or consumer, the client object will not be garbage collected.

Activity

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions