tursom/memgraph
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Pulsar e2e tests (#296)

Browse Source
This commit is contained in:
Antonio Andelic
2021-11-15 16:23:49 +01:00
committed by Antonio Andelic
parent b66cc66503
commit c7b045bffc
15 changed files with 1136 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
src/integrations/pulsar/consumer.cpp
View File

@@ -45,7 +45,8 @@ pulsar_client::Result ConsumeMessage(pulsar_client::Reader &reader, pulsar_clien
template <PulsarConsumer TConsumer>
utils::BasicResult<std::string, std::vector<Message>> GetBatch(TConsumer &consumer, const ConsumerInfo &info,
std::atomic<bool> &is_running) {
std::atomic<bool> &is_running,
const pulsar_client::MessageId &last_message_id) {
std::vector<Message> batch{};
const auto batch_size = info.batch_size.value_or(kDefaultBatchSize);
@@ -54,14 +55,16 @@ utils::BasicResult<std::string, std::vector<Message>> GetBatch(TConsumer &consum
auto remaining_timeout_in_ms = info.batch_interval.value_or(kDefaultBatchInterval).count();
auto start = std::chrono::steady_clock::now();
for (int64_t i = 0; remaining_timeout_in_ms > 0 && i < batch_size && is_running.load(); ++i) {
while (remaining_timeout_in_ms > 0 && batch.size() < batch_size && is_running) {
pulsar_client::Message message;
const auto result = ConsumeMessage(consumer, message, remaining_timeout_in_ms);
switch (result) {
case pulsar_client::Result::ResultTimeout:
return std::move(batch);
case pulsar_client::Result::ResultOk:
batch.emplace_back(Message{std::move(message)});
if (message.getMessageId() != last_message_id) {
batch.emplace_back(Message{std::move(message)});
}
break;
default:
spdlog::warn(fmt::format("Unexpected error while consuming message from consumer {}, error: {}",
@@ -87,15 +90,7 @@ class SpdlogLogger : public pulsar_client::Logger {
};
class SpdlogLoggerFactory : public pulsar_client::LoggerFactory {
pulsar_client::Logger *getLogger(const std::string & /*file_name*/) override {
if (!logger_) {
logger_ = std::make_unique<SpdlogLogger>();
}
return logger_.get();
}
private:
std::unique_ptr<SpdlogLogger> logger_;
pulsar_client::Logger *getLogger(const std::string & /*file_name*/) override { return new SpdlogLogger; }
};
pulsar_client::Client CreateClient(const std::string &service_url) {
@@ -111,6 +106,8 @@ std::span<const char> Message::Payload() const {
return {static_cast<const char *>(message_.getData()), message_.getLength()};
}
std::string_view Message::TopicName() const { return message_.getTopicName(); }
Consumer::Consumer(ConsumerInfo info, ConsumerFunction consumer_function)
: info_{std::move(info)},
client_{CreateClient(info_.service_url)},
@@ -203,7 +200,7 @@ void Consumer::Check(std::optional<std::chrono::milliseconds> timeout, std::opti
throw ConsumerCheckFailedException(info_.consumer_name, "Timeout reached");
}
auto maybe_batch = GetBatch(reader, info_, is_running_);
auto maybe_batch = GetBatch(reader, info_, is_running_, last_message_id_);
if (maybe_batch.HasError()) {
throw ConsumerCheckFailedException(info_.consumer_name, maybe_batch.GetError());
@@ -241,7 +238,7 @@ void Consumer::StartConsuming() {
utils::ThreadSetName(full_thread_name.substr(0, kMaxThreadNameSize));
while (is_running_) {
auto maybe_batch = GetBatch(consumer_, info_, is_running_);
auto maybe_batch = GetBatch(consumer_, info_, is_running_, last_message_id_);
if (maybe_batch.HasError()) {
spdlog::warn("Error happened in consumer {} while fetching messages: {}!", info_.consumer_name,