[fix][client-cpp] Fix Reader segfault when messageListenerThreads=0

[zhanglistar]

Fixes #500

Motivation

When using a Reader with ClientConfiguration::setMessageListenerThreads(0) (or when the listener executor is otherwise null), the application can crash with a segmentation fault in ConsumerImpl::messageReceived(). Two root causes were identified:

1. ExecutorServiceProvider(0) – When getMessageListenerThreads() is 0, the provider is constructed with an empty executor list. In get(), the code does idx %= executors_.size(), which is undefined behavior when size() == 0 (division by zero). This can lead to wrong or null executor pointers.

2. Null listenerExecutor_ – The Reader creates its internal consumer with a null ExecutorServicePtr. The consumer then falls back to client->getListenerExecutorProvider()->get(). If that returns null (e.g. due to the above), messageReceived and executeNotifyCallback call listenerExecutor_->postWork(...) without a null check, causing a null pointer dereference (SIGSEGV at address 0x0).

This was observed in the wild with Reader usage (e.g. periodic hasMessageAvailable() and readNext()) under moderate to high load or long runtime, with stack traces pointing at pulsar::ConsumerImpl::messageReceived.

Modifications

• lib/ExecutorService.cc

  • Construct with executors_(std::max(1, nthreads)) so the executor list is never empty when nthreads == 0, avoiding undefined behavior in get().
  • In get(), add an early if (executors_.empty()) return nullptr; guard before using executors_.size().
  • Add #include <algorithm> for std::max.

• lib/ConsumerImpl.cc

  • In messageReceived(), after the closing-state check, add a check for null listenerExecutor_: log an error, call increaseAvailablePermits(cnx), and return to avoid dereferencing null.
  • In the messageListener_ block, only call listenerExecutor_->postWork(...) when listenerExecutor_ is non-null.
  • In executeNotifyCallback(), when notifying a pending async receive: if listenerExecutor_ is non-null, post the callback as before; otherwise call notifyPendingReceivedCallback(...) on the current thread so the callback is still invoked and no null dereference occurs.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• ExecutorServiceProviderTest.ZeroThreadsReturnsValidExecutor (unit test, no broker): Creates ExecutorServiceProvider(0), calls get() three times, and asserts each return is non-null. Confirms that 0-thread configuration no longer leads to undefined behavior and that a valid executor is always returned.
• ReaderTest.testReaderWithZeroMessageListenerThreads (integration test, requires Pulsar broker): Creates a Client with setMessageListenerThreads(0), creates a Reader, produces several messages, then in a loop calls hasMessageAvailable() and readNext() until all messages are consumed. Asserts the correct count and content. Verifies that Reader path with 0 listener threads does not segfault and delivers messages correctly.

Run the unit test (no broker):

./tests/pulsar-tests --gtest_filter='ExecutorServiceProviderTest.ZeroThreadsReturnsValidExecutor'

Run the Reader integration test (with broker at pulsar://localhost:6650):

./tests/pulsar-tests --gtest_filter='ReaderTest.testReaderWithZeroMessageListenerThreads'

Documentation

• doc-required

• doc-not-needed
  (Behavior change is internal and backward-compatible: 0 listener threads is now handled safely; no API or user-facing doc update required.)

• doc

• doc-complete