[Bug] Segfault in BaseCommand::ByteSizeLong when closing a partitioned producer #619

Version

pulsar-client-cpp 3.4.2. The same two code paths are still present on current master:

• lib/Commands.cc: Commands::newPartitionMetadataRequest keeps a function-local static BaseCommand
• lib/PartitionedProducerImpl.cc: runPartitionUpdateTask ignores the timer error_code

Minimal reproduce step

1. Create a client with the default partitionsUpdateInterval (60 seconds).
2. createProducer on a partitioned topic (getPartitions() > 0), so the implementation is PartitionedProducerImpl and runPartitionUpdateTask is armed.
3. On process shutdown call producer.close() and then client.close().

producer.close() cancels partitionsUpdateTimer_. The wait handler still runs because it does not check ec:

partitionsUpdateTimer_->async_wait([weakSelf](const ASIO_ERROR& ec) {
    auto self = weakSelf.lock();
    if (self) {
        self->getPartitionMetadata();
    }
});

getPartitionMetadata() does not check that the producer is closing. It sends PARTITIONED_METADATA through Commands::newPartitionMetadataRequest, which serializes a function-local static protobuf message:

SharedBuffer Commands::newPartitionMetadataRequest(const std::string& topic, uint64_t requestId) {
    static BaseCommand cmd;
    static std::mutex mutex;
    std::lock_guard<std::mutex> lock(mutex);
    // ...
    SharedBuffer buffer = writeMessageWithSize(cmd);  // cmd.ByteSize() / ByteSizeLong()
}

client.close() then shuts the lookup service down and joins executor threads with a short timeout. The listener thread can still be inside ByteSizeLong() while that static BaseCommand is being destroyed. Destruction of the statics does not take mutex.

Workaround: clientConfiguration.setPartititionsUpdateInterval(0) before creating the partitioned producer. The timer is not created when the interval is 0. New partitions are then picked up only after the producer is recreated.

What did you expect to see?

producer.close() / client.close() stop the partition-update task and return without a segmentation fault.

What did you see instead?

Intermittent segfault on the Pulsar listener thread during shutdown:

#0  ??
#1  pulsar::proto::BaseCommand::ByteSizeLong() const
#2  pulsar::Commands::writeMessageWithSize(pulsar::proto::BaseCommand const&)
#3  pulsar::Commands::newPartitionMetadataRequest(...)
#4  pulsar::ClientConnection::newPartitionedMetadataLookup(...)
#5  pulsar::BinaryProtoLookupService::sendPartitionMetadataLookupRequest(...)
#6  std::_Function_handler<... BinaryProtoLookupService::sendPartitionMetadataLookupRequest ...>
#7  pulsar::BinaryProtoLookupService::getPartitionMetadataAsync(...)
#8  pulsar::RetryableLookupService::getPartitionMetadataAsync(...)::{lambda}
#9  pulsar::RetryableOperation<...>::runImpl(...)
#10 pulsar::RetryableOperation<...>::run()
#11 pulsar::RetryableOperationCache<...>::run(...)
#12 pulsar::RetryableLookupService::getPartitionMetadataAsync(...)
#13 pulsar::PartitionedProducerImpl::getPartitionMetadata()
#14 boost::asio::detail::wait_handler<PartitionedProducerImpl::runPartitionUpdateTask()::{lambda...}>::do_complete(...)
#15 boost::asio::detail::scheduler::run(...)
#16 pulsar::ExecutorService::start()::{lambda}::operator()
#17 start_thread
#0 is a jump through an invalid address inside protobuf ByteSizeLong() of that static BaseCommand.

Anything else?

Suggested fix:

1. In runPartitionUpdateTask, return when ec is set (operation_aborted).
2. In getPartitionMetadata, return unless state_ == Ready.
3. Stop using function-local static BaseCommand in Commands.cc. Use an automatic BaseCommand so shutdown cannot destroy the message while another thread is in ByteSizeLong(). The same static pattern is used by the other Commands::new* helpers.

A non-partitioned producer does not hit this path. PartitionedProducerImpl is created only when partitionMetadata->getPartitions() > 0.