Search before asking
Version
- OS: Ubuntu 20.04
- Pulsar: 2.10.2
- Client: 85b1b53
Minimal reproduce step
Modify the SampleProducer.cc:
#include <pulsar/Client.h>
#include <chrono>
#include <iostream>
#include <thread>
using namespace pulsar;
int main() {
Client client("pulsar://localhost:6650");
const auto topic = "my-topic";
Producer producer;
if (ResultOk !=
client.createProducer(
topic, ProducerConfiguration().setBatchingEnabled(true).setBatchingMaxMessages(2), producer)) {
return 1;
}
Consumer consumer;
if (ResultOk != client.subscribe(topic, "sub", consumer)) {
return 2;
}
constexpr int numMessages = 1000000;
std::thread backgroudWorker{[&producer] {
for (int i = 0; i < numMessages; i++) {
producer.sendAsync(MessageBuilder().setContent("msg-" + std::to_string(i)).build(), nullptr);
}
}};
Message msg;
for (int i = 0; i < numMessages; i++) {
consumer.receive(msg);
std::this_thread::sleep_for(std::chrono::milliseconds(10));
if (i % 1000 == 999) {
std::cout << "received " << msg.getDataAsString() << " from " << msg.getMessageId() << std::endl;
}
}
backgroudWorker.join();
client.close();
}
Then run the ./example/SampleProducer. Use the following script to monitor the memory usage increment.
#!/bin/bash
set -e
LOOP_CNT=0
for i in {1..1000}; do
PROC_ID=$(ps aux | grep SampleProducer | grep -v grep | awk '{print $2}')
if [[ $PROC_ID ]]; then
echo "LOOP_CNT: $LOOP_CNT"
set -x
cat /proc/$PROC_ID/status | grep VmData
set +x
LOOP_CNT=$((LOOP_CNT+1))
sleep 1
else
echo "No process running a.out is found"
fi
sleep 1
done
What did you expect to see?
The memory usage should keep stable.
What did you see instead?
VmData: 35900 kB
VmData: 35908 kB
VmData: 35912 kB
VmData: 36036 kB
...
VmData: 53844 kB
We can see the VmData is increasing.
Anything else?
The root cause is that we use a BatchAcknowledgementTracker to maintain a map from the MessageId (without batch index) to a bit set that represents which messages are already acknowledged in a batch. The entry are only removed in
|
this->batchAcknowledgementTracker_.deleteAckedMessage(messageId, proto::CommandAck::Cumulative); |
Are you willing to submit a PR?
Search before asking
Version
Minimal reproduce step
Modify the
SampleProducer.cc:Then run the
./example/SampleProducer. Use the following script to monitor the memory usage increment.What did you expect to see?
The memory usage should keep stable.
What did you see instead?
We can see the
VmDatais increasing.Anything else?
The root cause is that we use a
BatchAcknowledgementTrackerto maintain a map from the MessageId (without batch index) to a bit set that represents which messages are already acknowledged in a batch. The entry are only removed inpulsar-client-cpp/lib/ConsumerImpl.cc
Line 1063 in 85b1b53
Are you willing to submit a PR?