Skip to content

[C++] Remove namespace check for MultiTopicsConsumerImpl #9520

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
sijie merged 2 commits into from
Feb 8, 2021
Merged

[C++] Remove namespace check for MultiTopicsConsumerImpl #9520

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
86b7759
Remove namespace check for MultiTopicsConsumerImpl
BewareMyPower Feb 7, 2021
e82ee7c
Test topics in different namespace for Python client
BewareMyPower Feb 7, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 0 additions & 21 deletions pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,6 @@ MultiTopicsConsumerImpl::MultiTopicsConsumerImpl(ClientImplPtr client, const std
listenerExecutor_(client->getListenerExecutorProvider()->get()),
messageListener_(conf.getMessageListener()),
pendingReceives_(),
namespaceName_(topicName ? topicName->getNamespaceName() : std::shared_ptr<NamespaceName>()),
lookupServicePtr_(lookupServicePtr),
numberTopicPartitions_(std::make_shared<std::atomic<int>>(0)),
topics_(topics) {
Expand Down Expand Up @@ -97,9 +96,6 @@ void MultiTopicsConsumerImpl::handleOneTopicSubscribed(Result result, Consumer c
if (topicsNeedCreate->load() == 0) {
if (compareAndSetState(Pending, Ready)) {
LOG_INFO("Successfully Subscribed to Topics");
if (!namespaceName_) {
namespaceName_ = TopicName::get(topic)->getNamespaceName();
}
multiTopicsConsumerCreatedPromise_.setValue(shared_from_this());
} else {
LOG_ERROR("Unable to create Consumer - " << consumerStr_ << " Error - " << result);
Expand All @@ -122,13 +118,6 @@ Future<Result, Consumer> MultiTopicsConsumerImpl::subscribeOneTopicAsync(const s
return topicPromise->getFuture();
}

if (namespaceName_ && !(*namespaceName_ == *(topicName->getNamespaceName()))) {
LOG_ERROR("TopicName namespace not the same with topicsConsumer. wanted namespace: "
<< namespaceName_->toString() << " this topic: " << topic);
topicPromise->setFailed(ResultInvalidTopicName);
return topicPromise->getFuture();
}

if (state_ == Closed || state_ == Closing) {
LOG_ERROR("MultiTopicsConsumer already closed when subscribe.");
topicPromise->setFailed(ResultAlreadyClosed);
Expand Down Expand Up @@ -719,7 +708,6 @@ void MultiTopicsConsumerImpl::handleGetConsumerStats(Result res, BrokerConsumerS

std::shared_ptr<TopicName> MultiTopicsConsumerImpl::topicNamesValid(const std::vector<std::string>& topics) {
TopicNamePtr topicNamePtr = std::shared_ptr<TopicName>();
NamespaceNamePtr namespaceNamePtr = std::shared_ptr<NamespaceName>();

// all topics name valid, and all topics have same namespace
for (std::vector<std::string>::const_iterator itr = topics.begin(); itr != topics.end(); itr++) {
Expand All @@ -728,15 +716,6 @@ std::shared_ptr<TopicName> MultiTopicsConsumerImpl::topicNamesValid(const std::v
LOG_ERROR("Topic name invalid when init " << *itr);
return std::shared_ptr<TopicName>();
}

// all contains same namespace part
if (!namespaceNamePtr) {
namespaceNamePtr = topicNamePtr->getNamespaceName();
} else if (!(*namespaceNamePtr == *(topicNamePtr->getNamespaceName()))) {
LOG_ERROR("Different namespace name. expected: " << namespaceNamePtr->toString() << " now:"
<< topicNamePtr->getNamespaceName()->toString());
return std::shared_ptr<TopicName>();
}
}

return topicNamePtr;
Expand Down
1 change: 0 additions & 1 deletion pulsar-client-cpp/lib/MultiTopicsConsumerImpl.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -90,7 +90,6 @@ class MultiTopicsConsumerImpl : public ConsumerImplBase,
const std::string subscriptionName_;
std::string consumerStr_;
std::string topic_;
NamespaceNamePtr namespaceName_;
const ConsumerConfiguration conf_;
typedef std::map<std::string, ConsumerImplPtr> ConsumerMap;
ConsumerMap consumers_;
Expand Down
4 changes: 3 additions & 1 deletion pulsar-client-cpp/lib/PatternMultiTopicsConsumerImpl.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,9 @@ PatternMultiTopicsConsumerImpl::PatternMultiTopicsConsumerImpl(ClientImplPtr cli
patternString_(pattern),
pattern_(std::regex(pattern)),
autoDiscoveryTimer_(),
autoDiscoveryRunning_(false) {}
autoDiscoveryRunning_(false) {
namespaceName_ = TopicName::get(pattern)->getNamespaceName();
}
Comment thread
BewareMyPower marked this conversation as resolved.

const std::regex PatternMultiTopicsConsumerImpl::getPattern() { return pattern_; }

Expand Down
1 change: 1 addition & 0 deletions pulsar-client-cpp/lib/PatternMultiTopicsConsumerImpl.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,7 @@ class PatternMultiTopicsConsumerImpl : public MultiTopicsConsumerImpl {
typedef std::shared_ptr<boost::asio::deadline_timer> TimerPtr;
TimerPtr autoDiscoveryTimer_;
bool autoDiscoveryRunning_;
NamespaceNamePtr namespaceName_;

void resetAutoDiscoveryTimer();
void timerGetTopicsOfNamespace(const Result result, const NamespaceTopicsPtr topics);
Expand Down
4 changes: 2 additions & 2 deletions pulsar-client-cpp/python/pulsar_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -917,12 +917,12 @@ def test_topics_consumer(self):
client = Client(self.serviceUrl)
topic1 = 'persistent://public/default/my-python-topics-consumer-1'
topic2 = 'persistent://public/default/my-python-topics-consumer-2'
topic3 = 'persistent://public/default/my-python-topics-consumer-3'
topic3 = 'persistent://public/default-2/my-python-topics-consumer-3' # topic from different namespace
topics = [topic1, topic2, topic3]

url1 = self.adminUrl + '/admin/v2/persistent/public/default/my-python-topics-consumer-1/partitions'
url2 = self.adminUrl + '/admin/v2/persistent/public/default/my-python-topics-consumer-2/partitions'
url3 = self.adminUrl + '/admin/v2/persistent/public/default/my-python-topics-consumer-3/partitions'
url3 = self.adminUrl + '/admin/v2/persistent/public/default-2/my-python-topics-consumer-3/partitions'

doHttpPut(url1, '2')
doHttpPut(url2, '3')
Expand Down
56 changes: 32 additions & 24 deletions pulsar-client-cpp/tests/BasicEndToEndTest.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1837,33 +1837,41 @@ TEST(BasicEndToEndTest, testMultiTopicsConsumerDifferentNamespace) {
topicNames.push_back(topicName2);
topicNames.push_back(topicName3);

// call admin api to make topics partitioned
std::string url1 =
adminUrl + "admin/v2/persistent/public/default/testMultiTopicsConsumerDifferentNamespace1/partitions";
std::string url2 =
adminUrl +
"admin/v2/persistent/public/default-2/testMultiTopicsConsumerDifferentNamespace2/partitions";
std::string url3 =
adminUrl +
"admin/v2/persistent/public/default-3/testMultiTopicsConsumerDifferentNamespace3/partitions";

int res = makePutRequest(url1, "2");
ASSERT_FALSE(res != 204 && res != 409);
res = makePutRequest(url2, "3");
ASSERT_FALSE(res != 204 && res != 409);
res = makePutRequest(url3, "4");
ASSERT_FALSE(res != 204 && res != 409);
// key: message value integer, value: a pair of (topic, message id)
using MessageInfo = std::pair<std::string, MessageId>;
std::map<int, MessageInfo> messageIndexToInfo;
int index = 0;
// Produce some messages for each topic
for (const auto &topic : topicNames) {
Producer producer;
ProducerConfiguration producerConfig;
ASSERT_EQ(ResultOk, client.createProducer(topic, producerConfig, producer));

const auto message = MessageBuilder().setContent(std::to_string(index)).build();
MessageId messageId;
ASSERT_EQ(ResultOk, producer.send(message, messageId));
messageIndexToInfo[index] = std::make_pair(topic, messageId);
LOG_INFO("Send " << index << " to " << topic << ", " << messageId);

ASSERT_EQ(ResultOk, producer.close());
index++;
}

// empty topics
ConsumerConfiguration consConfig;
consConfig.setConsumerType(ConsumerShared);
consConfig.setSubscriptionInitialPosition(InitialPositionEarliest);
Consumer consumer;
Promise<Result, Consumer> consumerPromise;
client.subscribeAsync(topicNames, subName, consConfig, WaitForCallbackValue<Consumer>(consumerPromise));
Future<Result, Consumer> consumerFuture = consumerPromise.getFuture();
Result result = consumerFuture.get(consumer);
ASSERT_EQ(ResultInvalidTopicName, result);
LOG_INFO("subscribe on topics with different names should fail");
ASSERT_EQ(ResultOk, client.subscribe(topicNames, subName, consConfig, consumer));

for (int i = 0; i < index; i++) {
Message message;
ASSERT_EQ(ResultOk, consumer.receive(message, 3000));
ASSERT_EQ(ResultOk, consumer.acknowledge(message));
const int index = std::stoi(message.getDataAsString());
LOG_INFO("Receive " << index << " from " << message.getTopicName() << "," << message.getMessageId());
ASSERT_EQ(messageIndexToInfo.count(index), 1);
ASSERT_EQ(messageIndexToInfo[index], std::make_pair(message.getTopicName(), message.getMessageId()));
}

consumer.close();

client.shutdown();
Expand Down