PIP-74: Implemented memory limit in C++ producer

[merlimat]

Motivation

Similar to what already implemented in Java client, adding a client-wide memory limit for messages produced on all the C++ producers.

Modifications

Refactored the pendingMessagesQueue handling code to be similar to the Java implementation, using a "Deque" and a semaphore, rather that having all the logic collapsed into the queue implementation.

[BewareMyPower]

Did the apache/pulsar-build image change recently? The cpp CI failure was caused by

-- Found the following Boost libraries:
--   system
-- Found PythonLibs: /usr/lib/x86_64-linux-gnu/libpython3.5m.so (found version "3.5.2") 
-- PYTHON: 3.5.2
-- DETECTED Python 3
-- Boost version: 1.58.0

and

CMake Error at python/CMakeLists.txt:84 (MESSAGE):
  Could not find Boost Python library

because the test image apache/pulsar-build doesn't install the Boost.Python for Python3.

I see other tests detect Python2 but not Python3.

[sijie]

@BewareMyPower Can you review it again?

[sijie]

@BewareMyPower Can you review it again?

[BewareMyPower]

LGTM

[erobot]

pulsar/pulsar-client-cpp/lib/ProducerImpl.cc

Lines 249 to 270 in 430f1f3

	std::shared_ptr<ProducerImpl::PendingCallbacks> ProducerImpl::getPendingCallbacksWhenFailed() {
	auto callbacks = std::make_shared<PendingCallbacks>();
	callbacks->opSendMsgs.reserve(pendingMessagesQueue_.size());
	LOG_DEBUG(getName() << "# messages in pending queue : " << pendingMessagesQueue_.size());
	// Iterate over a copy of the pending messages queue, to trigger the future completion
	// without holding producer mutex.
	for (auto& op : pendingMessagesQueue_) {
	callbacks->opSendMsgs.push_back(op);
	}
	if (batchMessageContainer_) {
	OpSendMsg opSendMsg;
	if (batchMessageContainer_->createOpSendMsg(opSendMsg) == ResultOk) {
	callbacks->opSendMsgs.emplace_back(opSendMsg);
	}
	batchMessageContainer_->clear();
	}
	pendingMessagesQueue_.clear();
	return callbacks;
	}

@merlimat getPendingCallbacksWhenFailed is called when message timeouts. This function clears pendingMessagesQueue_ and batchMessageContainer_, so I think releaseSemaphore should be called here as well.

[merlimat]

@erobot That's completely correct. It was a regression introduced here. I've created PR to fix it: #10144

[Anonymitaet]

@merlimat thanks for your contribution!

I see you've added instructions of setMemoryLimit to C++ API doc (ClientConfiguration.h), do we need to add the instructions here? Thanks