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

Abierto
  • ¿Algo no está bien?
#619 0 comentarios 0 reacciones 0 asignados Ver en GitHub

Nadie ha tomado este issue todavía.

Evaluación

Dificultad
4/5
Tiempo estimado
3-5 días
Aptitud para principiantes
48/100
Tipo de issue
Error
Claridad
Bien especificado
Estado de actividad
Activo
Stack tecnológico
cpp
Área
backend, distributed-systems

Línea de trabajo

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.

Escrito por el modelo de indexación a partir del texto del issue.

Descripción

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.

Lenguaje dominante
C++
Estrellas
72
Forks
93
Sin PR fusionados en 30 d

Preparar el entorno

Primeros pasos

  1. Lee el issue completo y luego la guía de contribución del proyecto.
  2. Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
  3. Haz un fork del repositorio y trabaja en una rama.
  4. Abre un pull request que haga referencia al número del issue.

Más de apache/pulsar-client-cpp

Todos los issues de apache/pulsar-client-cpp

Issues similares

Más issues de C++

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.