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

オープン
  • 何か問題がありますか？
#619 コメント 0 件 リアクション 0 件 担当者 0 名 GitHub で見る

まだ誰も着手していません。

評価

難易度
4/5
見積もり時間
3〜5日
初心者へのやさしさ
48/100
issue の種類
バグ
明瞭さ
明確に書かれている
活発さ
活発
技術スタック
cpp
領域
backend, distributed-systems

調査の方向性

Read lib/PartitionedProducerImpl.cc at runPartitionUpdateTask and getPartitionMetadata, then inspect lib/Commands.cc and the other Commands::new* helpers for the static BaseCommand pattern. Trace how timer cancellation and producer shutdown interact, and run the partitioned producer shutdown tests. Done when closing a partitioned producer and client no longer triggers the reported shutdown segfault.

索引モデルが issue の本文から書いたものです。

説明

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.

主要言語
C++
スター
72
フォーク
93
30日以内にマージされた PR はありません

環境構築

はじめの一歩

  1. issue を最後まで読み、次にプロジェクトのコントリビューションガイドを読みます。
  2. 着手することを issue にコメントします — 二人が同じ作業をするのを防げます。
  3. リポジトリをフォークし、ブランチを切って変更します。
  4. issue 番号を参照したプルリクエストを送ります。

apache/pulsar-client-cpp のほかの issue

apache/pulsar-client-cpp の issue をすべて見る

似ている issue

C++ の issue をもっと見る

新しい issue をメールで受け取る

初心者向けの GitHub issue を短くまとめたダイジェスト。