You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "poorbarcode (via GitHub)" <gi...@apache.org> on 2024/06/05 19:22:27 UTC
[PR] [fix] [client] Fix pattern consumer create crash if a part of partitions of a topic have been deleted [pulsar]
poorbarcode opened a new pull request, #22854:
URL: https://github.com/apache/pulsar/pull/22854
### Motivation
#### Background
**1.Regarding the API getting topics by regexp pattern, there are two implementations:***
- `List<String> HttpLookupService.getPartitionedTopicMetadata(...)`
- `CommandGetTopicsOfNamespace BinaryProtoLookupService.getPartitionedTopicMetadata(...)`
Pulsar transferred both response types `List<String>` and `CommandGetTopicsOfNamespace` to a `GetTopicsResult` object. And discarded the partition information when doing transference. For example:
- Get a list `topic-1-partition-0`, `topic-1-partition-1`.
- The transferring operation will group them to `topic-1`.
**2.The behavior of Patten consumers**
- Before https://github.com/apache/pulsar/pull/5230, Patten consumers will try to create the missing partitions when it starts up.
- After https://github.com/apache/pulsar/pull/5230, Patten consumers only subscribe to the partitions that exist to lead the topics can be deleted automatically. In other words, it sets the variable `createTopicIfDoesNotExist` of Multi Topics Consumer to `false`.
---
#### Issue
1. When users are starting a Pattern consumer, the consumer will try to subscribe to all the partitions even if some partitions have been deleted before, and then the client crashes due to a Topic Not Exists Exception. You can reproduce the issue by the test [testConsumerAfterOnePartDeleted](https://github.com/poorbarcode/pulsar/pull/91/files#diff-c6bb7cf6a6d6f5c4673ec5837351d981c835024fbbb864f32113c05b8ab30b4eR142)
2. The Patten consumer that is started removes all partitions when one partition was deleted, even if there are still half of the partitions exist.
---
### Modifications
- Multi Topics Consumer only subscribes to the existing partitions if the config `createTopicIfDoesNotExist` is `false`.
- The Patten consumer that is started only removes the partitions that were deleted.
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
### Matching PR in forked repository
PR in forked repository: x
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Fix pattern consumer create crash if a part of partitions of a topic have been deleted [pulsar]
Posted by "mattisonchao (via GitHub)" <gi...@apache.org>.
mattisonchao commented on code in PR #22854:
URL: https://github.com/apache/pulsar/pull/22854#discussion_r1665793987
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java:
##########
@@ -585,19 +585,28 @@ private <T> CompletableFuture<Consumer<T>> patternTopicSubscribeAsync(ConsumerCo
lookup.getTopicsUnderNamespace(namespaceName, subscriptionMode, regex, null)
.thenAccept(getTopicsResult -> {
if (log.isDebugEnabled()) {
- log.debug("Get topics under namespace {}, topics.size: {},"
- + " topicsHash: {}, changed: {}, filtered: {}",
+ log.debug("Pattern consumer [{}] get topics under namespace {}, topics.size: {},"
+ + " topicsHash: {}, changed: {}, filtered: {}", conf.getSubscriptionName(),
namespaceName, getTopicsResult.getTopics().size(), getTopicsResult.getTopicsHash(),
getTopicsResult.isChanged(), getTopicsResult.isFiltered());
getTopicsResult.getTopics().forEach(topicName ->
- log.debug("Get topics under namespace {}, topic: {}", namespaceName, topicName));
+ log.debug("Pattern consumer [{}] get topics under namespace {}, topic: {}",
+ conf.getSubscriptionName(), namespaceName, topicName));
}
List<String> topicsList = getTopicsResult.getTopics();
if (!getTopicsResult.isFiltered()) {
topicsList = TopicList.filterTopics(getTopicsResult.getTopics(), pattern);
}
conf.getTopicNames().addAll(topicsList);
+
+ if (log.isInfoEnabled()) {
Review Comment:
It might make sense to use the debug level.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Fix pattern consumer create crash if a part of partitions of a topic have been deleted [pulsar]
Posted by "mattisonchao (via GitHub)" <gi...@apache.org>.
mattisonchao commented on code in PR #22854:
URL: https://github.com/apache/pulsar/pull/22854#discussion_r1665785500
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/LookupServiceTest.java:
##########
@@ -108,8 +108,8 @@ public void testGetTopicsOfGetTopicsResult(boolean isUsingHttpLookup) throws Exc
// Verify the new method "GetTopicsResult.nonPartitionedOrPartitionTopics" works as expected.
Collection<String> nonPartitionedOrPartitionTopics =
lookupService.getTopicsUnderNamespace(NamespaceName.get("public/default"),
Review Comment:
Why do we need to change this?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix] [client] Fix pattern consumer create crash if a part of partitions of a topic have been deleted [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #22854:
URL: https://github.com/apache/pulsar/pull/22854#discussion_r1643157512
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java:
##########
@@ -69,6 +69,31 @@ public class PatternMultiTopicsConsumerImpl<T> extends MultiTopicsConsumerImpl<T
private volatile Timeout recheckPatternTimeout = null;
private volatile String topicsHash;
+ /**
+ * This variable is used to make all tasks that will modify subscriptions will be executed one by one.
Review Comment:
Added a tool named `PatternConsumerUpdateQueue` to fix this issue
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Fix pattern consumer create crash if a part of partitions of a topic have been deleted [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #22854:
URL: https://github.com/apache/pulsar/pull/22854#discussion_r1668787772
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java:
##########
@@ -1573,4 +1548,51 @@ protected void setCurrentReceiverQueueSize(int newSize) {
CURRENT_RECEIVER_QUEUE_SIZE_UPDATER.set(this, newSize);
resumeReceivingFromPausedConsumersIfNeeded();
}
+
+ /**
+ * Get the exists partitions of a partitioned topic, the result does not contain the partitions which has not been
+ * created yet(in other words, the partitions that do not exist in the response of "pulsar-admin topics list").
+ * @return sorted partitions list if it is a partitioned topic; @return an empty list if it is a non-partitioned
+ * topic.
+ */
+ private CompletableFuture<List<Integer>> getExistsPartitions(String topic) {
+ TopicName topicName = TopicName.get(topic);
+ if (!topicName.isPersistent()) {
+ return FutureUtil.failedFuture(new IllegalArgumentException("The method getExistsPartitions"
+ + " does not support non-persistent topic yet."));
+ }
+ return client.getLookup().getTopicsUnderNamespace(topicName.getNamespaceObject(),
+ CommandGetTopicsOfNamespace.Mode.PERSISTENT,
+ TopicName.getPattern(topicName.getPartitionedTopicName()),
Review Comment:
Added
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Fix pattern consumer create crash if a part of partitions of a topic have been deleted [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #22854:
URL: https://github.com/apache/pulsar/pull/22854#discussion_r1668712267
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java:
##########
@@ -1573,4 +1548,51 @@ protected void setCurrentReceiverQueueSize(int newSize) {
CURRENT_RECEIVER_QUEUE_SIZE_UPDATER.set(this, newSize);
resumeReceivingFromPausedConsumersIfNeeded();
}
+
+ /**
+ * Get the exists partitions of a partitioned topic, the result does not contain the partitions which has not been
+ * created yet(in other words, the partitions that do not exist in the response of "pulsar-admin topics list").
+ * @return sorted partitions list if it is a partitioned topic; @return an empty list if it is a non-partitioned
+ * topic.
+ */
+ private CompletableFuture<List<Integer>> getExistsPartitions(String topic) {
+ TopicName topicName = TopicName.get(topic);
+ if (!topicName.isPersistent()) {
+ return FutureUtil.failedFuture(new IllegalArgumentException("The method getExistsPartitions"
+ + " does not support non-persistent topic yet."));
+ }
+ return client.getLookup().getTopicsUnderNamespace(topicName.getNamespaceObject(),
+ CommandGetTopicsOfNamespace.Mode.PERSISTENT,
+ TopicName.getPattern(topicName.getPartitionedTopicName()),
Review Comment:
Please help add a comment here that the server side will match all the partitions, e.g. the broker will return `abc-partition-0` and `abc-partition-1` if the provided pattern is `^abc$`.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Fix pattern consumer create crash if a part of partitions of a topic have been deleted [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #22854:
URL: https://github.com/apache/pulsar/pull/22854#discussion_r1665797519
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/LookupServiceTest.java:
##########
@@ -108,8 +108,8 @@ public void testGetTopicsOfGetTopicsResult(boolean isUsingHttpLookup) throws Exc
// Verify the new method "GetTopicsResult.nonPartitionedOrPartitionTopics" works as expected.
Collection<String> nonPartitionedOrPartitionTopics =
lookupService.getTopicsUnderNamespace(NamespaceName.get("public/default"),
Review Comment:
Sorry, it is a mistake. removed this change. Thanks
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix] [client] Fix pattern consumer create crash if a part of partitions of a topic have been deleted [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on PR #22854:
URL: https://github.com/apache/pulsar/pull/22854#issuecomment-2160140265
/pulsarbot rerun-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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix] [client] Fix pattern consumer create crash if a part of partitions of a topic have been deleted [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #22854:
URL: https://github.com/apache/pulsar/pull/22854#discussion_r1639672682
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java:
##########
@@ -1573,4 +1541,51 @@ protected void setCurrentReceiverQueueSize(int newSize) {
CURRENT_RECEIVER_QUEUE_SIZE_UPDATER.set(this, newSize);
resumeReceivingFromPausedConsumersIfNeeded();
}
+
+ /**
+ * Get the exists partitions of a partitioned topic, the result does not contain the partitions which has not been
+ * created yet(in other words, the partitions that do not exist in the response of "pulsar-admin topics list").
+ * @return sorted partitions list if it is a partitioned topic; @return an empty list if it is a non-partitioned
+ * topic.
+ */
+ private CompletableFuture<List<Integer>> getExistsPartitions(String topic) {
+ TopicName topicName = TopicName.get(topic);
+ if (!topicName.isPersistent()) {
+ return FutureUtil.failedFuture(new IllegalArgumentException("The method getExistsPartitions"
+ + " does not support non-persistent topic yet."));
+ }
+ return client.getLookup().getTopicsUnderNamespace(topicName.getNamespaceObject(),
+ CommandGetTopicsOfNamespace.Mode.PERSISTENT,
+ "^" + topicName.getPartitionedTopicName() + "$",
Review Comment:
Good suggestion, fixed
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Fix pattern consumer create crash if a part of partitions of a topic have been deleted [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #22854:
URL: https://github.com/apache/pulsar/pull/22854#discussion_r1665801286
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java:
##########
@@ -585,19 +585,28 @@ private <T> CompletableFuture<Consumer<T>> patternTopicSubscribeAsync(ConsumerCo
lookup.getTopicsUnderNamespace(namespaceName, subscriptionMode, regex, null)
.thenAccept(getTopicsResult -> {
if (log.isDebugEnabled()) {
- log.debug("Get topics under namespace {}, topics.size: {},"
- + " topicsHash: {}, changed: {}, filtered: {}",
+ log.debug("Pattern consumer [{}] get topics under namespace {}, topics.size: {},"
+ + " topicsHash: {}, changed: {}, filtered: {}", conf.getSubscriptionName(),
namespaceName, getTopicsResult.getTopics().size(), getTopicsResult.getTopicsHash(),
getTopicsResult.isChanged(), getTopicsResult.isFiltered());
getTopicsResult.getTopics().forEach(topicName ->
- log.debug("Get topics under namespace {}, topic: {}", namespaceName, topicName));
+ log.debug("Pattern consumer [{}] get topics under namespace {}, topic: {}",
+ conf.getSubscriptionName(), namespaceName, topicName));
}
List<String> topicsList = getTopicsResult.getTopics();
if (!getTopicsResult.isFiltered()) {
topicsList = TopicList.filterTopics(getTopicsResult.getTopics(), pattern);
}
conf.getTopicNames().addAll(topicsList);
+
+ if (log.isInfoEnabled()) {
Review Comment:
Changed all logs that related to topics changes to `Debug`, thanks
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Fix pattern consumer create crash if a part of partitions of a topic have been deleted [pulsar]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #22854:
URL: https://github.com/apache/pulsar/pull/22854#issuecomment-2208505903
## [Codecov](https://app.codecov.io/gh/apache/pulsar/pull/22854?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
Attention: Patch coverage is `80.47945%` with `57 lines` in your changes missing coverage. Please review.
> Project coverage is 73.45%. Comparing base [(`bbc6224`)](https://app.codecov.io/gh/apache/pulsar/commit/bbc62245c5ddba1de4b1e7cee4ab49334bc36277?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`af01c4f`)](https://app.codecov.io/gh/apache/pulsar/commit/af01c4f27af19a95f75c60360ea78e961c487eae?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
> Report is 435 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
[](https://app.codecov.io/gh/apache/pulsar/pull/22854?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
```diff
@@ Coverage Diff @@
## master #22854 +/- ##
============================================
- Coverage 73.57% 73.45% -0.12%
- Complexity 32624 33324 +700
============================================
Files 1877 1911 +34
Lines 139502 143274 +3772
Branches 15299 15629 +330
============================================
+ Hits 102638 105247 +2609
- Misses 28908 29980 +1072
- Partials 7956 8047 +91
```
| [Flag](https://app.codecov.io/gh/apache/pulsar/pull/22854/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [inttests](https://app.codecov.io/gh/apache/pulsar/pull/22854/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `27.53% <7.19%> (+2.95%)` | :arrow_up: |
| [systests](https://app.codecov.io/gh/apache/pulsar/pull/22854/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `24.75% <17.12%> (+0.42%)` | :arrow_up: |
| [unittests](https://app.codecov.io/gh/apache/pulsar/pull/22854/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `72.49% <80.47%> (-0.35%)` | :arrow_down: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Files](https://app.codecov.io/gh/apache/pulsar/pull/22854?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...va/org/apache/pulsar/client/impl/ConsumerBase.java](https://app.codecov.io/gh/apache/pulsar/pull/22854?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2FConsumerBase.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL0NvbnN1bWVyQmFzZS5qYXZh) | `74.39% <100.00%> (+0.26%)` | :arrow_up: |
| [...rg/apache/pulsar/client/impl/TopicListWatcher.java](https://app.codecov.io/gh/apache/pulsar/pull/22854?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2FTopicListWatcher.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL1RvcGljTGlzdFdhdGNoZXIuamF2YQ==) | `66.91% <100.00%> (-0.95%)` | :arrow_down: |
| [...g/apache/pulsar/common/lookup/GetTopicsResult.java](https://app.codecov.io/gh/apache/pulsar/pull/22854?src=pr&el=tree&filepath=pulsar-common%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fcommon%2Flookup%2FGetTopicsResult.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNvbW1vbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NvbW1vbi9sb29rdXAvR2V0VG9waWNzUmVzdWx0LmphdmE=) | `92.50% <100.00%> (+30.00%)` | :arrow_up: |
| [...ava/org/apache/pulsar/common/naming/TopicName.java](https://app.codecov.io/gh/apache/pulsar/pull/22854?src=pr&el=tree&filepath=pulsar-common%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fcommon%2Fnaming%2FTopicName.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNvbW1vbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NvbW1vbi9uYW1pbmcvVG9waWNOYW1lLmphdmE=) | `96.09% <100.00%> (+0.03%)` | :arrow_up: |
| [...java/org/apache/pulsar/common/util/FutureUtil.java](https://app.codecov.io/gh/apache/pulsar/pull/22854?src=pr&el=tree&filepath=pulsar-common%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fcommon%2Futil%2FFutureUtil.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNvbW1vbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NvbW1vbi91dGlsL0Z1dHVyZVV0aWwuamF2YQ==) | `76.61% <50.00%> (+2.06%)` | :arrow_up: |
| [...rg/apache/pulsar/client/impl/PulsarClientImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/22854?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2FPulsarClientImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL1B1bHNhckNsaWVudEltcGwuamF2YQ==) | `74.29% <37.50%> (-0.02%)` | :arrow_down: |
| [...ar/client/impl/PatternMultiTopicsConsumerImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/22854?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2FPatternMultiTopicsConsumerImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL1BhdHRlcm5NdWx0aVRvcGljc0NvbnN1bWVySW1wbC5qYXZh) | `87.28% <86.86%> (+3.82%)` | :arrow_up: |
| [...he/pulsar/client/impl/MultiTopicsConsumerImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/22854?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2FMultiTopicsConsumerImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL011bHRpVG9waWNzQ29uc3VtZXJJbXBsLmphdmE=) | `77.69% <71.66%> (-0.03%)` | :arrow_down: |
| [...pulsar/client/impl/PatternConsumerUpdateQueue.java](https://app.codecov.io/gh/apache/pulsar/pull/22854?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2FPatternConsumerUpdateQueue.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL1BhdHRlcm5Db25zdW1lclVwZGF0ZVF1ZXVlLmphdmE=) | `79.61% <79.61%> (ø)` | |
... and [479 files with indirect coverage changes](https://app.codecov.io/gh/apache/pulsar/pull/22854/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Fix pattern consumer create crash if a part of partitions of a topic have been deleted [pulsar]
Posted by "mattisonchao (via GitHub)" <gi...@apache.org>.
mattisonchao commented on code in PR #22854:
URL: https://github.com/apache/pulsar/pull/22854#discussion_r1665793987
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java:
##########
@@ -585,19 +585,28 @@ private <T> CompletableFuture<Consumer<T>> patternTopicSubscribeAsync(ConsumerCo
lookup.getTopicsUnderNamespace(namespaceName, subscriptionMode, regex, null)
.thenAccept(getTopicsResult -> {
if (log.isDebugEnabled()) {
- log.debug("Get topics under namespace {}, topics.size: {},"
- + " topicsHash: {}, changed: {}, filtered: {}",
+ log.debug("Pattern consumer [{}] get topics under namespace {}, topics.size: {},"
+ + " topicsHash: {}, changed: {}, filtered: {}", conf.getSubscriptionName(),
namespaceName, getTopicsResult.getTopics().size(), getTopicsResult.getTopicsHash(),
getTopicsResult.isChanged(), getTopicsResult.isFiltered());
getTopicsResult.getTopics().forEach(topicName ->
- log.debug("Get topics under namespace {}, topic: {}", namespaceName, topicName));
+ log.debug("Pattern consumer [{}] get topics under namespace {}, topic: {}",
+ conf.getSubscriptionName(), namespaceName, topicName));
}
List<String> topicsList = getTopicsResult.getTopics();
if (!getTopicsResult.isFiltered()) {
topicsList = TopicList.filterTopics(getTopicsResult.getTopics(), pattern);
}
conf.getTopicNames().addAll(topicsList);
+
+ if (log.isInfoEnabled()) {
Review Comment:
It might be debug level is making sense.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix] [client] Fix pattern consumer create crash if a part of partitions of a topic have been deleted [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #22854:
URL: https://github.com/apache/pulsar/pull/22854#discussion_r1628923187
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java:
##########
@@ -1067,30 +1074,43 @@ private void doSubscribeTopicPartitions(Schema<T> schema,
ConsumerConfigurationData<T> configurationData = getInternalConsumerConfig();
configurationData.setReceiverQueueSize(receiverQueueSize);
- futureList = IntStream
- .range(0, numPartitions)
- .mapToObj(
- partitionIndex -> {
- String partitionName = TopicName.get(topicName).getPartition(partitionIndex).toString();
- CompletableFuture<Consumer<T>> subFuture = new CompletableFuture<>();
- configurationData.setStartPaused(paused);
- ConsumerImpl<T> newConsumer = createInternalConsumer(configurationData, partitionName,
- partitionIndex, subFuture, createIfDoesNotExist, schema);
- synchronized (pauseMutex) {
- if (paused) {
- newConsumer.pause();
- } else {
- newConsumer.resume();
- }
- consumers.putIfAbsent(newConsumer.getTopic(), newConsumer);
+ CompletableFuture<List<Integer>> partitionsFuture;
+ if (createIfDoesNotExist || !TopicName.get(topicName).isPersistent()) {
+ partitionsFuture = CompletableFuture.completedFuture(IntStream.range(0, numPartitions)
+ .mapToObj(i -> Integer.valueOf(i))
+ .collect(Collectors.toList()));
+ } else {
+ partitionsFuture = getExistsPartitions(topicName.toString());
+ }
+ subscribeAllPartitionsFuture = partitionsFuture.thenCompose(partitions -> {
+ if (partitions.isEmpty()) {
+ partitionedTopics.remove(topicName, numPartitions);
+ return CompletableFuture.completedFuture(null);
+ }
+ List<CompletableFuture<Consumer<T>>> subscribeList = new ArrayList<>();
+ for (int partitionIndex : partitions) {
+ String partitionName = TopicName.get(topicName).getPartition(partitionIndex).toString();
+ CompletableFuture<Consumer<T>> subFuture = new CompletableFuture<>();
+ configurationData.setStartPaused(paused);
+ ConsumerImpl<T> newConsumer = createInternalConsumer(configurationData, partitionName,
+ partitionIndex, subFuture, createIfDoesNotExist, schema);
+ synchronized (pauseMutex) {
+ if (paused) {
+ newConsumer.pause();
+ } else {
+ newConsumer.resume();
}
- return subFuture;
- })
- .collect(Collectors.toList());
+ consumers.putIfAbsent(newConsumer.getTopic(), newConsumer);
Review Comment:
What if there's an existing entry for the key? does the `newConsumer` have to be closed in that case?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix] [client] Fix pattern consumer create crash if a part of partitions of a topic have been deleted [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #22854:
URL: https://github.com/apache/pulsar/pull/22854#discussion_r1628929247
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java:
##########
@@ -69,6 +69,31 @@ public class PatternMultiTopicsConsumerImpl<T> extends MultiTopicsConsumerImpl<T
private volatile Timeout recheckPatternTimeout = null;
private volatile String topicsHash;
+ /**
+ * This variable is used to make all tasks that will modify subscriptions will be executed one by one.
Review Comment:
Answering the memory leak question myself: it seems that memory leaks with chaining have been addressed a long time ago in the JVM: https://stackoverflow.com/a/73157976 . It should be fine to have the solution.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix] [client] Fix pattern consumer create crash if a part of partitions of a topic have been deleted [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #22854:
URL: https://github.com/apache/pulsar/pull/22854#discussion_r1628361663
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java:
##########
@@ -1573,4 +1541,51 @@ protected void setCurrentReceiverQueueSize(int newSize) {
CURRENT_RECEIVER_QUEUE_SIZE_UPDATER.set(this, newSize);
resumeReceivingFromPausedConsumersIfNeeded();
}
+
+ /**
+ * Get the exists partitions of a partitioned topic, the result does not contain the partitions which has not been
+ * created yet(in other words, the partitions that do not exist in the response of "pulsar-admin topics list").
+ * @return sorted partitions list if it is a partitioned topic; @return an empty list if it is a non-partitioned
+ * topic.
+ */
+ private CompletableFuture<List<Integer>> getExistsPartitions(String topic) {
+ TopicName topicName = TopicName.get(topic);
+ if (!topicName.isPersistent()) {
+ return FutureUtil.failedFuture(new IllegalArgumentException("The method getExistsPartitions"
+ + " does not support non-persistent topic yet."));
+ }
+ return client.getLookup().getTopicsUnderNamespace(topicName.getNamespaceObject(),
+ CommandGetTopicsOfNamespace.Mode.PERSISTENT,
+ "^" + topicName.getPartitionedTopicName() + "$",
Review Comment:
If this is intended to be an exact match, you might want to use Pattern.quote to escape any regex pattern characters in the 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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix] [client] Fix pattern consumer create crash if a part of partitions of a topic have been deleted [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #22854:
URL: https://github.com/apache/pulsar/pull/22854#discussion_r1628920203
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java:
##########
@@ -69,6 +69,31 @@ public class PatternMultiTopicsConsumerImpl<T> extends MultiTopicsConsumerImpl<T
private volatile Timeout recheckPatternTimeout = null;
private volatile String topicsHash;
+ /**
+ * This variable is used to make all tasks that will modify subscriptions will be executed one by one.
Review Comment:
Interesting solution. Does the current solution of chaining the futures introduce a memory leak?
One possible solution for limiting concurrency of asynchronous calls is to use [ConcurrencyReducer from com.spotify:completable-futures](https://github.com/spotify/completable-futures/blob/master/src/main/java/com/spotify/futures/ConcurrencyReducer.java).
In this case, limiting concurrency would be useful for a few locations where all asynchronous operations happen concurrently. For example, the code where `consumer.closeAsync()` is called happens all at once.
It might not be a problem in this case since binary protocol calls will go on the same set of connections. For Pulsar Admin API calls, there are problems for example in topic deletion where concurrency isn't limited (problem explained in https://github.com/apache/pulsar/pull/22541#issuecomment-2071568113).
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Fix pattern consumer create crash if a part of partitions of a topic have been deleted [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui merged PR #22854:
URL: https://github.com/apache/pulsar/pull/22854
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix] [client] Fix pattern consumer create crash if a part of partitions of a topic have been deleted [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on PR #22854:
URL: https://github.com/apache/pulsar/pull/22854#issuecomment-2173950941
Rebased master
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix] [client] Fix pattern consumer create crash if a part of partitions of a topic have been deleted [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #22854:
URL: https://github.com/apache/pulsar/pull/22854#discussion_r1639672447
##########
pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java:
##########
@@ -102,6 +102,10 @@ public static boolean isValid(String topic) {
}
}
+ public static String getPartitionPattern(String topic) {
+ return "^" + get(topic).getPartitionedTopicName().toString() + "-partition-[0-9]+$";
Review Comment:
Good suggestion, fixed
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix] [client] Fix pattern consumer create crash if a part of partitions of a topic have been deleted [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #22854:
URL: https://github.com/apache/pulsar/pull/22854#discussion_r1639675959
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java:
##########
@@ -1067,30 +1074,43 @@ private void doSubscribeTopicPartitions(Schema<T> schema,
ConsumerConfigurationData<T> configurationData = getInternalConsumerConfig();
configurationData.setReceiverQueueSize(receiverQueueSize);
- futureList = IntStream
- .range(0, numPartitions)
- .mapToObj(
- partitionIndex -> {
- String partitionName = TopicName.get(topicName).getPartition(partitionIndex).toString();
- CompletableFuture<Consumer<T>> subFuture = new CompletableFuture<>();
- configurationData.setStartPaused(paused);
- ConsumerImpl<T> newConsumer = createInternalConsumer(configurationData, partitionName,
- partitionIndex, subFuture, createIfDoesNotExist, schema);
- synchronized (pauseMutex) {
- if (paused) {
- newConsumer.pause();
- } else {
- newConsumer.resume();
- }
- consumers.putIfAbsent(newConsumer.getTopic(), newConsumer);
+ CompletableFuture<List<Integer>> partitionsFuture;
+ if (createIfDoesNotExist || !TopicName.get(topicName).isPersistent()) {
+ partitionsFuture = CompletableFuture.completedFuture(IntStream.range(0, numPartitions)
+ .mapToObj(i -> Integer.valueOf(i))
+ .collect(Collectors.toList()));
+ } else {
+ partitionsFuture = getExistsPartitions(topicName.toString());
+ }
+ subscribeAllPartitionsFuture = partitionsFuture.thenCompose(partitions -> {
+ if (partitions.isEmpty()) {
+ partitionedTopics.remove(topicName, numPartitions);
+ return CompletableFuture.completedFuture(null);
+ }
+ List<CompletableFuture<Consumer<T>>> subscribeList = new ArrayList<>();
+ for (int partitionIndex : partitions) {
+ String partitionName = TopicName.get(topicName).getPartition(partitionIndex).toString();
+ CompletableFuture<Consumer<T>> subFuture = new CompletableFuture<>();
+ configurationData.setStartPaused(paused);
+ ConsumerImpl<T> newConsumer = createInternalConsumer(configurationData, partitionName,
+ partitionIndex, subFuture, createIfDoesNotExist, schema);
+ synchronized (pauseMutex) {
+ if (paused) {
+ newConsumer.pause();
+ } else {
+ newConsumer.resume();
}
- return subFuture;
- })
- .collect(Collectors.toList());
+ consumers.putIfAbsent(newConsumer.getTopic(), newConsumer);
Review Comment:
Fixed
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix] [client] Fix pattern consumer create crash if a part of partitions of a topic have been deleted [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #22854:
URL: https://github.com/apache/pulsar/pull/22854#discussion_r1628364294
##########
pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java:
##########
@@ -102,6 +102,10 @@ public static boolean isValid(String topic) {
}
}
+ public static String getPartitionPattern(String topic) {
+ return "^" + get(topic).getPartitionedTopicName().toString() + "-partition-[0-9]+$";
Review Comment:
Use Pattern.quote to escape possible pattern characters in the 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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org