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

Offen
  • Stimmt etwas nicht?
#619 0 Kommentare 0 Reaktionen 0 zugewiesene Personen Auf GitHub ansehen

Dieses Issue hat noch niemand übernommen.

Bewertung

Schwierigkeit
4/5
Geschätzter Aufwand
3-5 Tage
Anfängerfreundlichkeit
48/100
Issue-Typ
Bug
Klarheit
Klar beschrieben
Aktivitätsstatus
Aktiv
Tech-Stack
cpp
Bereich
backend, distributed-systems

Rechercherichtung

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.

Vom Indexierungsmodell aus dem Issue-Text verfasst.

Beschreibung

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.

Vorherrschende Sprache
C++
Sterne
72
Forks
93
Keine gemergten PRs in 30 T.

Entwicklungsumgebung

Erste Schritte

  1. Lesen Sie das ganze Issue und danach den Beitragsleitfaden des Projekts.
  2. Schreiben Sie ins Issue, dass Sie es übernehmen — das erspart doppelte Arbeit.
  3. Forken Sie das Repository und arbeiten Sie in einem Branch.
  4. Öffnen Sie einen Pull Request, der die Issue-Nummer nennt.

Mehr aus apache/pulsar-client-cpp

Alle Issues in apache/pulsar-client-cpp

Ähnliche Issues

Weitere Issues zu C++

Neue Issues direkt in Ihr Postfach

Eine kurze Übersicht über anfängerfreundliche GitHub-Issues.