| /** |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| #include "MultiTopicsConsumerImpl.h" |
| |
| #include <stdexcept> |
| |
| #include "ClientImpl.h" |
| #include "ConsumerImpl.h" |
| #include "ExecutorService.h" |
| #include "LogUtils.h" |
| #include "LookupService.h" |
| #include "MessageImpl.h" |
| #include "MessagesImpl.h" |
| #include "MultiResultCallback.h" |
| #include "MultiTopicsBrokerConsumerStatsImpl.h" |
| #include "TopicName.h" |
| #include "UnAckedMessageTrackerDisabled.h" |
| #include "UnAckedMessageTrackerEnabled.h" |
| |
| DECLARE_LOG_OBJECT() |
| |
| using namespace pulsar; |
| |
| MultiTopicsConsumerImpl::MultiTopicsConsumerImpl(ClientImplPtr client, TopicNamePtr topicName, |
| int numPartitions, const std::string& subscriptionName, |
| const ConsumerConfiguration& conf, |
| LookupServicePtr lookupServicePtr, |
| const ConsumerInterceptorsPtr& interceptors, |
| const Commands::SubscriptionMode subscriptionMode, |
| boost::optional<MessageId> startMessageId) |
| : MultiTopicsConsumerImpl(client, {topicName->toString()}, subscriptionName, topicName, conf, |
| lookupServicePtr, interceptors, subscriptionMode, startMessageId) { |
| topicsPartitions_[topicName->toString()] = numPartitions; |
| } |
| |
| MultiTopicsConsumerImpl::MultiTopicsConsumerImpl(ClientImplPtr client, const std::vector<std::string>& topics, |
| const std::string& subscriptionName, TopicNamePtr topicName, |
| const ConsumerConfiguration& conf, |
| LookupServicePtr lookupServicePtr, |
| const ConsumerInterceptorsPtr& interceptors, |
| const Commands::SubscriptionMode subscriptionMode, |
| boost::optional<MessageId> startMessageId) |
| : ConsumerImplBase(client, topicName ? topicName->toString() : "EmptyTopics", |
| Backoff(milliseconds(100), seconds(60), milliseconds(0)), conf, |
| client->getListenerExecutorProvider()->get()), |
| client_(client), |
| subscriptionName_(subscriptionName), |
| conf_(conf), |
| incomingMessages_(conf.getReceiverQueueSize()), |
| messageListener_(conf.getMessageListener()), |
| lookupServicePtr_(lookupServicePtr), |
| numberTopicPartitions_(std::make_shared<std::atomic<int>>(0)), |
| topics_(topics), |
| subscriptionMode_(subscriptionMode), |
| startMessageId_(startMessageId), |
| interceptors_(interceptors) { |
| std::stringstream consumerStrStream; |
| consumerStrStream << "[Muti Topics Consumer: " |
| << "TopicName - " << topic_ << " - Subscription - " << subscriptionName << "]"; |
| consumerStr_ = consumerStrStream.str(); |
| |
| if (conf.getUnAckedMessagesTimeoutMs() != 0) { |
| if (conf.getTickDurationInMs() > 0) { |
| unAckedMessageTrackerPtr_.reset(new UnAckedMessageTrackerEnabled( |
| conf.getUnAckedMessagesTimeoutMs(), conf.getTickDurationInMs(), client, *this)); |
| } else { |
| unAckedMessageTrackerPtr_.reset( |
| new UnAckedMessageTrackerEnabled(conf.getUnAckedMessagesTimeoutMs(), client, *this)); |
| } |
| } else { |
| unAckedMessageTrackerPtr_.reset(new UnAckedMessageTrackerDisabled()); |
| } |
| auto partitionsUpdateInterval = static_cast<unsigned int>(client->conf().getPartitionsUpdateInterval()); |
| if (partitionsUpdateInterval > 0) { |
| partitionsUpdateTimer_ = listenerExecutor_->createDeadlineTimer(); |
| partitionsUpdateInterval_ = boost::posix_time::seconds(partitionsUpdateInterval); |
| lookupServicePtr_ = client->getLookup(); |
| } |
| |
| state_ = Pending; |
| } |
| |
| void MultiTopicsConsumerImpl::start() { |
| if (topics_.empty()) { |
| State state = Pending; |
| if (state_.compare_exchange_strong(state, Ready)) { |
| LOG_DEBUG("No topics passed in when create MultiTopicsConsumer."); |
| multiTopicsConsumerCreatedPromise_.setValue(get_shared_this_ptr()); |
| return; |
| } else { |
| LOG_ERROR("Consumer " << consumerStr_ << " in wrong state: " << state_); |
| multiTopicsConsumerCreatedPromise_.setFailed(ResultUnknownError); |
| return; |
| } |
| } |
| |
| // start call subscribeOneTopicAsync for each single topic |
| int topicsNumber = topics_.size(); |
| std::shared_ptr<std::atomic<int>> topicsNeedCreate = std::make_shared<std::atomic<int>>(topicsNumber); |
| // subscribe for each passed in topic |
| auto weakSelf = weak_from_this(); |
| for (std::vector<std::string>::const_iterator itr = topics_.begin(); itr != topics_.end(); itr++) { |
| auto topic = *itr; |
| subscribeOneTopicAsync(topic).addListener( |
| [this, weakSelf, topic, topicsNeedCreate](Result result, const Consumer& consumer) { |
| auto self = weakSelf.lock(); |
| if (self) { |
| handleOneTopicSubscribed(result, consumer, topic, topicsNeedCreate); |
| } |
| }); |
| } |
| } |
| |
| void MultiTopicsConsumerImpl::handleOneTopicSubscribed(Result result, Consumer consumer, |
| const std::string& topic, |
| std::shared_ptr<std::atomic<int>> topicsNeedCreate) { |
| if (result != ResultOk) { |
| state_ = Failed; |
| // Use the first failed result |
| auto expectedResult = ResultOk; |
| failedResult.compare_exchange_strong(expectedResult, result); |
| LOG_ERROR("Failed when subscribed to topic " << topic << " in TopicsConsumer. Error - " << result); |
| } else { |
| LOG_DEBUG("Subscribed to topic " << topic << " in TopicsConsumer "); |
| } |
| |
| if (--(*topicsNeedCreate) == 0) { |
| State state = Pending; |
| if (state_.compare_exchange_strong(state, Ready)) { |
| LOG_INFO("Successfully Subscribed to Topics"); |
| multiTopicsConsumerCreatedPromise_.setValue(get_shared_this_ptr()); |
| } else { |
| LOG_ERROR("Unable to create Consumer - " << consumerStr_ << " Error - " << result); |
| // unsubscribed all of the successfully subscribed partitioned consumers |
| // `shutdown()`, which set multiTopicsConsumerCreatedPromise_ with `failedResult`, will be called |
| // when `closeAsync` completes. |
| closeAsync(nullptr); |
| } |
| } |
| } |
| |
| // subscribe for passed in topic |
| Future<Result, Consumer> MultiTopicsConsumerImpl::subscribeOneTopicAsync(const std::string& topic) { |
| TopicNamePtr topicName; |
| ConsumerSubResultPromisePtr topicPromise = std::make_shared<Promise<Result, Consumer>>(); |
| if (!(topicName = TopicName::get(topic))) { |
| LOG_ERROR("TopicName invalid: " << topic); |
| topicPromise->setFailed(ResultInvalidTopicName); |
| return topicPromise->getFuture(); |
| } |
| |
| const auto state = state_.load(); |
| if (state == Closed || state == Closing) { |
| LOG_ERROR("MultiTopicsConsumer already closed when subscribe."); |
| topicPromise->setFailed(ResultAlreadyClosed); |
| return topicPromise->getFuture(); |
| } |
| |
| // subscribe for each partition, when all partitions completed, complete promise |
| Lock lock(mutex_); |
| auto entry = topicsPartitions_.find(topic); |
| if (entry == topicsPartitions_.end()) { |
| lock.unlock(); |
| lookupServicePtr_->getPartitionMetadataAsync(topicName).addListener( |
| [this, topicName, topicPromise](Result result, const LookupDataResultPtr& lookupDataResult) { |
| if (result != ResultOk) { |
| LOG_ERROR("Error Checking/Getting Partition Metadata while MultiTopics Subscribing- " |
| << consumerStr_ << " result: " << result) |
| topicPromise->setFailed(result); |
| return; |
| } |
| subscribeTopicPartitions(lookupDataResult->getPartitions(), topicName, subscriptionName_, |
| topicPromise); |
| }); |
| } else { |
| auto numPartitions = entry->second; |
| lock.unlock(); |
| subscribeTopicPartitions(numPartitions, topicName, subscriptionName_, topicPromise); |
| } |
| return topicPromise->getFuture(); |
| } |
| |
| void MultiTopicsConsumerImpl::subscribeTopicPartitions(int numPartitions, TopicNamePtr topicName, |
| const std::string& consumerName, |
| ConsumerSubResultPromisePtr topicSubResultPromise) { |
| std::shared_ptr<ConsumerImpl> consumer; |
| ConsumerConfiguration config = conf_.clone(); |
| auto client = client_.lock(); |
| if (!client) { |
| topicSubResultPromise->setFailed(ResultAlreadyClosed); |
| return; |
| } |
| ExecutorServicePtr internalListenerExecutor = client->getPartitionListenerExecutorProvider()->get(); |
| |
| auto weakSelf = weak_from_this(); |
| config.setMessageListener([this, weakSelf](Consumer consumer, const Message& msg) { |
| auto self = weakSelf.lock(); |
| if (self) { |
| messageReceived(consumer, msg); |
| } |
| }); |
| |
| int partitions = numPartitions == 0 ? 1 : numPartitions; |
| |
| // Apply total limit of receiver queue size across partitions |
| config.setReceiverQueueSize( |
| std::min(conf_.getReceiverQueueSize(), |
| (int)(conf_.getMaxTotalReceiverQueueSizeAcrossPartitions() / partitions))); |
| |
| Lock lock(mutex_); |
| topicsPartitions_[topicName->toString()] = partitions; |
| lock.unlock(); |
| numberTopicPartitions_->fetch_add(partitions); |
| |
| std::shared_ptr<std::atomic<int>> partitionsNeedCreate = std::make_shared<std::atomic<int>>(partitions); |
| |
| // non-partitioned topic |
| if (numPartitions == 0) { |
| // We don't have to add partition-n suffix |
| try { |
| consumer = std::make_shared<ConsumerImpl>(client, topicName->toString(), subscriptionName_, |
| config, topicName->isPersistent(), interceptors_, |
| internalListenerExecutor, true, NonPartitioned, |
| subscriptionMode_, startMessageId_); |
| } catch (const std::runtime_error& e) { |
| LOG_ERROR("Failed to create ConsumerImpl for " << topicName->toString() << ": " << e.what()); |
| topicSubResultPromise->setFailed(ResultConnectError); |
| return; |
| } |
| consumer->getConsumerCreatedFuture().addListener(std::bind( |
| &MultiTopicsConsumerImpl::handleSingleConsumerCreated, get_shared_this_ptr(), |
| std::placeholders::_1, std::placeholders::_2, partitionsNeedCreate, topicSubResultPromise)); |
| consumers_.emplace(topicName->toString(), consumer); |
| LOG_DEBUG("Creating Consumer for - " << topicName << " - " << consumerStr_); |
| consumer->start(); |
| |
| } else { |
| std::vector<ConsumerImplPtr> consumers; |
| for (int i = 0; i < numPartitions; i++) { |
| std::string topicPartitionName = topicName->getTopicPartitionName(i); |
| try { |
| consumer = std::make_shared<ConsumerImpl>(client, topicPartitionName, subscriptionName_, |
| config, topicName->isPersistent(), interceptors_, |
| internalListenerExecutor, true, Partitioned, |
| subscriptionMode_, startMessageId_); |
| } catch (const std::runtime_error& e) { |
| LOG_ERROR("Failed to create ConsumerImpl for " << topicPartitionName << ": " << e.what()); |
| topicSubResultPromise->setFailed(ResultConnectError); |
| return; |
| } |
| consumers.emplace_back(consumer); |
| } |
| for (size_t i = 0; i < consumers.size(); i++) { |
| std::string topicPartitionName = topicName->getTopicPartitionName(i); |
| auto&& consumer = consumers[i]; |
| consumer->getConsumerCreatedFuture().addListener(std::bind( |
| &MultiTopicsConsumerImpl::handleSingleConsumerCreated, get_shared_this_ptr(), |
| std::placeholders::_1, std::placeholders::_2, partitionsNeedCreate, topicSubResultPromise)); |
| consumer->setPartitionIndex(i); |
| consumers_.emplace(topicPartitionName, consumer); |
| LOG_DEBUG("Creating Consumer for - " << topicPartitionName << " - " << consumerStr_); |
| consumer->start(); |
| } |
| } |
| } |
| |
| void MultiTopicsConsumerImpl::handleSingleConsumerCreated( |
| Result result, ConsumerImplBaseWeakPtr consumerImplBaseWeakPtr, |
| std::shared_ptr<std::atomic<int>> partitionsNeedCreate, |
| ConsumerSubResultPromisePtr topicSubResultPromise) { |
| if (state_ == Failed) { |
| // one of the consumer creation failed, and we are cleaning up |
| topicSubResultPromise->setFailed(ResultAlreadyClosed); |
| LOG_ERROR("Unable to create Consumer " << consumerStr_ << " state == Failed, result: " << result); |
| return; |
| } |
| |
| int previous = partitionsNeedCreate->fetch_sub(1); |
| assert(previous > 0); |
| |
| if (result != ResultOk) { |
| topicSubResultPromise->setFailed(result); |
| LOG_ERROR("Unable to create Consumer - " << consumerStr_ << " Error - " << result); |
| return; |
| } |
| |
| LOG_INFO("Successfully Subscribed to a single partition of topic in TopicsConsumer. " |
| << "Partitions need to create : " << previous - 1); |
| |
| if (partitionsNeedCreate->load() == 0) { |
| if (partitionsUpdateTimer_) { |
| runPartitionUpdateTask(); |
| } |
| topicSubResultPromise->setValue(Consumer(get_shared_this_ptr())); |
| } |
| } |
| |
| void MultiTopicsConsumerImpl::unsubscribeAsync(ResultCallback originalCallback) { |
| LOG_INFO("[ Topics Consumer " << topic_ << "," << subscriptionName_ << "] Unsubscribing"); |
| |
| auto callback = [this, originalCallback](Result result) { |
| if (result == ResultOk) { |
| shutdown(); |
| LOG_INFO(getName() << "Unsubscribed successfully"); |
| } else { |
| state_ = Ready; |
| LOG_WARN(getName() << "Failed to unsubscribe: " << result); |
| } |
| if (originalCallback) { |
| originalCallback(result); |
| } |
| }; |
| |
| const auto state = state_.load(); |
| if (state == Closing || state == Closed) { |
| callback(ResultAlreadyClosed); |
| return; |
| } |
| state_ = Closing; |
| |
| std::shared_ptr<std::atomic<int>> consumerUnsubed = std::make_shared<std::atomic<int>>(0); |
| auto self = get_shared_this_ptr(); |
| int numConsumers = 0; |
| consumers_.forEachValue( |
| [&numConsumers, &consumerUnsubed, &self, callback](const ConsumerImplPtr& consumer) { |
| numConsumers++; |
| consumer->unsubscribeAsync([self, consumerUnsubed, callback](Result result) { |
| self->handleUnsubscribedAsync(result, consumerUnsubed, callback); |
| }); |
| }); |
| if (numConsumers == 0) { |
| // No need to unsubscribe, since the list matching the regex was empty |
| callback(ResultOk); |
| } |
| } |
| |
| void MultiTopicsConsumerImpl::handleUnsubscribedAsync(Result result, |
| std::shared_ptr<std::atomic<int>> consumerUnsubed, |
| ResultCallback callback) { |
| (*consumerUnsubed)++; |
| |
| if (result != ResultOk) { |
| state_ = Failed; |
| LOG_ERROR("Error Closing one of the consumers in TopicsConsumer, result: " |
| << result << " subscription - " << subscriptionName_); |
| } |
| |
| if (consumerUnsubed->load() == numberTopicPartitions_->load()) { |
| LOG_DEBUG("Unsubscribed all of the partition consumer for TopicsConsumer. - " << consumerStr_); |
| Result result1 = (state_ != Failed) ? ResultOk : ResultUnknownError; |
| // The `callback` is a wrapper of user provided callback, it's not null and will call `shutdown()` if |
| // unsubscribe succeeds. |
| callback(result1); |
| return; |
| } |
| } |
| |
| void MultiTopicsConsumerImpl::unsubscribeOneTopicAsync(const std::string& topic, ResultCallback callback) { |
| Lock lock(mutex_); |
| std::map<std::string, int>::iterator it = topicsPartitions_.find(topic); |
| if (it == topicsPartitions_.end()) { |
| lock.unlock(); |
| LOG_ERROR("TopicsConsumer does not subscribe topic : " << topic << " subscription - " |
| << subscriptionName_); |
| callback(ResultTopicNotFound); |
| return; |
| } |
| int numberPartitions = it->second; |
| lock.unlock(); |
| |
| const auto state = state_.load(); |
| if (state == Closing || state == Closed) { |
| LOG_ERROR("TopicsConsumer already closed when unsubscribe topic: " << topic << " subscription - " |
| << subscriptionName_); |
| callback(ResultAlreadyClosed); |
| return; |
| } |
| |
| TopicNamePtr topicName; |
| if (!(topicName = TopicName::get(topic))) { |
| LOG_ERROR("TopicName invalid: " << topic); |
| callback(ResultUnknownError); |
| } |
| std::shared_ptr<std::atomic<int>> consumerUnsubed = std::make_shared<std::atomic<int>>(0); |
| |
| for (int i = 0; i < numberPartitions; i++) { |
| std::string topicPartitionName = topicName->getTopicPartitionName(i); |
| auto optConsumer = consumers_.find(topicPartitionName); |
| if (!optConsumer) { |
| LOG_ERROR("TopicsConsumer not subscribed on topicPartitionName: " << topicPartitionName); |
| callback(ResultUnknownError); |
| continue; |
| } |
| |
| optConsumer.value()->unsubscribeAsync( |
| std::bind(&MultiTopicsConsumerImpl::handleOneTopicUnsubscribedAsync, get_shared_this_ptr(), |
| std::placeholders::_1, consumerUnsubed, numberPartitions, topicName, topicPartitionName, |
| callback)); |
| } |
| } |
| |
| void MultiTopicsConsumerImpl::handleOneTopicUnsubscribedAsync( |
| Result result, std::shared_ptr<std::atomic<int>> consumerUnsubed, int numberPartitions, |
| TopicNamePtr topicNamePtr, std::string& topicPartitionName, ResultCallback callback) { |
| (*consumerUnsubed)++; |
| |
| if (result != ResultOk) { |
| state_ = Failed; |
| LOG_ERROR("Error Closing one of the consumers in TopicsConsumer, result: " |
| << result << " topicPartitionName - " << topicPartitionName); |
| } |
| |
| LOG_DEBUG("Successfully Unsubscribed one Consumer. topicPartitionName - " << topicPartitionName); |
| |
| auto optConsumer = consumers_.remove(topicPartitionName); |
| if (optConsumer) { |
| optConsumer.value()->pauseMessageListener(); |
| } |
| |
| if (consumerUnsubed->load() == numberPartitions) { |
| LOG_DEBUG("Unsubscribed all of the partition consumer for TopicsConsumer. - " << consumerStr_); |
| std::map<std::string, int>::iterator it = topicsPartitions_.find(topicNamePtr->toString()); |
| if (it != topicsPartitions_.end()) { |
| numberTopicPartitions_->fetch_sub(numberPartitions); |
| Lock lock(mutex_); |
| topicsPartitions_.erase(it); |
| lock.unlock(); |
| } |
| if (state_ != Failed) { |
| callback(ResultOk); |
| } else { |
| callback(ResultUnknownError); |
| } |
| unAckedMessageTrackerPtr_->removeTopicMessage(topicNamePtr->toString()); |
| return; |
| } |
| } |
| |
| void MultiTopicsConsumerImpl::closeAsync(ResultCallback originalCallback) { |
| std::weak_ptr<MultiTopicsConsumerImpl> weakSelf{get_shared_this_ptr()}; |
| auto callback = [weakSelf, originalCallback](Result result) { |
| auto self = weakSelf.lock(); |
| if (self) { |
| self->shutdown(); |
| if (result != ResultOk) { |
| LOG_WARN(self->getName() << "Failed to close consumer: " << result); |
| if (result != ResultAlreadyClosed) { |
| self->state_ = Failed; |
| } |
| } |
| } |
| if (originalCallback) { |
| originalCallback(result); |
| } |
| }; |
| const auto state = state_.load(); |
| if (state == Closing || state == Closed) { |
| callback(ResultAlreadyClosed); |
| return; |
| } |
| |
| state_ = Closing; |
| |
| cancelTimers(); |
| |
| auto consumers = consumers_.move(); |
| *numberTopicPartitions_ = 0; |
| if (consumers.empty()) { |
| LOG_DEBUG("TopicsConsumer have no consumers to close " |
| << " topic" << topic_ << " subscription - " << subscriptionName_); |
| callback(ResultAlreadyClosed); |
| return; |
| } |
| |
| auto numConsumers = std::make_shared<std::atomic<size_t>>(consumers.size()); |
| for (auto&& kv : consumers) { |
| auto& name = kv.first; |
| auto& consumer = kv.second; |
| consumer->closeAsync([name, numConsumers, callback](Result result) { |
| const auto numConsumersLeft = --*numConsumers; |
| LOG_DEBUG("Closing the consumer for partition - " << name << " numConsumersLeft - " |
| << numConsumersLeft); |
| |
| if (result != ResultOk) { |
| LOG_ERROR("Closing the consumer failed for partition - " << name << " with error - " |
| << result); |
| } |
| if (numConsumersLeft == 0) { |
| callback(result); |
| } |
| }); |
| } |
| |
| // fail pending receive |
| failPendingReceiveCallback(); |
| failPendingBatchReceiveCallback(); |
| |
| // cancel timer |
| batchReceiveTimer_->cancel(); |
| } |
| |
| void MultiTopicsConsumerImpl::messageReceived(Consumer consumer, const Message& msg) { |
| LOG_DEBUG("Received Message from one of the topic - " << consumer.getTopic() |
| << " message:" << msg.getDataAsString()); |
| msg.impl_->setTopicName(consumer.impl_->topic_); |
| |
| Lock lock(pendingReceiveMutex_); |
| if (!pendingReceives_.empty()) { |
| ReceiveCallback callback = pendingReceives_.front(); |
| pendingReceives_.pop(); |
| lock.unlock(); |
| auto weakSelf = weak_from_this(); |
| listenerExecutor_->postWork([this, weakSelf, msg, callback]() { |
| auto self = weakSelf.lock(); |
| if (self) { |
| notifyPendingReceivedCallback(ResultOk, msg, callback); |
| } |
| }); |
| return; |
| } |
| |
| if (incomingMessages_.full()) { |
| lock.unlock(); |
| } |
| |
| // add message to block queue. |
| // when messages queue is full, will block listener thread on ConsumerImpl, |
| // then will not send permits to broker, will broker stop push message. |
| incomingMessages_.push(msg); |
| incomingMessagesSize_.fetch_add(msg.getLength()); |
| |
| // try trigger pending batch messages |
| Lock batchOptionLock(batchReceiveOptionMutex_); |
| if (hasEnoughMessagesForBatchReceive()) { |
| ConsumerImplBase::notifyBatchPendingReceivedCallback(); |
| } |
| batchOptionLock.unlock(); |
| |
| if (messageListener_) { |
| listenerExecutor_->postWork( |
| std::bind(&MultiTopicsConsumerImpl::internalListener, get_shared_this_ptr(), consumer)); |
| } |
| } |
| |
| void MultiTopicsConsumerImpl::internalListener(Consumer consumer) { |
| Message m; |
| incomingMessages_.pop(m); |
| try { |
| Consumer self{get_shared_this_ptr()}; |
| messageListener_(self, m); |
| messageProcessed(m); |
| } catch (const std::exception& e) { |
| LOG_ERROR("Exception thrown from listener of Partitioned Consumer" << e.what()); |
| } |
| } |
| |
| Result MultiTopicsConsumerImpl::receive(Message& msg) { |
| if (state_ != Ready) { |
| return ResultAlreadyClosed; |
| } |
| |
| if (messageListener_) { |
| LOG_ERROR("Can not receive when a listener has been set"); |
| return ResultInvalidConfiguration; |
| } |
| incomingMessages_.pop(msg); |
| messageProcessed(msg); |
| |
| return ResultOk; |
| } |
| |
| Result MultiTopicsConsumerImpl::receive(Message& msg, int timeout) { |
| if (state_ != Ready) { |
| return ResultAlreadyClosed; |
| } |
| |
| if (messageListener_) { |
| LOG_ERROR("Can not receive when a listener has been set"); |
| return ResultInvalidConfiguration; |
| } |
| |
| if (incomingMessages_.pop(msg, std::chrono::milliseconds(timeout))) { |
| messageProcessed(msg); |
| return ResultOk; |
| } else { |
| if (state_ != Ready) { |
| return ResultAlreadyClosed; |
| } |
| return ResultTimeout; |
| } |
| } |
| |
| void MultiTopicsConsumerImpl::receiveAsync(ReceiveCallback callback) { |
| Message msg; |
| |
| // fail the callback if consumer is closing or closed |
| if (state_ != Ready) { |
| callback(ResultAlreadyClosed, msg); |
| return; |
| } |
| |
| Lock lock(pendingReceiveMutex_); |
| if (incomingMessages_.pop(msg, std::chrono::milliseconds(0))) { |
| lock.unlock(); |
| messageProcessed(msg); |
| callback(ResultOk, msg); |
| } else { |
| pendingReceives_.push(callback); |
| } |
| } |
| |
| void MultiTopicsConsumerImpl::failPendingReceiveCallback() { |
| Message msg; |
| |
| incomingMessages_.close(); |
| |
| Lock lock(pendingReceiveMutex_); |
| while (!pendingReceives_.empty()) { |
| ReceiveCallback callback = pendingReceives_.front(); |
| pendingReceives_.pop(); |
| auto weakSelf = weak_from_this(); |
| listenerExecutor_->postWork([this, weakSelf, msg, callback]() { |
| auto self = weakSelf.lock(); |
| if (self) { |
| notifyPendingReceivedCallback(ResultAlreadyClosed, msg, callback); |
| } |
| }); |
| } |
| lock.unlock(); |
| } |
| |
| void MultiTopicsConsumerImpl::notifyPendingReceivedCallback(Result result, const Message& msg, |
| const ReceiveCallback& callback) { |
| if (result == ResultOk) { |
| unAckedMessageTrackerPtr_->add(msg.getMessageId()); |
| } |
| callback(result, msg); |
| } |
| |
| void MultiTopicsConsumerImpl::acknowledgeAsync(const MessageId& msgId, ResultCallback callback) { |
| if (state_ != Ready) { |
| interceptors_->onAcknowledge(Consumer(shared_from_this()), ResultAlreadyClosed, msgId); |
| callback(ResultAlreadyClosed); |
| return; |
| } |
| |
| const std::string& topicPartitionName = msgId.getTopicName(); |
| if (topicPartitionName.empty()) { |
| LOG_ERROR("MessageId without a topic name cannot be acknowledged for a multi-topics consumer"); |
| callback(ResultOperationNotSupported); |
| return; |
| } |
| auto optConsumer = consumers_.find(topicPartitionName); |
| |
| if (optConsumer) { |
| unAckedMessageTrackerPtr_->remove(msgId); |
| optConsumer.value()->acknowledgeAsync(msgId, callback); |
| } else { |
| LOG_ERROR("Message of topic: " << topicPartitionName << " not in unAckedMessageTracker"); |
| callback(ResultUnknownError); |
| } |
| } |
| |
| void MultiTopicsConsumerImpl::acknowledgeAsync(const MessageIdList& messageIdList, ResultCallback callback) { |
| if (state_ != Ready) { |
| callback(ResultAlreadyClosed); |
| return; |
| } |
| |
| std::unordered_map<std::string, MessageIdList> topicToMessageId; |
| for (const MessageId& messageId : messageIdList) { |
| auto topicName = messageId.getTopicName(); |
| if (topicName.empty()) { |
| LOG_ERROR("MessageId without a topic name cannot be acknowledged for a multi-topics consumer"); |
| callback(ResultOperationNotSupported); |
| return; |
| } |
| topicToMessageId[topicName].emplace_back(messageId); |
| } |
| |
| auto needCallBack = std::make_shared<std::atomic<int>>(topicToMessageId.size()); |
| auto cb = [callback, needCallBack](Result result) { |
| if (result != ResultOk) { |
| LOG_ERROR("Filed when acknowledge list: " << result); |
| // set needCallBack is -1 to avoid repeated callback. |
| needCallBack->store(-1); |
| callback(result); |
| return; |
| } |
| if (--(*needCallBack) == 0) { |
| callback(result); |
| } |
| }; |
| for (const auto& kv : topicToMessageId) { |
| auto optConsumer = consumers_.find(kv.first); |
| if (optConsumer) { |
| unAckedMessageTrackerPtr_->remove(kv.second); |
| optConsumer.value()->acknowledgeAsync(kv.second, cb); |
| } else { |
| LOG_ERROR("Message of topic: " << kv.first << " not in consumers"); |
| callback(ResultUnknownError); |
| } |
| } |
| } |
| |
| void MultiTopicsConsumerImpl::acknowledgeCumulativeAsync(const MessageId& msgId, ResultCallback callback) { |
| msgId.getTopicName(); |
| auto optConsumer = consumers_.find(msgId.getTopicName()); |
| if (optConsumer) { |
| unAckedMessageTrackerPtr_->removeMessagesTill(msgId); |
| optConsumer.value()->acknowledgeCumulativeAsync(msgId, callback); |
| } |
| } |
| |
| void MultiTopicsConsumerImpl::negativeAcknowledge(const MessageId& msgId) { |
| auto optConsumer = consumers_.find(msgId.getTopicName()); |
| |
| if (optConsumer) { |
| unAckedMessageTrackerPtr_->remove(msgId); |
| optConsumer.value()->negativeAcknowledge(msgId); |
| } |
| } |
| |
| MultiTopicsConsumerImpl::~MultiTopicsConsumerImpl() { shutdown(); } |
| |
| Future<Result, ConsumerImplBaseWeakPtr> MultiTopicsConsumerImpl::getConsumerCreatedFuture() { |
| return multiTopicsConsumerCreatedPromise_.getFuture(); |
| } |
| const std::string& MultiTopicsConsumerImpl::getSubscriptionName() const { return subscriptionName_; } |
| |
| const std::string& MultiTopicsConsumerImpl::getTopic() const { return *topic_; } |
| |
| const std::string& MultiTopicsConsumerImpl::getName() const { return consumerStr_; } |
| |
| void MultiTopicsConsumerImpl::shutdown() { |
| cancelTimers(); |
| incomingMessages_.clear(); |
| topicsPartitions_.clear(); |
| unAckedMessageTrackerPtr_->clear(); |
| interceptors_->close(); |
| auto client = client_.lock(); |
| if (client) { |
| client->cleanupConsumer(this); |
| } |
| consumers_.clear(); |
| topicsPartitions_.clear(); |
| if (failedResult != ResultOk) { |
| multiTopicsConsumerCreatedPromise_.setFailed(failedResult); |
| } else { |
| multiTopicsConsumerCreatedPromise_.setFailed(ResultAlreadyClosed); |
| } |
| state_ = Closed; |
| } |
| |
| bool MultiTopicsConsumerImpl::isClosed() { return state_ == Closed; } |
| |
| bool MultiTopicsConsumerImpl::isOpen() { return state_ == Ready; } |
| |
| void MultiTopicsConsumerImpl::receiveMessages() { |
| const auto receiverQueueSize = conf_.getReceiverQueueSize(); |
| consumers_.forEachValue([receiverQueueSize](const ConsumerImplPtr& consumer) { |
| consumer->sendFlowPermitsToBroker(consumer->getCnx().lock(), receiverQueueSize); |
| LOG_DEBUG("Sending FLOW command for consumer - " << consumer->getConsumerId()); |
| }); |
| } |
| |
| Result MultiTopicsConsumerImpl::pauseMessageListener() { |
| if (!messageListener_) { |
| return ResultInvalidConfiguration; |
| } |
| consumers_.forEachValue([](const ConsumerImplPtr& consumer) { consumer->pauseMessageListener(); }); |
| return ResultOk; |
| } |
| |
| Result MultiTopicsConsumerImpl::resumeMessageListener() { |
| if (!messageListener_) { |
| return ResultInvalidConfiguration; |
| } |
| consumers_.forEachValue([](const ConsumerImplPtr& consumer) { consumer->resumeMessageListener(); }); |
| return ResultOk; |
| } |
| |
| void MultiTopicsConsumerImpl::redeliverUnacknowledgedMessages() { |
| LOG_DEBUG("Sending RedeliverUnacknowledgedMessages command for partitioned consumer."); |
| consumers_.forEachValue( |
| [](const ConsumerImplPtr& consumer) { consumer->redeliverUnacknowledgedMessages(); }); |
| unAckedMessageTrackerPtr_->clear(); |
| } |
| |
| void MultiTopicsConsumerImpl::redeliverUnacknowledgedMessages(const std::set<MessageId>& messageIds) { |
| if (messageIds.empty()) { |
| return; |
| } |
| if (conf_.getConsumerType() != ConsumerShared && conf_.getConsumerType() != ConsumerKeyShared) { |
| redeliverUnacknowledgedMessages(); |
| return; |
| } |
| |
| LOG_DEBUG("Sending RedeliverUnacknowledgedMessages command for partitioned consumer."); |
| std::unordered_map<std::string, std::set<MessageId>> topicToMessageId; |
| for (const MessageId& messageId : messageIds) { |
| auto topicName = messageId.getTopicName(); |
| topicToMessageId[topicName].emplace(messageId); |
| } |
| |
| for (const auto& kv : topicToMessageId) { |
| auto optConsumer = consumers_.find(kv.first); |
| if (optConsumer) { |
| optConsumer.value()->redeliverUnacknowledgedMessages(kv.second); |
| } else { |
| LOG_ERROR("Message of topic: " << kv.first << " not in consumers"); |
| } |
| } |
| } |
| |
| int MultiTopicsConsumerImpl::getNumOfPrefetchedMessages() const { return incomingMessages_.size(); } |
| |
| void MultiTopicsConsumerImpl::getBrokerConsumerStatsAsync(BrokerConsumerStatsCallback callback) { |
| if (state_ != Ready) { |
| callback(ResultConsumerNotInitialized, BrokerConsumerStats()); |
| return; |
| } |
| Lock lock(mutex_); |
| MultiTopicsBrokerConsumerStatsPtr statsPtr = |
| std::make_shared<MultiTopicsBrokerConsumerStatsImpl>(numberTopicPartitions_->load()); |
| LatchPtr latchPtr = std::make_shared<Latch>(numberTopicPartitions_->load()); |
| lock.unlock(); |
| |
| size_t i = 0; |
| consumers_.forEachValue([this, &latchPtr, &statsPtr, &i, callback](const ConsumerImplPtr& consumer) { |
| size_t index = i++; |
| auto weakSelf = weak_from_this(); |
| consumer->getBrokerConsumerStatsAsync( |
| [this, weakSelf, latchPtr, statsPtr, index, callback](Result result, BrokerConsumerStats stats) { |
| auto self = weakSelf.lock(); |
| if (self) { |
| handleGetConsumerStats(result, stats, latchPtr, statsPtr, index, callback); |
| } |
| }); |
| }); |
| } |
| |
| void MultiTopicsConsumerImpl::getLastMessageIdAsync(BrokerGetLastMessageIdCallback callback) { |
| callback(ResultOperationNotSupported, GetLastMessageIdResponse()); |
| } |
| |
| void MultiTopicsConsumerImpl::handleGetConsumerStats(Result res, BrokerConsumerStats brokerConsumerStats, |
| LatchPtr latchPtr, |
| MultiTopicsBrokerConsumerStatsPtr statsPtr, size_t index, |
| BrokerConsumerStatsCallback callback) { |
| Lock lock(mutex_); |
| if (res == ResultOk) { |
| latchPtr->countdown(); |
| statsPtr->add(brokerConsumerStats, index); |
| } else { |
| lock.unlock(); |
| callback(res, BrokerConsumerStats()); |
| return; |
| } |
| if (latchPtr->getCount() == 0) { |
| lock.unlock(); |
| callback(ResultOk, BrokerConsumerStats(statsPtr)); |
| } |
| } |
| |
| std::shared_ptr<TopicName> MultiTopicsConsumerImpl::topicNamesValid(const std::vector<std::string>& topics) { |
| TopicNamePtr topicNamePtr = std::shared_ptr<TopicName>(); |
| |
| // all topics name valid, and all topics have same namespace |
| for (std::vector<std::string>::const_iterator itr = topics.begin(); itr != topics.end(); itr++) { |
| // topic name valid |
| if (!(topicNamePtr = TopicName::get(*itr))) { |
| LOG_ERROR("Topic name invalid when init " << *itr); |
| return std::shared_ptr<TopicName>(); |
| } |
| } |
| |
| return topicNamePtr; |
| } |
| |
| void MultiTopicsConsumerImpl::seekAsync(const MessageId& msgId, ResultCallback callback) { |
| callback(ResultOperationNotSupported); |
| } |
| |
| void MultiTopicsConsumerImpl::seekAsync(uint64_t timestamp, ResultCallback callback) { |
| if (state_ != Ready) { |
| callback(ResultAlreadyClosed); |
| return; |
| } |
| |
| MultiResultCallback multiResultCallback(callback, consumers_.size()); |
| consumers_.forEachValue([×tamp, &multiResultCallback](ConsumerImplPtr consumer) { |
| consumer->seekAsync(timestamp, multiResultCallback); |
| }); |
| } |
| |
| void MultiTopicsConsumerImpl::setNegativeAcknowledgeEnabledForTesting(bool enabled) { |
| consumers_.forEachValue([enabled](const ConsumerImplPtr& consumer) { |
| consumer->setNegativeAcknowledgeEnabledForTesting(enabled); |
| }); |
| } |
| |
| bool MultiTopicsConsumerImpl::isConnected() const { |
| if (state_ != Ready) { |
| return false; |
| } |
| |
| return !consumers_.findFirstValueIf( |
| [](const ConsumerImplPtr& consumer) { return !consumer->isConnected(); }); |
| } |
| |
| uint64_t MultiTopicsConsumerImpl::getNumberOfConnectedConsumer() { |
| uint64_t numberOfConnectedConsumer = 0; |
| consumers_.forEachValue([&numberOfConnectedConsumer](const ConsumerImplPtr& consumer) { |
| if (consumer->isConnected()) { |
| numberOfConnectedConsumer++; |
| } |
| }); |
| return numberOfConnectedConsumer; |
| } |
| void MultiTopicsConsumerImpl::runPartitionUpdateTask() { |
| partitionsUpdateTimer_->expires_from_now(partitionsUpdateInterval_); |
| auto weakSelf = weak_from_this(); |
| partitionsUpdateTimer_->async_wait([weakSelf](const boost::system::error_code& ec) { |
| // If two requests call runPartitionUpdateTask at the same time, the timer will fail, and it |
| // cannot continue at this time, and the request needs to be ignored. |
| auto self = weakSelf.lock(); |
| if (self && !ec) { |
| self->topicPartitionUpdate(); |
| } |
| }); |
| } |
| void MultiTopicsConsumerImpl::topicPartitionUpdate() { |
| using namespace std::placeholders; |
| Lock lock(mutex_); |
| auto topicsPartitions = topicsPartitions_; |
| lock.unlock(); |
| for (const auto& item : topicsPartitions) { |
| auto topicName = TopicName::get(item.first); |
| auto currentNumPartitions = item.second; |
| auto weakSelf = weak_from_this(); |
| lookupServicePtr_->getPartitionMetadataAsync(topicName).addListener( |
| [this, weakSelf, topicName, currentNumPartitions](Result result, |
| const LookupDataResultPtr& lookupDataResult) { |
| auto self = weakSelf.lock(); |
| if (self) { |
| this->handleGetPartitions(topicName, result, lookupDataResult, currentNumPartitions); |
| } |
| }); |
| } |
| } |
| void MultiTopicsConsumerImpl::handleGetPartitions(TopicNamePtr topicName, Result result, |
| const LookupDataResultPtr& lookupDataResult, |
| int currentNumPartitions) { |
| if (state_ != Ready) { |
| return; |
| } |
| if (!result) { |
| const auto newNumPartitions = static_cast<unsigned int>(lookupDataResult->getPartitions()); |
| if (newNumPartitions > currentNumPartitions) { |
| LOG_INFO("new partition count: " << newNumPartitions |
| << " current partition count: " << currentNumPartitions); |
| auto partitionsNeedCreate = |
| std::make_shared<std::atomic<int>>(newNumPartitions - currentNumPartitions); |
| ConsumerSubResultPromisePtr topicPromise = std::make_shared<Promise<Result, Consumer>>(); |
| Lock lock(mutex_); |
| topicsPartitions_[topicName->toString()] = newNumPartitions; |
| lock.unlock(); |
| numberTopicPartitions_->fetch_add(newNumPartitions - currentNumPartitions); |
| for (unsigned int i = currentNumPartitions; i < newNumPartitions; i++) { |
| subscribeSingleNewConsumer(newNumPartitions, topicName, i, topicPromise, |
| partitionsNeedCreate); |
| } |
| // `runPartitionUpdateTask()` will be called in `handleSingleConsumerCreated()` |
| return; |
| } |
| } else { |
| LOG_WARN("Failed to getPartitionMetadata: " << strResult(result)); |
| } |
| runPartitionUpdateTask(); |
| } |
| |
| void MultiTopicsConsumerImpl::subscribeSingleNewConsumer( |
| int numPartitions, TopicNamePtr topicName, int partitionIndex, |
| ConsumerSubResultPromisePtr topicSubResultPromise, |
| std::shared_ptr<std::atomic<int>> partitionsNeedCreate) { |
| ConsumerConfiguration config = conf_.clone(); |
| auto client = client_.lock(); |
| if (!client) { |
| topicSubResultPromise->setFailed(ResultAlreadyClosed); |
| return; |
| } |
| ExecutorServicePtr internalListenerExecutor = client->getPartitionListenerExecutorProvider()->get(); |
| auto weakSelf = weak_from_this(); |
| config.setMessageListener([this, weakSelf](Consumer consumer, const Message& msg) { |
| auto self = weakSelf.lock(); |
| if (self) { |
| messageReceived(consumer, msg); |
| } |
| }); |
| |
| // Apply total limit of receiver queue size across partitions |
| config.setReceiverQueueSize( |
| std::min(conf_.getReceiverQueueSize(), |
| (int)(conf_.getMaxTotalReceiverQueueSizeAcrossPartitions() / numPartitions))); |
| |
| std::string topicPartitionName = topicName->getTopicPartitionName(partitionIndex); |
| |
| auto consumer = std::make_shared<ConsumerImpl>( |
| client, topicPartitionName, subscriptionName_, config, topicName->isPersistent(), interceptors_, |
| internalListenerExecutor, true, Partitioned, subscriptionMode_, startMessageId_); |
| consumer->getConsumerCreatedFuture().addListener( |
| [this, weakSelf, partitionsNeedCreate, topicSubResultPromise]( |
| Result result, const ConsumerImplBaseWeakPtr& consumerImplBaseWeakPtr) { |
| auto self = weakSelf.lock(); |
| if (self) { |
| handleSingleConsumerCreated(result, consumerImplBaseWeakPtr, partitionsNeedCreate, |
| topicSubResultPromise); |
| } |
| }); |
| consumer->setPartitionIndex(partitionIndex); |
| consumer->start(); |
| consumers_.emplace(topicPartitionName, consumer); |
| LOG_INFO("Add Creating Consumer for - " << topicPartitionName << " - " << consumerStr_ |
| << " consumerSize: " << consumers_.size()); |
| } |
| |
| bool MultiTopicsConsumerImpl::hasEnoughMessagesForBatchReceive() const { |
| if (batchReceivePolicy_.getMaxNumMessages() <= 0 && batchReceivePolicy_.getMaxNumBytes() <= 0) { |
| return false; |
| } |
| return (batchReceivePolicy_.getMaxNumMessages() > 0 && |
| incomingMessages_.size() >= batchReceivePolicy_.getMaxNumMessages()) || |
| (batchReceivePolicy_.getMaxNumBytes() > 0 && |
| incomingMessagesSize_ >= batchReceivePolicy_.getMaxNumBytes()); |
| } |
| |
| void MultiTopicsConsumerImpl::notifyBatchPendingReceivedCallback(const BatchReceiveCallback& callback) { |
| auto messages = std::make_shared<MessagesImpl>(batchReceivePolicy_.getMaxNumMessages(), |
| batchReceivePolicy_.getMaxNumBytes()); |
| Message msg; |
| while (incomingMessages_.popIf( |
| msg, [&messages](const Message& peekMsg) { return messages->canAdd(peekMsg); })) { |
| messageProcessed(msg); |
| messages->add(msg); |
| } |
| auto weakSelf = weak_from_this(); |
| listenerExecutor_->postWork([weakSelf, callback, messages]() { |
| auto self = weakSelf.lock(); |
| if (self) { |
| callback(ResultOk, messages->getMessageList()); |
| } |
| }); |
| } |
| |
| void MultiTopicsConsumerImpl::messageProcessed(Message& msg) { |
| incomingMessagesSize_.fetch_sub(msg.getLength()); |
| unAckedMessageTrackerPtr_->add(msg.getMessageId()); |
| } |
| |
| std::shared_ptr<MultiTopicsConsumerImpl> MultiTopicsConsumerImpl::get_shared_this_ptr() { |
| return std::dynamic_pointer_cast<MultiTopicsConsumerImpl>(shared_from_this()); |
| } |
| |
| void MultiTopicsConsumerImpl::beforeConnectionChange(ClientConnection& cnx) { |
| throw std::runtime_error("The connection_ field should not be modified for a MultiTopicsConsumerImpl"); |
| } |
| |
| void MultiTopicsConsumerImpl::cancelTimers() noexcept { |
| if (partitionsUpdateTimer_) { |
| boost::system::error_code ec; |
| partitionsUpdateTimer_->cancel(ec); |
| } |
| } |
| |
| void MultiTopicsConsumerImpl::hasMessageAvailableAsync(HasMessageAvailableCallback callback) { |
| if (incomingMessagesSize_ > 0) { |
| callback(ResultOk, true); |
| return; |
| } |
| |
| auto hasMessageAvailable = std::make_shared<std::atomic<bool>>(); |
| auto needCallBack = std::make_shared<std::atomic<int>>(consumers_.size()); |
| auto self = get_shared_this_ptr(); |
| |
| consumers_.forEachValue([self, needCallBack, callback, hasMessageAvailable](ConsumerImplPtr consumer) { |
| consumer->hasMessageAvailableAsync( |
| [self, needCallBack, callback, hasMessageAvailable](Result result, bool hasMsg) { |
| if (result != ResultOk) { |
| LOG_ERROR("Filed when acknowledge list: " << result); |
| // set needCallBack is -1 to avoid repeated callback. |
| needCallBack->store(-1); |
| callback(result, false); |
| return; |
| } |
| |
| if (hasMsg) { |
| hasMessageAvailable->store(hasMsg); |
| } |
| |
| if (--(*needCallBack) == 0) { |
| callback(result, hasMessageAvailable->load() || self->incomingMessagesSize_ > 0); |
| } |
| }); |
| }); |
| } |
