You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2020/08/16 19:01:43 UTC
[GitHub] [pulsar] BewareMyPower opened a new pull request #7823: [pulsar-client] Avoid subscribing the same topic again
BewareMyPower opened a new pull request #7823:
URL: https://github.com/apache/pulsar/pull/7823
### Motivation
The current key of `MultiTopicsConsumerImpl.topics` is the topic name passed by user. The `topicNameValid` method checks if the name is valid and `topics` doesn't contain the key.
However, if a multi topics consumer subscribed a partition of a subscribed partitioned topic, `subscribeAsync` succeed and a new `ConsumerImpl` of the same partition was created, which is redundant.
Also, if a multi topics consumer subscribed `public/default/topic` or `persistent://public/default/topic`, while the initial subscribed topic is `topic`, the redundant consumers would be created.
### Modifications
- Use full topic name as key of `MultiTopicsConsumerImpl.topics`
- Check both full topic name and full partitioned topic name not exist in `MultiTopicsConsumerImpl.topics` when `subscribeAsync` is called
- Add a unit test for subscribing a partition of a subscribed partitioned topic
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] BewareMyPower commented on a change in pull request #7823: [pulsar-client] Avoid subscribing the same topic again
Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on a change in pull request #7823:
URL: https://github.com/apache/pulsar/pull/7823#discussion_r471320837
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
##########
@@ -720,16 +720,36 @@ private void removeExpiredMessagesFromQueue(Set<MessageId> messageIds) {
}
}
- private boolean topicNameValid(String topicName) {
- return TopicName.isValid(topicName) && !topics.containsKey(topicName);
+ private TopicName getTopicName(String topic) {
+ try {
+ return TopicName.get(topic);
+ } catch (Exception ignored) {
+ return null;
+ }
+ }
+
+ private String getFullTopicName(String topic) {
+ TopicName topicName = getTopicName(topic);
+ return (topicName != null) ? topicName.toString() : null;
+ }
+
+ private void removeTopic(String topic) {
+ String fullTopicName = getFullTopicName(topic);
+ if (fullTopicName != null) {
+ topics.remove(topic);
+ }
}
// subscribe one more given topic
public CompletableFuture<Void> subscribeAsync(String topicName, boolean createTopicIfDoesNotExist) {
- if (!topicNameValid(topicName)) {
+ TopicName topicNameInstance = getTopicName(topicName);
Review comment:
@codelipenghui Sorry I haven't noticed the later part of your review before. I didn't change the original behavior in this PR because it needs relative changes with unit tests. I'll do it soon.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] BewareMyPower commented on a change in pull request #7823: [pulsar-client] Avoid subscribing the same topic again
Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on a change in pull request #7823:
URL: https://github.com/apache/pulsar/pull/7823#discussion_r471197144
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
##########
@@ -720,16 +720,36 @@ private void removeExpiredMessagesFromQueue(Set<MessageId> messageIds) {
}
}
- private boolean topicNameValid(String topicName) {
- return TopicName.isValid(topicName) && !topics.containsKey(topicName);
+ private TopicName getTopicName(String topic) {
+ try {
+ return TopicName.get(topic);
+ } catch (Exception ignored) {
+ return null;
+ }
+ }
+
+ private String getFullTopicName(String topic) {
+ TopicName topicName = getTopicName(topic);
+ return (topicName != null) ? topicName.toString() : null;
+ }
+
+ private void removeTopic(String topic) {
+ String fullTopicName = getFullTopicName(topic);
+ if (fullTopicName != null) {
+ topics.remove(topic);
+ }
}
// subscribe one more given topic
public CompletableFuture<Void> subscribeAsync(String topicName, boolean createTopicIfDoesNotExist) {
- if (!topicNameValid(topicName)) {
+ TopicName topicNameInstance = getTopicName(topicName);
Review comment:
I validate it in line 746 to 748. Because `TopicName` could be reused, if I wrap the validation in `topicNameValid`, I have to create `TopicName` to get full topic name and partitioned topic name again. I can't think an elegant way to do it so I create a `TopicName` instance at first of `subscribeAsync` and do checks
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] BewareMyPower commented on pull request #7823: [pulsar-client] Avoid subscribing the same topic again
Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on pull request #7823:
URL: https://github.com/apache/pulsar/pull/7823#issuecomment-674623374
/pulsarbot run-failure-checks
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] BewareMyPower commented on pull request #7823: [pulsar-client] Avoid subscribing the same topic again
Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on pull request #7823:
URL: https://github.com/apache/pulsar/pull/7823#issuecomment-674815821
/pulsarbot run-failure-checks
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui merged pull request #7823: [pulsar-client] Avoid subscribing the same topic again
Posted by GitBox <gi...@apache.org>.
codelipenghui merged pull request #7823:
URL: https://github.com/apache/pulsar/pull/7823
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] BewareMyPower commented on a change in pull request #7823: [pulsar-client] Avoid subscribing the same topic again
Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on a change in pull request #7823:
URL: https://github.com/apache/pulsar/pull/7823#discussion_r471197959
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
##########
@@ -720,16 +720,36 @@ private void removeExpiredMessagesFromQueue(Set<MessageId> messageIds) {
}
}
- private boolean topicNameValid(String topicName) {
- return TopicName.isValid(topicName) && !topics.containsKey(topicName);
+ private TopicName getTopicName(String topic) {
+ try {
+ return TopicName.get(topic);
+ } catch (Exception ignored) {
+ return null;
+ }
+ }
+
+ private String getFullTopicName(String topic) {
+ TopicName topicName = getTopicName(topic);
+ return (topicName != null) ? topicName.toString() : null;
+ }
+
+ private void removeTopic(String topic) {
+ String fullTopicName = getFullTopicName(topic);
+ if (fullTopicName != null) {
+ topics.remove(topic);
+ }
Review comment:
`getTopicName` and `getFullTopicName` are just for ignoring the exceptions because we will throw a `PulsarClientException` later. `removeTopic` is to ensure that the key to remove is a full topic name.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a change in pull request #7823: [pulsar-client] Avoid subscribing the same topic again
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #7823:
URL: https://github.com/apache/pulsar/pull/7823#discussion_r471189958
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
##########
@@ -720,16 +720,36 @@ private void removeExpiredMessagesFromQueue(Set<MessageId> messageIds) {
}
}
- private boolean topicNameValid(String topicName) {
- return TopicName.isValid(topicName) && !topics.containsKey(topicName);
+ private TopicName getTopicName(String topic) {
+ try {
+ return TopicName.get(topic);
+ } catch (Exception ignored) {
+ return null;
+ }
+ }
+
+ private String getFullTopicName(String topic) {
+ TopicName topicName = getTopicName(topic);
+ return (topicName != null) ? topicName.toString() : null;
+ }
+
+ private void removeTopic(String topic) {
+ String fullTopicName = getFullTopicName(topic);
+ if (fullTopicName != null) {
+ topics.remove(topic);
+ }
}
// subscribe one more given topic
public CompletableFuture<Void> subscribeAsync(String topicName, boolean createTopicIfDoesNotExist) {
- if (!topicNameValid(topicName)) {
+ TopicName topicNameInstance = getTopicName(topicName);
Review comment:
Why remove `!topicNameValid(topicName)` ? I think we should return `Topic name not valid` when the topic name does not valid and return `Already subscribe to {}` for the duplicate topic name.
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
##########
@@ -720,16 +720,36 @@ private void removeExpiredMessagesFromQueue(Set<MessageId> messageIds) {
}
}
- private boolean topicNameValid(String topicName) {
- return TopicName.isValid(topicName) && !topics.containsKey(topicName);
+ private TopicName getTopicName(String topic) {
+ try {
+ return TopicName.get(topic);
+ } catch (Exception ignored) {
+ return null;
+ }
+ }
+
+ private String getFullTopicName(String topic) {
+ TopicName topicName = getTopicName(topic);
+ return (topicName != null) ? topicName.toString() : null;
+ }
+
+ private void removeTopic(String topic) {
+ String fullTopicName = getFullTopicName(topic);
+ if (fullTopicName != null) {
+ topics.remove(topic);
+ }
Review comment:
It seems that we do not need to add these methods?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org