Error: Failed to send message: AlreadyClosed

[arijamx]

Hi there,
Why this example with 1 second timeout works fine:

const client = new Pulsar.Client({
    serviceUrl: 'pulsar://localhost:6650',
})

const producer = await client.createProducer({
    topic: 'persistent://public/default/my-topic',
})

setTimeout(() => {
    producer.send({
        data: Buffer.from('My Message'),
    })
}, 1000)

But this example with a few more seconds timeout occurs an error?

const client = new Pulsar.Client({
    serviceUrl: 'pulsar://localhost:6650',
})

const producer = await client.createProducer({
    topic: 'persistent://public/default/my-topic',
})

setTimeout(() => {
    producer.send({
        data: Buffer.from('My Message'),
    })
}, 9000)

Full error message:

You have triggered an unhandledRejection, you may have forgotten to catch a Promise rejection:
Error: Failed to send message: AlreadyClosed

[arijamx]

Is there any kind of activity here to solve this issue?
To be more specific, I run pulsar server inside Docker like this as described here:

$ docker run -it \
  -p 6650:6650 \
  -p 8080:8080 \
  -v $PWD/data:/pulsar/data \
  apachepulsar/pulsar:2.0.0-rc1-incubating \
  bin/pulsar standalone

And here is the pulsar server logs:

18:33:03.492 [pulsar-io-50-4] INFO  org.apache.pulsar.broker.service.ServerCnx - New connection from /172.17.0.1:55506
18:33:03.607 [pulsar-io-50-4] INFO  org.apache.pulsar.broker.service.ServerCnx - [/172.17.0.1:55506][persistent://public/default/my-topic] Creating producer. producerId=0
18:33:03.722 [Thread-5] INFO  org.apache.pulsar.broker.service.ServerCnx - [/172.17.0.1:55506] persistent://public/default/my-topic configured with schema false
18:33:03.728 [Thread-5] INFO  org.apache.pulsar.broker.service.ServerCnx - [/172.17.0.1:55506] Created new producer: Producer{topic=PersistentTopic{topic=persistent://public/default/my-topic}, client=/172.17.0.1:55506, producerName=standalone-8-2, producerId=0}
18:33:19.431 [pulsar-io-50-4] INFO  org.apache.pulsar.broker.service.ServerCnx - Closed connection from /172.17.0.1:55506
18:33:20.480 [pulsar-web-57-1] INFO  org.eclipse.jetty.server.RequestLog - 127.0.0.1 - - [04/Mar/2020:18:33:20 +0000] "GET /admin/v2/persistent/public/functions/coordinate/stats HTTP/1.1" 200 882 "-" "Pulsar-Java-v2.5.0" 9

I see on log 18:33:19.431 there is a connection close after a few second from producer creation. so this is not clear enough for me that this is the client behavior or pulsar server.

[alexyans]

I have the same problem. Sending a message immediately after initializing the producer works fine, so it can't be a TLS handshake issue. The connection seems to timeout after a second or so.

[alexyans]

@alijmlzd did you try rolling back to an older version? Any luck finding one?

[arijamx]

@alexyans No, but I will try that and share my result.
By the way, what version you recommend to start?

[alexyans]

I tried 2.4 of the client as it was the one recommended by Kafkaesque, but the problem was still there for me. Didn't bother going further back as this is unspecified behavior and most likely caused by the pulsar server, and I have no control of what version Kafkaesque uses. It appears that the server explicitly closes the connection for some reason.

For my use case, I worked around the problem by always creating the producer right before i emit a message. Flushing and closing after use should prevent leaks.

[arijamx]

For my use case, I worked around the problem by always creating the producer right before I emit a message. Flushing and closing after use should prevent leaks.

Oh, as a quick fix its a good idea. but what if I have about more than 1000 send messages per second? Is it ok to create a producer, send and then flush and close the producer each time?

[alexyans]

That would be hard to tell without looking at the code. We know the client uses a connection pool for the original client connection, but it's not clear to me whether the producers/consumers create separate connections or piggyback on the one client connection. I'm not exactly sure what happens when you create a producer, it could be cheaper than it looks.

But yeah, I share your frustration. Hope it gets resolved soon.

[equanz]

It seems that setTimeout callback function run after pulsar client was destructed.
I think if use setTimeout in this case, should to use Promise and wait for timeout.

Here is an example.

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

async function produceTest(timeout) {
  const client = new Pulsar.Client({
    serviceUrl: "pulsar://localhost:6650"
  });

  const producer = await client.createProducer({
    topic: "persistent://public/default/my-topic"
  });

  await new Promise((resolve, reject) => {
    setTimeout(() => {
      producer.send({
        data: Buffer.from("My Message")
      });
      resolve();
    }, timeout);
  });

  await client.close();
}

produceTest(1000);
produceTest(9000);

[alexyans]

@equanz your example handles the rejection, sure. But there shouldn't be a rejection in the first place. A producer closing after some timeout is unspecified behavior. It also drifts from the convention: a producer connection should remain open as long as you need it open. Otherwise there would be no need to explicitly close it as per the docs. If it closes itself no matter what you do, it is safe from leaks anyway.

[equanz]

As above, it seems that client has already been destructed when executing send after timeout ms. Therefore, it seems that the connection was closed by not Broker but client.

Here is the destructor implementation in C++ client(Node.js client wraps C++ client)

• https://github.com/apache/pulsar/blob/master/pulsar-client-cpp/lib/ClientImpl.cc#L128
  • https://github.com/apache/pulsar/blob/master/pulsar-client-cpp/lib/ClientImpl.cc#L541-L568

I think you need to avoid client destruction before producing.
(Therefore, in my example, await is put on setTimeout)

[sijie]

"AlreadyClosed" indicates that the client is already closed. You need to avoid client destruction.

See a producer example: http://pulsar.apache.org/docs/en/client-libraries-node/#producer-example

[yosiat]

How I avoid client destruction?

In the following example, I create a producer and produce a message every 1 second.

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',
    sendTimeoutMs: 30000,
    batchingEnabled: true,
  });

  let i =0;
  async function produceMessage() {
    i = i + 1;

    const msg = `my-message-${i}`;
    const ret = await producer.send({
      data: Buffer.from(msg),
    });
    console.log(`Sent message: ${msg}`);
    console.log(ret);

    setTimeout(produceMessage, 1000);
  }

  produceMessage();
})();

