[Python] Flaky-test: test_listener_name_client

[BewareMyPower]

test_listener_name_client is flaky. It fails sporadically.

example failure

======================================================================
ERROR: test_listener_name_client (__main__.PulsarTest)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "pulsar_test.py", line 1190, in test_listener_name_client
    client.close()
  File "/usr/local/lib/python3.8/dist-packages/pulsar/__init__.py", line 890, in close
    self._client.close()
_pulsar.AlreadyClosed: Pulsar error: AlreadyClosed

[RobertIndie]

Another failure: https://github.com/apache/pulsar/runs/6936983274?check_suite_focus=true#step:11:1721

[BewareMyPower]

It's easy to reproduce it locally. It's a bug when processing a partitioned topic. Because test_listener_name_client reuses a partitioned topic created by test_get_partitioned_topic_name.

We can reproduce it by creating a partitioned topic in advance:

./bin/pulsar-admin topics create-partitioned-topic partitioned_topic_name_test -p 3

Then, run the following Python script:

from pulsar import Client
import pulsar

client = Client('pulsar://localhost:6650', listener_name='test')
try:
    producer = client.create_producer("persistent://public/default/partitioned_topic_name_test")
except pulsar.PulsarException as e:
    print(e)

client.close()

There is a high chance that it failed with AlreadyClosed:

2022-06-17 23:31:22.111 INFO  [0x111691600] ClientConnection:189 | [<none> -> pulsar://localhost:6650] Create ClientConnection, timeout=10000
2022-06-17 23:31:22.111 INFO  [0x111691600] ConnectionPool:96 | Created connection for pulsar://localhost:6650
2022-06-17 23:31:22.113 INFO  [0x700001194000] ClientConnection:375 | [127.0.0.1:62177 -> 127.0.0.1:6650] Connected to broker
2022-06-17 23:31:22.115 INFO  [0x700001194000] HandlerBase:64 | [persistent://public/default/partitioned_topic_name_test-partition-0, ] Getting connection from pool
2022-06-17 23:31:22.115 INFO  [0x700001194000] HandlerBase:64 | [persistent://public/default/partitioned_topic_name_test-partition-1, ] Getting connection from pool
2022-06-17 23:31:22.115 INFO  [0x700001194000] HandlerBase:64 | [persistent://public/default/partitioned_topic_name_test-partition-2, ] Getting connection from pool
2022-06-17 23:31:22.116 ERROR [0x700001194000] ClientConnection:1001 | [127.0.0.1:62177 -> 127.0.0.1:6650] Failed lookup req_id: 2 error: ServiceUnitNotReady msg: org.apache.pulsar.broker.PulsarServerException: the broker do not have test listener
2022-06-17 23:31:22.116 ERROR [0x700001194000] PartitionedProducerImpl:150 | Unable to create Producer for partition - 0 Error - ServiceUnitNotReady
2022-06-17 23:31:22.116 ERROR [0x700001194000] ClientConnection:597 | [127.0.0.1:62177 -> 127.0.0.1:6650] Read operation failed: Bad file descriptor
Pulsar error: ServiceUnitNotReady
2022-06-17 23:31:22.116 INFO  [0x700001194000] ClientConnection:1560 | [127.0.0.1:62177 -> 127.0.0.1:6650] Connection closed
2022-06-17 23:31:22.116 INFO  [0x111691600] ClientImpl:500 | Closing Pulsar client with 1 producers and 0 consumers
2022-06-17 23:31:22.116 ERROR [0x700001194000] PartitionedProducerImpl:289 | Closing the producer failed for partition - 0
2022-06-17 23:31:22.116 ERROR [0x111691600] PartitionedProducerImpl:289 | Closing the producer failed for partition - 1
2022-06-17 23:31:22.116 INFO  [0x700001194000] ProducerImpl:665 | [persistent://public/default/partitioned_topic_name_test-partition-2, ] Closing producer for topic persistent://public/default/partitioned_topic_name_test-partition-2
2022-06-17 23:31:22.116 INFO  [0x111691600] ProducerImpl:630 | Producer - [persistent://public/default/partitioned_topic_name_test-partition-2, ] , [batching  = off]
2022-06-17 23:31:22.116 INFO  [0x111691600] ProducerImpl:630 | Producer - [persistent://public/default/partitioned_topic_name_test-partition-1, ] , [batching  = off]
2022-06-17 23:31:22.116 INFO  [0x111691600] ProducerImpl:630 | Producer - [persistent://public/default/partitioned_topic_name_test-partition-0, ] , [batching  = off]
2022-06-17 23:31:22.117 INFO  [0x111691600] ClientConnection:263 | [127.0.0.1:62177 -> 127.0.0.1:6650] Destroyed connection
Traceback (most recent call last):
  File "/Users/xuyunze/github.com/BewareMyPower/pulsar/pulsar-client-cpp/python/1.py", line 10, in <module>
    client.close()
  File "/Users/xuyunze/github.com/BewareMyPower/pulsar/pulsar-client-cpp/python/pulsar/__init__.py", line 890, in close
    self._client.close()
_pulsar.AlreadyClosed: Pulsar error: AlreadyClosed

[BewareMyPower]

There is also a chance that the application stuck. I used lldb to print the stacktraces:

Thread 1:

* thread #1, queue = 'com.apple.main-thread', stop reason = signal SIGSTOP
  * frame #0: 0x00007ff811dfa3ea libsystem_kernel.dylib`__psynch_cvwait + 10
    frame #1: 0x00007ff811e34a6f libsystem_pthread.dylib`_pthread_cond_wait + 1249
    frame #2: 0x00007ff811d92d33 libc++.1.dylib`std::__1::condition_variable::__do_timed_wait(std::__1::unique_lock<std::__1::mutex>&, std::__1::chrono::time_point<std::__1::chrono::system_clock, std::__1::chrono::duration<long long, std::__1::ratio<1l, 1000000000l> > >) + 93
    frame #3: 0x0000000106cce726 _pulsar.so`std::__1::cv_status std::__1::condition_variable::wait_for<long long, std::__1::ratio<1l, 1000000000l> >(this=0x0000600003500058, __lk=0x00007ff7bfefe420, __d=0x00007ff7bfefe2c0) at __mutex_base:463:9
    frame #4: 0x0000000106cce37c _pulsar.so`void std::__1::condition_variable::__do_timed_wait<std::__1::chrono::steady_clock>(this=0x0000600003500058, __lk=0x00007ff7bfefe420, __tp=time_point<std::__1::chrono::steady_clock, std::__1::chrono::duration<long long, std::__1::ratio<1, 1000000000> > > @ 0x00007ff7bfefe2d8) at __mutex_base:518:5
    frame #5: 0x0000000106cce11b _pulsar.so`std::__1::cv_status std::__1::condition_variable::wait_until<std::__1::chrono::steady_clock, std::__1::chrono::duration<long long, std::__1::ratio<1l, 1000000000l> > >(this=0x0000600003500058, __lk=0x00007ff7bfefe420, __t=0x00007ff7bfefe3c8) at __mutex_base:421:5
    frame #6: 0x0000000106dd32d9 _pulsar.so`bool std::__1::condition_variable::wait_until<std::__1::chrono::steady_clock, std::__1::chrono::duration<long long, std::__1::ratio<1l, 1000000000l> >, bool pulsar::Future<bool, pulsar::Result>::get<std::__1::chrono::duration<long long, std::__1::ratio<1l, 1000l> > >(bool&, pulsar::Result&, std::__1::chrono::duration<long long, std::__1::ratio<1l, 1000l> >)::'lambda'()>(this=0x0000600003500058, __lk=0x00007ff7bfefe420, __t=0x00007ff7bfefe3c8, __pred=(unnamed class) @ 0x00007ff7bfefe380)::'lambda'()) at __mutex_base:433:13
    frame #7: 0x0000000106dd3277 _pulsar.so`bool std::__1::condition_variable::wait_for<long long, std::__1::ratio<1l, 1000l>, bool pulsar::Future<bool, pulsar::Result>::get<std::__1::chrono::duration<long long, std::__1::ratio<1l, 1000l> > >(bool&, pulsar::Result&, std::__1::chrono::duration<long long, std::__1::ratio<1l, 1000l> >)::'lambda'()>(this=0x0000600003500058, __lk=0x00007ff7bfefe420, __d=0x00007ff7bfefe450, __pred=(unnamed class) @ 0x00007ff7bfefe3e8)::'lambda'()) at __mutex_base:477:12
    frame #8: 0x0000000106dcf4cc _pulsar.so`bool pulsar::Future<bool, pulsar::Result>::get<std::__1::chrono::duration<long long, std::__1::ratio<1l, 1000l> > >(this=0x00007ff7bfefe508, res=0x00007ff7bfefe52b, value=0x00007ff7bfefe52c, d=(__rep_ = 100)) at Future.h:86:39
    frame #9: 0x0000000106dcf18d _pulsar.so`waitForAsyncResult(func= Lambda in File client.cc at Line 103)>)>) at utils.cc:34:52
    frame #10: 0x0000000106d3627c _pulsar.so`Client_close(client=0x0000000100ae6d38) at client.cc:103:5
    frame #11: 0x0000000106d5e821 _pulsar.so`_object* boost::python::detail::invoke<int, void (*)(pulsar::Client&), boost::python::arg_from_python<pulsar::Client&> >((null)=invoke_tag_<true, false> @ 0x00007ff7bfefe628, (null)=0x00007ff7bfefe64c, f=0x00006000000081c8, ac0=0x00007ff7bfefe670)(pulsar::Client&), boost::python::arg_from_python<pulsar::Client&>&) at invoke.hpp:79:5
    frame #12: 0x0000000106d5e79c _pulsar.so`boost::python::detail::caller_arity<1u>::impl<void (this=0x00006000000081c8, args_=0x0000000100ae7100, (null)=0x0000000000000000)(pulsar::Client&), boost::python::default_call_policies, boost::mpl::vector2<void, pulsar::Client&> >::operator()(_object*, _object*) at caller.hpp:233:32
    frame #13: 0x0000000106d5e699 _pulsar.so`boost::python::objects::caller_py_function_impl<boost::python::detail::caller<void (*)(pulsar::Client&), boost::python::default_call_policies, boost::mpl::vector2<void, pulsar::Client&> > >::operator(this=0x00006000000081c0, args=0x0000000100ae7100, kw=0x0000000000000000)(_object*, _object*) at py_function.hpp:38:16
    frame #14: 0x0000000104076d39 libboost_python39-mt.dylib`boost::python::objects::function::call(_object*, _object*) const + 665
    frame #15: 0x000000010407921a libboost_python39-mt.dylib`boost::detail::function::void_function_ref_invoker0<boost::python::objects::(anonymous namespace)::bind_return, void>::invoke(boost::detail::function::function_buffer&) + 26
    frame #16: 0x000000010407ee77 libboost_python39-mt.dylib`boost::python::detail::exception_handler::operator()(boost::function0<void> const&) const + 71
    frame #17: 0x0000000106cc6b25 _pulsar.so`boost::python::detail::translate_exception<PulsarException, void (*)(PulsarException const&)>::operator(this=0x0000600000c06108, handler=0x0000600000c06100, f=0x00007ff7bfefe948, translate=(_pulsar.so`translateException(PulsarException const&) at pulsar.cc:35))(boost::python::detail::exception_handler const&, boost::function0<void> const&, void (*)(PulsarException const&)) const at translate_exception.hpp:46:20
    frame #18: 0x0000000106cc6a9a _pulsar.so`bool boost::_bi::list3<boost::arg<1>, boost::arg<2>, boost::_bi::value<void (*)(PulsarException const&)> >::operator(this=0x0000600000c06110, (null)=type<bool> @ 0x00007ff7bfefe858, f=0x0000600000c06108, a=0x00007ff7bfefe888, (null)=0)<bool, boost::python::detail::translate_exception<PulsarException, void (*)(PulsarException const&)>, boost::_bi::rrlist2<boost::python::detail::exception_handler const&, boost::function0<void> const&> >(boost::_bi::type<bool>, boost::python::detail::translate_exception<PulsarException, void (*)(PulsarException const&)>&, boost::_bi::rrlist2<boost::python::detail::exception_handler const&, boost::function0<void> const&>&, long) at bind.hpp:367:16
    frame #19: 0x0000000106cc69e5 _pulsar.so`bool boost::_bi::bind_t<bool, boost::python::detail::translate_exception<PulsarException, void (*)(PulsarException const&)>, boost::_bi::list3<boost::arg<1>, boost::arg<2>, boost::_bi::value<void (*)(PulsarException const&)> > >::operator(this=0x0000600000c06108, a1=0x0000600000c06100, a2=0x00007ff7bfefe948)<boost::python::detail::exception_handler const&, boost::function0<void> const&>(boost::python::detail::exception_handler const&, boost::function0<void> const&) at bind.hpp:1297:16
    frame #20: 0x0000000106cc65fd _pulsar.so`boost::detail::function::function_obj_invoker2<boost::_bi::bind_t<bool, boost::python::detail::translate_exception<PulsarException, void (*)(PulsarException const&)>, boost::_bi::list3<boost::arg<1>, boost::arg<2>, boost::_bi::value<void (*)(PulsarException const&)> > >, bool, boost::python::detail::exception_handler const&, boost::function0<void> const&>::invoke(function_obj_ptr=0x0000600000c06108, a0=0x0000600000c06100, a1=0x00007ff7bfefe948) at function_template.hpp:137:18

Other threads:

* thread #2
  * frame #0: 0x00007ff811dfc34e libsystem_kernel.dylib`kevent + 10
    frame #1: 0x0000000106e81ce8 _pulsar.so`boost::asio::detail::kqueue_reactor::run(this=0x0000000100d36620, usec=-1, ops=0x0000700006a46ab8) at kqueue_reactor.ipp:440:20
    frame #2: 0x0000000106e8347e _pulsar.so`boost::asio::detail::scheduler::do_run_one(this=0x0000000100d33bf0, lock=0x0000700006a46a20, this_thread=0x0000700006a46a58, ec=0x0000700006a46dd0) at scheduler.ipp:476:16
    frame #3: 0x0000000106e83117 _pulsar.so`boost::asio::detail::scheduler::run(this=0x0000000100d33bf0, ec=0x0000700006a46dd0) at scheduler.ipp:210:10
    frame #4: 0x0000000106ff92b1 _pulsar.so`boost::asio::io_context::run(this=0x0000600003704268, ec=0x0000700006a46dd0) at io_context.ipp:71:16
    frame #5: 0x0000000106ff8d08 _pulsar.so`pulsar::ExecutorService::start(this=0x0000600000209628)::$_0::operator()() const at ExecutorService.cc:43:30
    frame #6: 0x0000000106ff8b75 _pulsar.so`decltype(__f=0x0000600000209628)::$_0>(fp)()) std::__1::__invoke<pulsar::ExecutorService::start()::$_0>(pulsar::ExecutorService::start()::$_0&&) at type_traits:3918:1
    frame #7: 0x0000000106ff8b15 _pulsar.so`void std::__1::__thread_execute<std::__1::unique_ptr<std::__1::__thread_struct, std::__1::default_delete<std::__1::__thread_struct> >, pulsar::ExecutorService::start()::$_0>(__t=size=2, (null)=__tuple_indices<> @ 0x0000700006a46f68)::$_0>&, std::__1::__tuple_indices<>) at thread:287:5
    frame #8: 0x0000000106ff82f2 _pulsar.so`void* std::__1::__thread_proxy<std::__1::tuple<std::__1::unique_ptr<std::__1::__thread_struct, std::__1::default_delete<std::__1::__thread_struct> >, pulsar::ExecutorService::start()::$_0> >(__vp=0x0000600000209620) at thread:298:5
    frame #9: 0x00007ff811e344e1 libsystem_pthread.dylib`_pthread_start + 125
    frame #10: 0x00007ff811e2ff6b libsystem_pthread.dylib`thread_start + 15

* thread #3
  * frame #0: 0x00007ff811dfa3ea libsystem_kernel.dylib`__psynch_cvwait + 10
    frame #1: 0x00007ff811e34a6f libsystem_pthread.dylib`_pthread_cond_wait + 1249
    frame #2: 0x0000000106e84244 _pulsar.so`void boost::asio::detail::posix_event::wait<boost::asio::detail::conditionally_enabled_mutex::scoped_lock>(this=0x0000000100d5b728, lock=0x0000700006ac9e40) at posix_event.hpp:119:7
    frame #3: 0x0000000106e83cba _pulsar.so`boost::asio::detail::conditionally_enabled_event::wait(this=0x0000000100d5b720, lock=0x0000700006ac9e40) at conditionally_enabled_event.hpp:97:14
    frame #4: 0x0000000106e83591 _pulsar.so`boost::asio::detail::scheduler::do_run_one(this=0x0000000100d5b6a0, lock=0x0000700006ac9e40, this_thread=0x0000700006ac9e78, ec=0x0000700006ac9f40) at scheduler.ipp:501:21
    frame #5: 0x0000000106e83117 _pulsar.so`boost::asio::detail::scheduler::run(this=0x0000000100d5b6a0, ec=0x0000700006ac9f40) at scheduler.ipp:210:10
    frame #6: 0x0000000106eb635d _pulsar.so`boost::asio::detail::resolver_service_base::work_scheduler_runner::operator(this=0x000060000000c938)() at resolver_service_base.ipp:38:21
    frame #7: 0x0000000106eb6329 _pulsar.so`boost::asio::detail::posix_thread::func<boost::asio::detail::resolver_service_base::work_scheduler_runner>::run(this=0x000060000000c930) at posix_thread.hpp:86:7
    frame #8: 0x0000000106e82ef1 _pulsar.so`::boost_asio_detail_posix_thread_function(arg=0x000060000000c930) at posix_thread.ipp:74:13
    frame #9: 0x00007ff811e344e1 libsystem_pthread.dylib`_pthread_start + 125
    frame #10: 0x00007ff811e2ff6b libsystem_pthread.dylib`thread_start + 15

* thread #4
  * frame #0: 0x00007ff811dfc34e libsystem_kernel.dylib`kevent + 10
    frame #1: 0x0000000106e81ce8 _pulsar.so`boost::asio::detail::kqueue_reactor::run(this=0x0000000100204510, usec=-1, ops=0x0000700006b4cab8) at kqueue_reactor.ipp:440:20
    frame #2: 0x0000000106e8347e _pulsar.so`boost::asio::detail::scheduler::do_run_one(this=0x00000001002043f0, lock=0x0000700006b4ca20, this_thread=0x0000700006b4ca58, ec=0x0000700006b4cdd0) at scheduler.ipp:476:16
    frame #3: 0x0000000106e83117 _pulsar.so`boost::asio::detail::scheduler::run(this=0x00000001002043f0, ec=0x0000700006b4cdd0) at scheduler.ipp:210:10
    frame #4: 0x0000000106ff92b1 _pulsar.so`boost::asio::io_context::run(this=0x00006000037080e8, ec=0x0000700006b4cdd0) at io_context.ipp:71:16
    frame #5: 0x0000000106ff8d08 _pulsar.so`pulsar::ExecutorService::start(this=0x0000600000201768)::$_0::operator()() const at ExecutorService.cc:43:30
    frame #6: 0x0000000106ff8b75 _pulsar.so`decltype(__f=0x0000600000201768)::$_0>(fp)()) std::__1::__invoke<pulsar::ExecutorService::start()::$_0>(pulsar::ExecutorService::start()::$_0&&) at type_traits:3918:1
    frame #7: 0x0000000106ff8b15 _pulsar.so`void std::__1::__thread_execute<std::__1::unique_ptr<std::__1::__thread_struct, std::__1::default_delete<std::__1::__thread_struct> >, pulsar::ExecutorService::start()::$_0>(__t=size=2, (null)=__tuple_indices<> @ 0x0000700006b4cf68)::$_0>&, std::__1::__tuple_indices<>) at thread:287:5
    frame #8: 0x0000000106ff82f2 _pulsar.so`void* std::__1::__thread_proxy<std::__1::tuple<std::__1::unique_ptr<std::__1::__thread_struct, std::__1::default_delete<std::__1::__thread_struct> >, pulsar::ExecutorService::start()::$_0> >(__vp=0x0000600000201760) at thread:298:5
    frame #9: 0x00007ff811e344e1 libsystem_pthread.dylib`_pthread_start + 125
    frame #10: 0x00007ff811e2ff6b libsystem_pthread.dylib`thread_start + 15