ISSUE-7851: pulsar-client-cpp crashes consumer due to thread safty

[sijie]

Original Issue: apache#7851

---

Describe the bug
When running the pulsar-client-cpp on 2.5.2 we are experiencing the following stack trace. Our code instanciates and holds open a client then time slices the consumers on a number of threads. Each thread subscribe, consumes message followed by immediate acknowledge, then closes the consumer. The stack trace has been seen between a few hours and a month apart. We finally received a core file for the crash shown below:

#0 0x00007f03272b02c7 in raise () from /lib64/libc.so.6
#1 0x00007f03272b19b8 in abort () from /lib64/libc.so.6
#2 0x0000000004075565 in Basics::Backtrace::DoCoreDump(char const*, bool) ()
#3 0x00000000040c95c5 in Basics::GlobalSignalHandlers::logCoreDump() ()
#4 0x00000000040ca6a5 in Basics::sigHandler_withinATryCatch(int, siginfo*, void*) ()
#5 0x00000000040ca77e in Basics::sigHandler(int, siginfo*, void*) ()
#6
#7 0x00007f018313414e in pulsar::MessageId::operator< (this=0x7f030ea6c9e8, other=
@0x7f00ebffaf00: {impl_ = std::shared_ptr (count 3, weak 0) 0x7f030f68b538})
at /usr/local/src/apache-pulsar-microfocus/pulsar-client-cpp/lib/MessageId.cc:100
#8 0x00007f01831c63f0 in operator() (this=0x7f030ea6c9c0, _y=@0x7f00ebffaf00: {impl = std::shared_ptr (count 3, weak 0) 0x7f030f68b538},
__x=) at /usr/include/c++/4.8.2/bits/stl_function.h:235
#9 std::_Rb_tree<pulsar::MessageId, std::pair<pulsar::MessageId const, boost::dynamic_bitset<unsigned long, std::allocator > >, std::_Select1st<std::pair<pulsar::MessageId const, boost::dynamic_bitset<unsigned long, std::allocator > > >, std::lesspulsar::MessageId, std::allocator<std::pair<pulsar::MessageId const, boost::dynamic_bitset<unsigned long, std::allocator > > > >::_M_get_insert_hint_unique_pos (
this=this@entry=0x7f030ea6c9c0, __position=, _k=@0x7f00ebffaf00: {impl = std::shared_ptr (count 3, weak 0) 0x7f030f68b538})
at /usr/include/c++/4.8.2/bits/stl_tree.h:1422
#10 0x00007f01831c6419 in std::_Rb_tree<pulsar::MessageId, std::pair<pulsar::MessageId const, boost::dynamic_bitset<unsigned long, std::allocator > >, std::_Select1st<std::pair<pulsar::MessageId const, boost::dynamic_bitset<unsigned long, std::allocator > > >, std::lesspulsar::MessageId, std::allocator<std::pair<pulsar::MessageId const, boost::dynamic_bitset<unsigned long, std::allocator > > > >::M_insert_unique<std::pair<pulsar::MessageId, boost::dynamic_bitset<unsigned long, std::allocator > > >(std::_Rb_tree_const_iterator<std::pair<pulsar::MessageId const, boost::dynamic_bitset<unsigned long, std::allocator > > >, std::pair<pulsar::MessageId, boost::dynamic_bitset<unsigned long, std::allocator > >&&) (this=this@entry=0x7f030ea6c9c0, __position=, _position@entry=
{first = {impl = std::shared_ptr (count -660842048, weak 32512) 0xffffffffffffffff}, second = {static bits_per_block = <error reading variable: No global symbol "boost::dynamic_bitset<unsigned long, std::allocator >::bits_per_block".>, static npos = , static ulong_width = <error reading variable: No global symbol "boost::dynamic_bitset<unsigned long, std::allocator >::ulong_width".>, m_bits = std::vector of length 128, capacity -25369645 = {139645905820120, 139645905820096, 139645903947864, 139645903947840, 139645895612888, 139645895612864, 139645903478456, 139645903478432, 139645905415224, 139645905415200, 139645907124328, 139645907124304, 139645897524632, 139645897524608, 139645896941944, 139645896941920, 139645903479768, 139645903479744, 139645897213704, 139645897213680, 139645897213832, 139645897213808, 139645902580088, 139645902580064, 139645902579928, 139645902579904, 139645896941448, 139645896941424, 139645897213320, 139645897213296, 139645901676504, 139645901676480, 139645899269576, 139645899269552, 139645905827016, 139645905826992, 139645898799784, 139645898799760, 139645901615560, 139645901615536, 139645903352184, 139645903352160, 139645905772872, 139645905772848, 139645899479656, 139645899479632, 139645905308152, 139645905308128, 139645897513528, 139645897513504, 139645896668632, 139645896668608, 139645898799128, 139645898799104, 139645905331112, 139645905331088, 139645905331304, 139645905331280, 139645898799256, 139645898799232, 139645899213224, 139645899213200, 139645905418312, 139645905418288, 139645903262888, 139645903262864, 139645896344696, 139645896344672, 139645899177384, 139645899177360, 139645905401368, 139645905401344, 139645907120840, 139645907120816, 139645896458248, 139645896458224, 139645898790536, 139645898790512, 139645897192552, 139645897192528, 139645899378632, 139645899378608, 112513136, 216, 112528512, 433, 139645899296624, 139645900272320, 139645898787520, 139645898012976, 112486928, 61818273202178, 112528416, 273, 139645895573880, 139645895573880, 0, 0, 45035996273704978, 45035996273704978, 1, 114046456, 0, 0, 0, 139645898784768, 114046456, 139645898012928, 0, 0, 139645897425536, 139642271694848, 0, 139645897425560, 139645897425560, 0, 114046456, 139645897424640, 0, 0, 139645897424848, 139642271694848, 0, 139645897425640, 139645897425640, 0, 336, 68}, m_num_bits = 139645694467488}},
__v=__v@entry=<unknown type in /usr/lib/debug/usr/local/itom-di-pulsarudx/lib/libitom-di-pulsarudx-9.2.1.so.debug, CU 0x59efd9, DIE 0x5aa2ab>)
at /usr/include/c++/4.8.2/bits/stl_tree.h:1478
#11 0x00007f01831c4d19 in insert<std::pair<pulsar::MessageId, boost::dynamic_bitset<> >, void> (
__x=<unknown type in /usr/lib/debug/usr/local/itom-di-pulsarudx/lib/libitom-di-pulsarudx-9.2.1.so.debug, CU 0x59efd9, DIE 0x5aa2ab>,
_position=, this=0x7f030ea6c9c0) at /usr/include/c++/4.8.2/bits/stl_map.h:657
#12 pulsar::BatchAcknowledgementTracker::receivedMessage (this=this@entry=0x7f030ea6c998, message=warning: RTTI symbol not found for class 'std::_Sp_counted_ptr_inplace<pulsar::MessageImpl, std::allocatorpulsar::MessageImpl, (__gnu_cxx::_Lock_policy)2>'
warning: RTTI symbol not found for class 'std::_Sp_counted_ptr_inplace<pulsar::MessageImpl, std::allocatorpulsar::MessageImpl, (__gnu_cxx::_Lock_policy)2>'
---Type to continue, or q to quit---
@0x7f00ebffb350: {impl = std::shared_ptr (count 1, weak 0) 0x7f030f6a7148})
at /usr/local/src/apache-pulsar-microfocus/pulsar-client-cpp/lib/BatchAcknowledgementTracker.cc:61
#13 0x00007f0183113a2e in pulsar::ConsumerImpl::receiveIndividualMessagesFromBatch (this=this@entry=0x7f030ea6bcc8, cnx=warning: RTTI symbol not found for class 'std::_Sp_counted_ptr<pulsar::ClientConnection*, (__gnu_cxx::_Lock_policy)2>'
warning: RTTI symbol not found for class 'std::_Sp_counted_ptr<pulsar::ClientConnection*, (__gnu_cxx::_Lock_policy)2>'
std::shared_ptr (count 7, weak 4) 0x7f030c018fd0, batchedMessage=warning: RTTI symbol not found for class 'std::_Sp_counted_ptr_inplace<pulsar::MessageImpl, std::allocatorpulsar::MessageImpl, (__gnu_cxx::_Lock_policy)2>'
warning: RTTI symbol not found for class 'std::_Sp_counted_ptr_inplace<pulsar::MessageImpl, std::allocatorpulsar::MessageImpl, (__gnu_cxx::Lock_policy)2>'
@0x7f00ebffb350: {impl = std::shared_ptr (count 1, weak 0) 0x7f030f6a7148},
redeliveryCount=0) at /usr/local/src/apache-pulsar-microfocus/pulsar-client-cpp/lib/ConsumerImpl.cc:369