Output:

2020-03-19 14:33:49.010 INFO  ConnectionPool:85 | Created connection for pulsar://localhost:6650
2020-03-19 14:33:49.011 INFO  ClientConnection:330 | [127.0.0.1:64239 -> 127.0.0.1:6650] Connected to broker
2020-03-19 14:33:49.111 INFO  BatchMessageContainer:43 | { BatchContainer [size = 0] [batchSizeInBytes_ = 0] [maxAllowedMessageBatchSizeInBytes_ = 131072] [maxAllowedNumMessagesInBatch_ = 1000] [topicName = persistent://public/default/my-topic] [producerName_ = ] [batchSizeInBytes_ = 0] [numberOfBatchesSent = 0] [averageBatchSize = 0]} BatchMessageContainer constructed
2020-03-19 14:33:49.111 INFO  HandlerBase:53 | [persistent://public/default/my-topic, ] Getting connection from pool
2020-03-19 14:33:49.155 INFO  ProducerImpl:151 | [persistent://public/default/my-topic, ] Created producer on broker [127.0.0.1:64239 -> 127.0.0.1:6650]
Sent message: my-message-1
null
Sent message: my-message-2
null
Sent message: my-message-3
null
Sent message: my-message-4
null
Sent message: my-message-5
null
Sent message: my-message-6
null
Sent message: my-message-7
null
Sent message: my-message-8
null
2020-03-19 14:33:57.128 INFO  ClientConnection:1349 | [127.0.0.1:64239 -> 127.0.0.1:6650] Connection closed
(node:37259) UnhandledPromiseRejectionWarning: Error: Failed to send message: AlreadyClosed
(node:37259) UnhandledPromiseRejectionWarning: Unhandled promise rejection. This error originated either by throwing inside of an async function without a catch block, or by rejecting a promise which was not handled with .catch(). (rejection id: 1)
(node:37259) [DEP0018] DeprecationWarning: Unhandled promise rejections are deprecated. In the future, promise rejections that are not handled will terminate the Node.js process with a non-zero exit code.
2020-03-19 14:33:57.312 INFO  ProducerImpl:472 | Producer - [persistent://public/default/my-topic, standalone-1-71] , [batchMessageContainer = { BatchContainer [size = 0] [batchSizeInBytes_ = 0] [maxAllowedMessageBatchSizeInBytes_ = 131072] [maxAllowedNumMessagesInBatch_ = 1000] [topicName = persistent://public/default/my-topic] [producerName_ = standalone-1-71] [batchSizeInBytes_ = 0] [numberOfBatchesSent = 8] [averageBatchSize = 1]}]
2020-03-19 14:33:57.312 INFO  BatchMessageContainer:170 | [numberOfBatchesSent = 8] [averageBatchSize = 1]
2020-03-19 14:33:57.312 INFO  ClientConnection:235 | [127.0.0.1:64239 -> 127.0.0.1:6650] Destroyed connection

[alexyans]

This is a playground example though. In a real system you can't necessarily control the rate of messages produced, as they correspond to real-time events. You could send mock keep-alive messages but it's ugly and a waste of resoures.

[yosiat]

@alexyans yeah this is an example, but in real system, I want to use the same producer and publish messages every now and then. we have a system that publish a message every one minute, in this case I will need to create new producer every-time.

I think there is no ref maintaned to producer and this why nodejs decided to close the producer. I think we need to fix this.