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

Đang mở
  • Có gì không ổn?
#619 0 bình luận 0 reaction 0 người được giao Xem trên GitHub

Chưa có ai nhận issue này.

Đánh giá

Độ khó
4/5
Thời gian dự kiến
3-5 ngày
Mức phù hợp với người mới
48/100
Loại issue
Lỗi
Độ rõ ràng
Đặc tả rõ ràng
Mức độ hoạt động
Sôi nổi
Công nghệ
cpp
Lĩnh vực
backend, distributed-systems

Hướng nghiên cứu

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.

Do mô hình lập chỉ mục viết ra từ nội dung của issue.

Mô tả

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.

Ngôn ngữ chính
C++
Star
72
Fork
93
Không có pull request nào được merge trong 30 ngày

Chuẩn bị môi trường

Bắt đầu từ đâu

  1. Đọc hết issue, rồi đọc hướng dẫn đóng góp của dự án.
  2. Bình luận trên issue rằng bạn sẽ nhận — tránh hai người làm cùng một việc.
  3. Fork repository và làm thay đổi trên một nhánh.
  4. Mở pull request có tham chiếu số hiệu của issue.

Issue khác của apache/pulsar-client-cpp

Tất cả issue của apache/pulsar-client-cpp

Issue tương tự

Thêm issue về C++

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.