apache / apache/pulsar

Call to subscribe() with list of topics causes seg fault if Pulsar broker is started after

Open
#14,448 2 comments 0 reactions 0 assignees View on GitHub
help wanted lifecycle/stale Stale type/bug
Dominant language
Java
Stars
15.3k
Forks
3.8k
Avg merge
1d 14h
Merged PRs (30d)
160

Description

**Describe the bug**

- Only seen with list of topics (multi-topic subscription) being passed as the `topic` argument to `subscribe()`; not seen with single topic string as `topic` argument
- Only seen when Pulsar service starts up after application has already attempted repeatedly to subscribe, i.e., starting the application and calling `subscribe()` works fine with topic list arg if it all takes place after Pulsar service has already started up
- Logging shows failure to connect to broker before the broker has been started, due to `ConnectError` as expected, but then when broker appears on the network, connection succeeds to broker but there is an immediate segfault
- Using the Apache Pulsar Python client v 2.8.0
- Repeatable, happens every time

Here is the logging, note that the call to `subscribe()` is performed by the application in a loop so there are repeated attempts in the case of failure to subscribe due to `ConnectError `when the broker is absent:

Actual topic and consumer names have been redacted.

```bash
####snip

2022-02-23 12:33:26.620 INFO [140494236182336] ConnectionPool:84 | Created connection for pulsar+ssl://172.17.0.3:6651/
2022-02-23 12:33:26.621 ERROR [140494072645376] ClientConnection:422 | [ -> pulsar+ssl://172.17.0.3:6651/] Failed to establish connection: Connection refused
2022-02-23 12:33:26.621 INFO [140494072645376] ClientConnection:1446 | [ -> pulsar+ssl://172.17.0.3:6651/] Connection closed
2022-02-23 12:33:26.621 ERROR [140494072645376] MultiTopicsConsumerImpl:142 | Error Checking/Getting Partition Metadata while MultiTopics Subscribing- [Muti Topics Consumer: TopicName - persistent://topic/name/redacted-TopicsConsumerFakeName-2ebN84bd9c - Subscription - consumer_name_redacted] result: ConnectError
2022-02-23 12:33:26.621 ERROR [140494072645376] MultiTopicsConsumerImpl:91 | Failed when subscribed to topic persistent://topic/name/redacted in TopicsConsumer. Error - ConnectError
2022-02-23 12:33:26.621 ERROR [140494072645376] MultiTopicsConsumerImpl:101 | Unable to create Consumer - [Muti Topics Consumer: TopicName - persistent://topic/name/redacted-TopicsConsumerFakeName-2ebN84bd9c - Subscription - consumer_name_redacted] Error - ConnectError
2022-02-23 12:33:26.621 INFO [140494072645376] ClientConnection:261 | [ -> pulsar+ssl://172.17.0.3:6651/] Destroyed connection
2022-02-23 12:33:27.624 INFO [140494236182336] ConnectionPool:74 | Deleting stale connection from pool for pulsar+ssl://172.17.0.3:6651/ use_count: -1 @ 0
2022-02-23 12:33:27.624 INFO [140494236182336] ConnectionPool:84 | Created connection for pulsar+ssl://172.17.0.3:6651/
2022-02-23 12:33:27.625 ERROR [140494072645376] ClientConnection:422 | [ -> pulsar+ssl://172.17.0.3:6651/] Failed to establish connection: Connection refused
2022-02-23 12:33:27.625 INFO [140494072645376] ClientConnection:1446 | [ -> pulsar+ssl://172.17.0.3:6651/] Connection closed
2022-02-23 12:33:27.625 ERROR [140494072645376] MultiTopicsConsumerImpl:142 | Error Checking/Getting Partition Metadata while MultiTopics Subscribing- [Muti Topics Consumer: TopicName - persistent://topic/name/redacted-TopicsConsumerFakeName-9e31eb5eNN - Subscription - consumer_name_redacted] result: ConnectError
2022-02-23 12:33:27.625 ERROR [140494072645376] MultiTopicsConsumerImpl:91 | Failed when subscribed to topic persistent://topic/name/redacted in TopicsConsumer. Error - ConnectError
2022-02-23 12:33:27.625 ERROR [140494072645376] MultiTopicsConsumerImpl:101 | Unable to create Consumer - [Muti Topics Consumer: TopicName - persistent://topic/name/redacted-TopicsConsumerFakeName-9e31eb5eNN - Subscription - consumer_name_redacted] Error - ConnectError
2022-02-23 12:33:27.625 INFO [140494072645376] ClientConnection:261 | [ -> pulsar+ssl://172.17.0.3:6651/] Destroyed connection
2022-02-23 12:33:28.627 INFO [140494236182336] ConnectionPool:74 | Deleting stale connection from pool for pulsar+ssl://172.17.0.3:6651/ use_count: -1 @ 0
2022-02-23 12:33:28.628 INFO [140494236182336] ConnectionPool:84 | Created connection for pulsar+ssl://172.17.0.3:6651/
2022-02-23 12:33:28.628 INFO [140494072645376] ClientConnection:372 | [172.17.0.1:52782 -> 172.17.0.3:6651] Connected to broker
2022-02-23 12:33:30.525 INFO [140494072645376] HandlerBase:55 | [persistent://topic/name/redacted, consumer_name_redacted, 0] Getting connection from pool
2022-02-23 12:33:30.647 ERROR [140494072645376] ClientConnection:910 | [172.17.0.1:52782 -> 172.17.0.3:6651] Failed lookup req_id: 2 error: ServiceUnitNotReady
2022-02-23 12:33:30.647 ERROR [140494072645376] MultiTopicsConsumerImpl:91 | Failed when subscribed to topic persistent://topic/name/redacted in TopicsConsumer. Error - ConnectError
2022-02-23 12:33:30.647 ERROR [140494072645376] MultiTopicsConsumerImpl:101 | Unable to create Consumer - [Muti Topics Consumer: TopicName - persistent://topic/name/redacted-TopicsConsumerFakeName-e6a0ab6054 - Subscription - consumer_name_redacted] Error - ConnectError
2022-02-23 12:33:30.647 ERROR [140494072645376] MultiTopicsConsumerImpl:412 | Closing the consumer failed for partition - persistent://topic/name/redacted with error - AlreadyClosed
Segmentation fault (core dumped)
```