Now understanding that we are doing a close, focus in on Line #12. This shows that we are manipulating a RB tree which is not inherently thread safe. And we have two possible competing threads:

1. The receiving message thread which is receiving and processing messages with handleRead.. which is many traces further down in the stack. The above is the stack for handleRead Thread.
2. The second thread is triggered by a Consumer::close(). This thread is allowed to interfere with the BatchAcknowledgementTracker, in that it eventually calls the clear() on the method. The BatchAcknowledgementTracker::clear() methods is not guarded by the mutex within the class so the clear() is able to proceed at the same time the insert of a new record is happening. The objects in the tree are then removed, while the iteration is happening. Resulting in the core dump above.

To Reproduce
Steps to reproduce the behavior:
Since this is 1/1 Billion to 1/1Trillion failure rate per message. This is not advisable to attempt to reproduce and simply wait for the failure to occur while the software grinds on processing messages and closing the consumers()...

Create a Unit test that executes 2 threads.

1. inserts messages into the BatchAcknowledgementTracker simulating reads happening.
2. The other calling the clear() method on the batch acknowledgment tracker.

Core dump should occur fairly quickly.

Expected behavior
Software runs without crashing.

Desktop (please complete the following information):

• OS: Centos

Additional context
Code inspection of BatchAcknowledgmentTracker and UnAcknowledgeMessageTrackerEnabled classes both show similar problems with the clear() not using a mutex and thus not thread safe.
'