Fix use-after-free crash in getBrokerConsumerStatsAsync when connection is closed

[dragonls]

Fixes #536

Motivation

A use-after-free crash occurs in MultiTopicsConsumerImpl::getBrokerConsumerStatsAsync() when the underlying connection is closed while an asynchronous consumer stats request is pending.

The root cause has two aspects:

1. Reference capture of stack variables: The lambda captures local variables (latchPtr, statsPtr, i) by reference. If the callback is invoked after the function returns, these references become dangling.

2. Direct this capture in lambda: Both the outer and inner lambdas capture this directly. Although there's a weakSelf.lock() check, the actual function call handleGetConsumerStats(...) is invoked through the captured this pointer (implicit this->handleGetConsumerStats(...)), not through self->. When ClientConnection::close() calls setFailed() on pending promises, it synchronously triggers the registered callbacks, potentially causing use-after-free if the MultiTopicsConsumerImpl has been destroyed.

Modifications

1. Use shared_ptr<atomic> for index: Replace stack variable size_t i with std::shared_ptr<std::atomic<size_t>> for thread-safe indexing and proper lifetime management.

2. Create weakSelf outside lambdas: Move weak_from_this() call outside to avoid capturing this in the outer lambda.

3. Remove this capture from all lambdas: Neither the outer nor inner lambda captures this anymore.

4. Call member function through self->: Change handleGetConsumerStats(...) to self->handleGetConsumerStats(...) to ensure the call goes through the shared_ptr, not the raw this pointer.

Verifying this change

• Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

The crash scenario is difficult to reproduce in unit tests as it requires precise timing of connection closure during async operations. However, the fix follows the standard C++ weak_ptr/shared_ptr pattern for preventing use-after-free in asynchronous callbacks.

Documentation

• doc-required
  (Your PR needs to update docs and you will update later)

• doc-not-needed
  (Please explain why)

• doc
  (Your PR contains doc changes)

• doc-complete
  (Docs have been already added)

[BewareMyPower]

The callback of forEachValue is immediately executed in the current method.

pulsar-client-cpp/lib/SynchronizedHashMap.h

Lines 95 to 99 in 0b893a1

	void forEachValue(ValueFunc&& each) {
	Lock lock{mutex_};
	for (auto&& kv : data_) {
	each(kv.second);
	}

so I believe this patch does not fix the actual issue

[dragonls]

Yes, but in consumers_.forEachValue, it also calls consumer->getBrokerConsumerStatsAsync, which passed the weakSelf into the callback. The callback is executed later.

[BewareMyPower]

The lambda of getBrokerConsumerStatsAsync captures all variables by value:

consumer->getBrokerConsumerStatsAsync([this, weakSelf, latchPtr, statsPtr, index, callback](

[BewareMyPower]

I believe I've figured out the cause, let me create a new fix

[dragonls]

I believe I've figured out the cause, let me create a new fix

Nice, looking forward to it.

[BewareMyPower]

Replace this PR by #538