Here is the relevant segment of application code which calls subscribe

```python
def handle_message(self, consumer, message):
self._logger.info(f"Received msg from {message.topic_name()}")

def register_multitopic_listeners(self, topics_list, name, **kwargs):
consumer = self._client.subscribe(
topics_list, name,
consumer_type=pulsar.ConsumerType.Shared,
message_listener=self.handle_message, **kwargs)
return consumer
```

- The listener is a simple stub to eliminate possibility something sbusequent in the real listener application code was causing the problem.
- `register_multitopic_listeners()` is called repeatedly in a loop containing try/except until succeeds

It looks like the problem is in the underlying C++ library that the Python library serves as a wrapper for.

I ran the application in `gdb` in the same scenario and saw the following:

```bash
####snip

2022-02-23 16:27:52.747 INFO [140737190545152] ClientConnection:372 | [172.17.0.1:42206 -> 172.17.0.3:6651] Connected to broker
[New Thread 0x7fffed3fa700 (LWP 272)]
2022-02-23 16:27:55.034 INFO [140737190545152] HandlerBase:55 | [persistent://topic/name/redacted, consumer_name_redacted, 0] Getting connection from pool
2022-02-23 16:27:55.199 ERROR [140737190545152] ClientConnection:910 | [172.17.0.1:42206 -> 172.17.0.3:6651] Failed lookup req_id: 2 error: ServiceUnitNotReady
2022-02-23 16:27:55.199 ERROR [140737190545152] MultiTopicsConsumerImpl:91 | Failed when subscribed to topic persistent://topic/name/redacted in TopicsConsumer. Error - ConnectError
2022-02-23 16:27:55.199 ERROR [140737190545152] MultiTopicsConsumerImpl:101 | Unable to create Consumer - [Muti Topics Consumer: TopicName - persistent://topic/name/redacted-TopicsConsumerFakeName-ba9b7c85e1 - Subscription - consumer_name_redacted] Error - ConnectError
2022-02-23 16:27:55.199 ERROR [140737190545152] MultiTopicsConsumerImpl:412 | Closing the consumer failed for partition - persistent://topic/name/redacted with error - AlreadyClosed

Program received signal SIGSEGV, Segmentation fault.
[Switching to Thread 0x7fffee3fc700 (LWP 270)]
0x00007fffef4709ce in pulsar::MultiTopicsConsumerImpl::closeAsync(std::function) (this=0xb04948, callback=...) at /pulsar/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc:389
389 /pulsar/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc: No such file or directory.
```

