[Bug] Synchronous send might stuck or crash when sending many messages

[BewareMyPower]

Search before asking

• I searched in the issues and found nothing similar.

Version

• OS: macOS 13.4
• Pulsar: 3.0.0
• Client: 3.2.0 or the latest commit (804f87b)

Minimal reproduce step

#include <pulsar/Client.h>
using namespace pulsar;

int main(int argc, char* argv[]) {
  Client client("pulsar://localhost:6650");

  Producer producer;
  Result result =
      client.createProducer("persistent://public/default/my-topic", producer);
  if (result != ResultOk) {
    return -1;
  }

  // Send synchronously
  for (int i = 0; i < 10000; i++) {
    Message msg = MessageBuilder().setContent("content").build();
    Result res = producer.send(msg);
    if (res != ResultOk) {
      std::cerr << "Failed to send " << i << res << std::endl;
      break;
    }
    // NOTE: It might stuck, if we logged for each message, it would work well.
    if (i % 100 == 99) {
      std::cout << std::time(nullptr) << " " << i << std::endl;
    }
  }

  client.close();
  return 0;
}

Use C++ 17. You can use this project to reproduce.

git clone https://github.com/bewaremypower/pulsar-client-cpp-demo.git
cd pulsar-client-cpp-demo
cmake -B build
cmake --build build
./build/send_example.out

What did you expect to see?

All messages were sent successfully.

What did you see instead?

It stuck after sending 900 messages.

1688385630 99
1688385630 199
1688385630 299
1688385630 399
1688385631 499
1688385631 599
1688385631 699
1688385631 799
1688385631 899

Then I used lldb to attach this process and print c to continue running.

1688385892 1099
1688385892 1199
...
1688385901 5999
1688385901 6099
libc++abi: terminating due to uncaught exception of type std::__1::system_error: condition_variable wait failed: Invalid argument

It continues sending 5000 messages and crashed.

Anything else?

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!

[BewareMyPower]

Asynchronous send works well.

#include <pulsar/Client.h>

#include <atomic>
#include <chrono>
#include <thread>
using namespace pulsar;

int main(int argc, char* argv[]) {
  const int n = (argc > 1) ? std::stoi(argv[1]) : 10000;
  Client client("pulsar://localhost:6650");

  Producer producer;
  ProducerConfiguration conf;
  conf.setBlockIfQueueFull(true);
  conf.setBatchingEnabled(false);
  Result result = client.createProducer("persistent://public/default/my-topic",
                                        conf, producer);
  if (result != ResultOk) {
    return -1;
  }

  // Send synchronously
  std::atomic_int num_sent{0};
  for (int i = 0; i < n; i++) {
    Message msg = MessageBuilder().setContent("content").build();
    producer.sendAsync(msg,
                       [i, &num_sent](Result result, const MessageId& msgId) {
                         std::cout << i << " " << msgId << std::endl;
                         num_sent++;
                       });
  }

  while (num_sent < n) {
    std::this_thread::sleep_for(std::chrono::milliseconds(1));
  }

  client.close();
  return 0;
}