and got the following backtrace from gdb:

```bash
#0 0x00007fffef4709ce in pulsar::MultiTopicsConsumerImpl::closeAsync(std::function) (this=0xb04948, callback=...) at /pulsar/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc:389
#1 0x00007fffef472970 in pulsar::MultiTopicsConsumerImpl::handleOneTopicSubscribed (this=0xb04948, result=pulsar::ResultConnectError, consumer=..., topic=..., topicsNeedCreate=...)
at /pulsar/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc:103
#2 0x00007fffef478699 in _M_call&, pulsar::Result, pulsar::Consumer const&, std::basic_string, std::allocator >&, std::shared_ptr >&> (__ptr=..., this=) at /opt/rh/devtoolset-2/root/usr/include/c++/4.8.2/functional:558
#3 operator()&, pulsar::Result, const pulsar::Consumer&, std::basic_string, std::allocator >&, std::shared_ptr >&, void> (__object=..., this=) at /opt/rh/devtoolset-2/root/usr/include/c++/4.8.2/functional:610
#4 __call (__args=..., this=) at /opt/rh/devtoolset-2/root/usr/include/c++/4.8.2/functional:1296
#5 operator() (this=) at /opt/rh/devtoolset-2/root/usr/include/c++/4.8.2/functional:1355
#6 std::_Function_handler >)> (std::shared_ptr, std::_Placeholder<1>, std::_Placeholder<2>, std::string, std::shared_ptr >)> >::_M_invoke(std::_Any_data const&, pulsar::Result, pulsar::Consumer const&) (__functor=..., __args#0=, __args#1=...) at /opt/rh/devtoolset-2/root/usr/include/c++/4.8.2/functional:2071
#7 0x00007fffef47a1ef in pulsar::Promise::setFailed (this=, result=pulsar::ResultConnectError) at /pulsar/pulsar-client-cpp/lib/Future.h:130
#8 0x00007fffef473450 in pulsar::MultiTopicsConsumerImpl::handleSingleConsumerCreated (this=0xb04948, result=pulsar::ResultConnectError, consumerImplBaseWeakPtr=..., partitionsNeedCreate=...,
topicSubResultPromise=...) at /pulsar/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc:212
#9 0x00007fffef479969 in _M_call&, pulsar::Result, std::weak_ptr const&, std::shared_ptr >&, std::shared_ptr >&> (__ptr=..., this=) at /opt/rh/devtoolset-2/root/usr/include/c++/4.8.2/functional:558
#10 operator()&, pulsar::Result, const std::weak_ptr&, std::shared_ptr >&, std::shared_ptr >&, void> (__object=..., this=) at /opt/rh/devtoolset-2/root/usr/include/c++/4.8.2/functional:610
#11 __call&, 0ul, 1ul, 2ul, 3ul, 4ul> (__args=..., this=) at /opt/rh/devtoolset-2/root/usr/include/c++/4.8.2/functional:1296
#12 operator()&, void> (this=) at /opt/rh/devtoolset-2/root/usr/include/c++/4.8.2/functional:1355
#13 std::_Function_handler const&), std::_Bind, std::shared_ptr >, std::shared_ptr >)> (std::shared_ptr, std::_Placeholder<1>, std::_Placeholder<2>, std::shared_ptr >, std::shared_ptr >)> >::_M_invoke(std::_Any_data const&, pulsar::Result, std::weak_ptr const&) (
__functor=..., __args#0=, __args#1=...) at /opt/rh/devtoolset-2/root/usr/include/c++/4.8.2/functional:2071
#14 0x00007fffef44d56f in pulsar::Promise >::setFailed (this=this@entry=0x7fffe00028b8, result=pulsar::ResultConnectError)
at /pulsar/pulsar-client-cpp/lib/Future.h:130
#15 0x00007fffef445c36 in pulsar::ConsumerImpl::connectionFailed (this=0x7fffe0001c18, result=) at /pulsar/pulsar-client-cpp/lib/ConsumerImpl.cc:200
#16 0x00007fffef4639d5 in pulsar::HandlerBase::handleNewConnection (result=pulsar::ResultConnectError, connection=..., weakHandler=...) at /pulsar/pulsar-client-cpp/lib/HandlerBase.cc:79
#17 0x00007fffef464464 in __call&, 0ul, 1ul, 2ul> (__args=..., this=)
at /opt/rh/devtoolset-2/root/usr/include/c++/4.8.2/functional:1296
#18 operator()&, void> (this=) at /opt/rh/devtoolset-2/root/usr/include/c++/4.8.2/functional:1355
#19 std::_Function_handler const&), std::_Bind, std::_Placeholder<2>, std::weak_ptr))(pulsar::Result, std::weak_ptr, std::weak_ptr)> >::_M_invoke(std::_Any_data const&, pulsar::Result, std::weak_ptr const&) (__functor=...,
__args#0=, __args#1=...) at /opt/rh/devtoolset-2/root/usr/include/c++/4.8.2/functional:2071
#20 0x00007fffef42bf4f in pulsar::Promise >::setFailed (this=this@entry=0x7fffee3fad90, result=pulsar::ResultConnectError)
at /pulsar/pulsar-client-cpp/lib/Future.h:130
#21 0x00007fffef4232b8 in pulsar::ClientImpl::handleLookup (this=, result=, data=..., promise=...) at /pulsar/pulsar-client-cpp/lib/ClientImpl.cc:416
#22 0x00007fffef4298dc in _M_call&, pulsar::Result, std::shared_ptr const&, pulsar::Promise >&> (__ptr=..., this=0x7fffe0006e00) at /opt/rh/devtoolset-2/root/usr/include/c++/4.8.2/functional:558
#23 operator()&, pulsar::Result, const std::shared_ptr&, pulsar::Promise >&, void> (
__object=..., this=0x7fffe0006e00) at /opt/rh/devtoolset-2/root/usr/include/c++/4.8.2/functional:610

####snip
```

**To Reproduce**
Steps to reproduce the behavior:
1. Start up Python application that calls subscribe() with list of topics
2. Observe that it continually tries and fails to connect to Pulsar broker as per logging above
3. Start up Pulsar service
4. After a short while, get segfault in the Pulsar library

**Expected behavior**
I'd expect to be able to call subscribe() repeatedly if the Pulsar broker is not yet started and get a ConnectError each time, then once the Pulsar broker is up, I'd expect subscribe to connect successfully after possibly a few attempts and to start receiving data sent to the subscribed topics.

**Screenshots**
If applicable, add screenshots to help explain your problem.

**Desktop (please complete the following information):**
- OS: Centos7

**Additional context**
Add any other context about the problem here.

Contributor guide

Open the contributing guide

Research direction

Reproduce the delayed-broker scenario with the Python client and a multi-topic subscribe, then inspect pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc, especially closeAsync, handleOneTopicSubscribed, and handleSingleConsumerCreated. Use the supplied gdb backtrace to trace the ConnectError and AlreadyClosed cleanup path; done means repeated failed attempts followed by broker recovery no longer cause a segmentation fault.

Written by the indexing model from the issue text.

Assessment

Tech stack
cpp, python
Domain
distributed-systems
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
42/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.