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 2022/11/25 11:14:20 UTC
[GitHub] [pulsar] lordcheng10 opened a new pull request, #18633: [fix][client] For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block
lordcheng10 opened a new pull request, #18633:
URL: https://github.com/apache/pulsar/pull/18633
### Motivation
For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block.
**Steps to reproduce:**
Start the following program twice. When starting the second time, the subscribe method will block and will not throw an exception: ConsumerBusyException: Exclusive consumer is already connected
```
String topic = "test1";
List<Integer> partitions = new ArrayList<>();
for (int i = 0; i < 120; i++) {
partitions.add(i);
}
List<String> realTopics = partitions.stream()
.map(idx -> topic + "-partition-" + idx)
.collect(Collectors.toList());
Map<String, Object> consumerParams = Collections.emptyMap();
System.out.println("begin subscribe " + LocalTime.now());
Consumer<byte[]> consumer;
try {
consumer = client.newConsumer()
.loadConf(consumerParams)
.topics(realTopics)
.subscriptionName(testName)
.subscriptionType(SubscriptionType.Exclusive)
.subscriptionInitialPosition(SubscriptionInitialPosition.Latest)
.subscribe();
} catch (Exception ex) {
System.out.println("after subscribe " + LocalTime.now() );
}
```
The reasons are as follows:
When the 1121 line of code is executed, the consumers list may be empty, which will cause the subscribeFuture to fail to complete, and eventually cause the subscribe method to block forever and not throw an exception:
https://github.com/apache/pulsar/blob/355ba6002fc33b2716f59ca5d3a4fac667ecf3e1/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java#L1121-L1144
### Modifications
Make subscribeFuture complete when the filtered consumer collection is empty：
```
List<ConsumerImpl> filterConsumers = consumers.values().stream().filter(consumer1 -> {
String consumerTopicName = consumer1.getTopic();
if (TopicName.get(consumerTopicName).getPartitionedTopicName()
.equals(TopicName.get(topicName).getPartitionedTopicName())) {
toCloseNum.incrementAndGet();
return true;
} else {
return false;
}
}).collect(Collectors.toList());
if (filterConsumers.isEmpty()) {
subscribeFuture.completeExceptionally(error);
return;
}
```
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. Please attach the local preview screenshots (run `sh start.sh` at `pulsar/site2/website`) to your PR description, or else your PR might not get merged. -->
- [ ] `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: <!-- ENTER URL HERE -->
<!--
After opening this PR, the build in apache/pulsar will fail and instructions will
be provided for opening a PR in the PR author's forked repository.
apache/pulsar pull requests should be first tested in your own fork since the
apache/pulsar CI based on GitHub Actions has constrained resources and quota.
GitHub Actions provides separate quota for pull requests that are executed in
a forked repository.
The tests will be run in the forked repository until all PR review comments have
been handled, the tests pass and the PR is approved by a reviewer.
-->
--
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
[GitHub] [pulsar] lordcheng10 commented on a diff in pull request #18633: [fix][client] For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block
Posted by GitBox <gi...@apache.org>.
lordcheng10 commented on code in PR #18633:
URL: https://github.com/apache/pulsar/pull/18633#discussion_r1035452413
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java:
##########
@@ -539,6 +540,38 @@ public void testTopicNameValid() throws Exception{
}).get();
}
+ @Test
+ public void testExclusiveSubscribe() throws Exception {
+ final String topicName = "persistent://prop/use/ns-abc/testTopicNameValid";
+ TenantInfoImpl tenantInfo = createDefaultTenantInfo();
+ admin.tenants().createTenant("prop", tenantInfo);
+ admin.topics().createPartitionedTopic(topicName, 3);
+ List<Integer> partitions = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ partitions.add(i);
+ }
+ List<String> realTopics = partitions.stream()
+ .map(idx -> topicName + "-partition-" + idx)
+ .collect(Collectors.toList());
+ Consumer<byte[]> consumer1 = pulsarClient.newConsumer()
+ .topics(realTopics)
+ .subscriptionName("subscriptionName")
+ .subscriptionType(SubscriptionType.Exclusive)
+ .subscribeAsync().get(10, TimeUnit.SECONDS);
+
+
+ try {
+ Consumer<byte[]> consumer2 = pulsarClient.newConsumer()
+ .topics(realTopics)
+ .subscriptionName("subscriptionName")
+ .subscriptionType(SubscriptionType.Exclusive)
+ .subscribeAsync().get(10, TimeUnit.SECONDS);
+ } catch (Exception e) {
Review Comment:
After using the subscribe() interface instead of subscribeAsync().get(10, TimeUnit.SECONDS), there will be no TimeoutException.
There will only be PulsarClientException, so I use PulsarClientException instead
--
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
[GitHub] [pulsar] lordcheng10 commented on pull request #18633: [fix][client] For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block
Posted by GitBox <gi...@apache.org>.
lordcheng10 commented on PR #18633:
URL: https://github.com/apache/pulsar/pull/18633#issuecomment-1328944708
> @lordcheng10 it is very nice of you to explain this to me. i agree with you. it seems in MultiTopicConsumer the CompletableFuture<Consumer> subscription is merged among all partition.
>
> in my opinion if when one partition trigger exception logic, only remove the relevant consumer is more elegant.
>
> definity the current modification works, look forward to other people's idea.
If a topic has a partition subscribe failure, then all partitions should fail, so all partitions corresponding to the topic should be removed from consumers.
--
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
[GitHub] [pulsar] lordcheng10 commented on a diff in pull request #18633: [fix][client] For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block
Posted by GitBox <gi...@apache.org>.
lordcheng10 commented on code in PR #18633:
URL: https://github.com/apache/pulsar/pull/18633#discussion_r1035445167
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java:
##########
@@ -539,6 +540,38 @@ public void testTopicNameValid() throws Exception{
}).get();
}
+ @Test
+ public void testExclusiveSubscribe() throws Exception {
+ final String topicName = "persistent://prop/use/ns-abc/testTopicNameValid";
+ TenantInfoImpl tenantInfo = createDefaultTenantInfo();
+ admin.tenants().createTenant("prop", tenantInfo);
+ admin.topics().createPartitionedTopic(topicName, 3);
+ List<Integer> partitions = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ partitions.add(i);
+ }
+ List<String> realTopics = partitions.stream()
+ .map(idx -> topicName + "-partition-" + idx)
+ .collect(Collectors.toList());
+ Consumer<byte[]> consumer1 = pulsarClient.newConsumer()
+ .topics(realTopics)
+ .subscriptionName("subscriptionName")
+ .subscriptionType(SubscriptionType.Exclusive)
+ .subscribeAsync().get(10, TimeUnit.SECONDS);
Review Comment:
I will fix
--
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
[GitHub] [pulsar] lordcheng10 commented on pull request #18633: [fix][client] For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block
Posted by GitBox <gi...@apache.org>.
lordcheng10 commented on PR #18633:
URL: https://github.com/apache/pulsar/pull/18633#issuecomment-1331556351
> > @lordcheng10 it is very nice of you to explain this to me. i agree with you. it seems in MultiTopicConsumer the CompletableFuture subscription is merged among all partition.
> > in my opinion if when one partition trigger exception logic, only remove the relevant consumer is more elegant.
> > definity the current modification works, look forward to other people's idea.
>
> If a topic has a partition subscribe failure, then all partitions should fail, so all partitions of this topic should be removed from consumers. @WJL3333
IMO, it's fine to remove all partitions or just one partition
--
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
[GitHub] [pulsar] lordcheng10 commented on pull request #18633: [fix][client] For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block
Posted by GitBox <gi...@apache.org>.
lordcheng10 commented on PR #18633:
URL: https://github.com/apache/pulsar/pull/18633#issuecomment-1327397732
> Could you add a test to avoid regression?
OK
--
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
[GitHub] [pulsar] WJL3333 commented on pull request #18633: [fix][client] For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block
Posted by GitBox <gi...@apache.org>.
WJL3333 commented on PR #18633:
URL: https://github.com/apache/pulsar/pull/18633#issuecomment-1327685939
> When the 1121 line of code is executed, the consumers list may be empty
i wonder why this will happen? is there some race-condition here ?
--
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
[GitHub] [pulsar] lordcheng10 commented on pull request #18633: [fix][client] For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block
Posted by GitBox <gi...@apache.org>.
lordcheng10 commented on PR #18633:
URL: https://github.com/apache/pulsar/pull/18633#issuecomment-1334674405
ping @BewareMyPower @codelipenghui
--
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
[GitHub] [pulsar] BewareMyPower commented on pull request #18633: [fix][client] For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block
Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on PR #18633:
URL: https://github.com/apache/pulsar/pull/18633#issuecomment-1336741744
@lordcheng10 Could you check my comment [here](https://github.com/apache/pulsar/pull/18633/files#r1039101336)?
--
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
[GitHub] [pulsar] BewareMyPower commented on a diff in pull request #18633: [fix][client] For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block
Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on code in PR #18633:
URL: https://github.com/apache/pulsar/pull/18633#discussion_r1034243195
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java:
##########
@@ -539,6 +540,38 @@ public void testTopicNameValid() throws Exception{
}).get();
}
+ @Test
+ public void testExclusiveSubscribe() throws Exception {
+ final String topicName = "persistent://prop/use/ns-abc/testTopicNameValid";
+ TenantInfoImpl tenantInfo = createDefaultTenantInfo();
+ admin.tenants().createTenant("prop", tenantInfo);
+ admin.topics().createPartitionedTopic(topicName, 3);
+ List<Integer> partitions = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ partitions.add(i);
+ }
+ List<String> realTopics = partitions.stream()
+ .map(idx -> topicName + "-partition-" + idx)
+ .collect(Collectors.toList());
+ Consumer<byte[]> consumer1 = pulsarClient.newConsumer()
+ .topics(realTopics)
Review Comment:
+1, we can subscribe the partitioned topic directly instead of all these partitions.
```java
Consumer<byte[]> consumer1 = pulsarClient.newConsumer()
.topic(topicName)
```
--
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
[GitHub] [pulsar] BewareMyPower commented on a diff in pull request #18633: [fix][client] For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block
Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on code in PR #18633:
URL: https://github.com/apache/pulsar/pull/18633#discussion_r1034231788
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java:
##########
@@ -539,6 +540,38 @@ public void testTopicNameValid() throws Exception{
}).get();
}
+ @Test
+ public void testExclusiveSubscribe() throws Exception {
+ final String topicName = "persistent://prop/use/ns-abc/testTopicNameValid";
Review Comment:
I think it's consistent with the other tests in `TopicsConsumerImplTest`.
--
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
[GitHub] [pulsar] WJL3333 commented on pull request #18633: [fix][client] For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block
Posted by GitBox <gi...@apache.org>.
WJL3333 commented on PR #18633:
URL: https://github.com/apache/pulsar/pull/18633#issuecomment-1327987273
@lordcheng10 it is very nice of you to explain this to me. i agree with you.
it seems in MultiTopicConsumer the CompletableFuture<Consumer<T>> subscription is merged among all partition.
in my opinion if when one partition trigger exception logic, only remove the relevant consumer is more elegant.
definity the current modification works, look forward to other people's idea.
--
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
[GitHub] [pulsar] lordcheng10 commented on a diff in pull request #18633: [fix][client] For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block
Posted by GitBox <gi...@apache.org>.
lordcheng10 commented on code in PR #18633:
URL: https://github.com/apache/pulsar/pull/18633#discussion_r1035444352
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java:
##########
@@ -539,6 +540,38 @@ public void testTopicNameValid() throws Exception{
}).get();
}
+ @Test
+ public void testExclusiveSubscribe() throws Exception {
+ final String topicName = "persistent://prop/use/ns-abc/testTopicNameValid";
+ TenantInfoImpl tenantInfo = createDefaultTenantInfo();
+ admin.tenants().createTenant("prop", tenantInfo);
+ admin.topics().createPartitionedTopic(topicName, 3);
+ List<Integer> partitions = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ partitions.add(i);
+ }
+ List<String> realTopics = partitions.stream()
+ .map(idx -> topicName + "-partition-" + idx)
+ .collect(Collectors.toList());
+ Consumer<byte[]> consumer1 = pulsarClient.newConsumer()
+ .topics(realTopics)
Review Comment:
@codelipenghui @BewareMyPower
Direct subscription to partition topic will not trigger this bug
--
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
[GitHub] [pulsar] lordcheng10 commented on pull request #18633: [fix][client] For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block
Posted by GitBox <gi...@apache.org>.
lordcheng10 commented on PR #18633:
URL: https://github.com/apache/pulsar/pull/18633#issuecomment-1336707912
ping @BewareMyPower @codelipenghui
--
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
[GitHub] [pulsar] lordcheng10 commented on pull request #18633: [fix][client] For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block
Posted by GitBox <gi...@apache.org>.
lordcheng10 commented on PR #18633:
URL: https://github.com/apache/pulsar/pull/18633#issuecomment-1337004867
> @lordcheng10 Could you check my comment [here](https://github.com/apache/pulsar/pull/18633/files#r1039101336)?
@BewareMyPower Fixed,PTAL,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
[GitHub] [pulsar] lordcheng10 commented on pull request #18633: [fix][client] For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block
Posted by GitBox <gi...@apache.org>.
lordcheng10 commented on PR #18633:
URL: https://github.com/apache/pulsar/pull/18633#issuecomment-1337281048
/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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] lordcheng10 commented on a diff in pull request #18633: [fix][client] For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block
Posted by GitBox <gi...@apache.org>.
lordcheng10 commented on code in PR #18633:
URL: https://github.com/apache/pulsar/pull/18633#discussion_r1035444911
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java:
##########
@@ -539,6 +540,38 @@ public void testTopicNameValid() throws Exception{
}).get();
}
+ @Test
+ public void testExclusiveSubscribe() throws Exception {
+ final String topicName = "persistent://prop/use/ns-abc/testTopicNameValid";
+ TenantInfoImpl tenantInfo = createDefaultTenantInfo();
+ admin.tenants().createTenant("prop", tenantInfo);
+ admin.topics().createPartitionedTopic(topicName, 3);
+ List<Integer> partitions = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ partitions.add(i);
+ }
+ List<String> realTopics = partitions.stream()
+ .map(idx -> topicName + "-partition-" + idx)
+ .collect(Collectors.toList());
+ Consumer<byte[]> consumer1 = pulsarClient.newConsumer()
+ .topics(realTopics)
+ .subscriptionName("subscriptionName")
+ .subscriptionType(SubscriptionType.Exclusive)
+ .subscribeAsync().get(10, TimeUnit.SECONDS);
+
+
+ try {
+ Consumer<byte[]> consumer2 = pulsarClient.newConsumer()
+ .topics(realTopics)
+ .subscriptionName("subscriptionName")
+ .subscriptionType(SubscriptionType.Exclusive)
+ .subscribeAsync().get(10, TimeUnit.SECONDS);
+ } catch (Exception e) {
+ String errorLog = e.getMessage();
+ assertTrue(errorLog != null && errorLog.contains("Exclusive consumer is already connected"));
+ }
Review Comment:
The bug will not be triggered after consumer1 is closed. @lifepuzzlefun
--
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
[GitHub] [pulsar] lordcheng10 commented on pull request #18633: [fix][client] For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block
Posted by GitBox <gi...@apache.org>.
lordcheng10 commented on PR #18633:
URL: https://github.com/apache/pulsar/pull/18633#issuecomment-1327353696
@codelipenghui @Jason918 @AnonHxy PTAL,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
[GitHub] [pulsar] lordcheng10 commented on pull request #18633: [fix][client] For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block
Posted by GitBox <gi...@apache.org>.
lordcheng10 commented on PR #18633:
URL: https://github.com/apache/pulsar/pull/18633#issuecomment-1327922876
> > When the 1121 line of code is executed, the consumers list may be empty
>
>
>
> i wonder why this will happen? is there some race-condition here ?
Yes
--
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
[GitHub] [pulsar] lordcheng10 commented on pull request #18633: [fix][client] For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block
Posted by GitBox <gi...@apache.org>.
lordcheng10 commented on PR #18633:
URL: https://github.com/apache/pulsar/pull/18633#issuecomment-1327983887
> > When the 1121 line of code is executed, the consumers list may be empty
>
> i wonder why this will happen? is there some race-condition here ?
For example:
The test1 topic has 10 partitions, and the handleSubscribeOneTopicError method is executed for each partition.
Assuming that test1-partition-0 executes the handleSubscribeOneTopicError method first, all the consumers collection will be removed, so that when the subsequent partition executes handleSubscribeOneTopicError, the corresponding subscribeResult will not be completed, resulting in blocking:
https://github.com/apache/pulsar/blob/006559dbfa5613e4b4b622416375e5bcbd1d580d/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java#L1130-L1139
--
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
[GitHub] [pulsar] lordcheng10 commented on pull request #18633: [fix][client] For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block
Posted by GitBox <gi...@apache.org>.
lordcheng10 commented on PR #18633:
URL: https://github.com/apache/pulsar/pull/18633#issuecomment-1328930677
> in my opinion if when one partition trigger exception logic, only remove the relevant consumer is more elegant.
@WJL3333
I disagree.
All partitions of a topic, if one subscribe fails, then all partitions should fail, so all partitions of the topic should be removed from consumers here.
--
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
[GitHub] [pulsar] WJL3333 commented on pull request #18633: [fix][client] For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block
Posted by GitBox <gi...@apache.org>.
WJL3333 commented on PR #18633:
URL: https://github.com/apache/pulsar/pull/18633#issuecomment-1329100187
> > @lordcheng10 it is very nice of you to explain this to me. i agree with you. it seems in MultiTopicConsumer the CompletableFuture subscription is merged among all partition.
> > in my opinion if when one partition trigger exception logic, only remove the relevant consumer is more elegant.
> > definity the current modification works, look forward to other people's idea.
>
> If a topic has a partition subscribe failure, then all partitions should fail, so all partitions of this topic should be removed from consumers. @WJL3333
i mean only the caller who put object in the map should remove when get response from server. and after all response is received, we can process the whole subscription is success or not.
--
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
[GitHub] [pulsar] BewareMyPower commented on a diff in pull request #18633: [fix][client] For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block
Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on code in PR #18633:
URL: https://github.com/apache/pulsar/pull/18633#discussion_r1039101336
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java:
##########
@@ -539,6 +540,38 @@ public void testTopicNameValid() throws Exception{
}).get();
}
+ @Test
+ public void testExclusiveSubscribe() throws Exception {
+ final String topicName = "persistent://prop/use/ns-abc/testTopicNameValid";
+ TenantInfoImpl tenantInfo = createDefaultTenantInfo();
+ admin.tenants().createTenant("prop", tenantInfo);
+ admin.topics().createPartitionedTopic(topicName, 3);
+ List<Integer> partitions = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ partitions.add(i);
+ }
+ List<String> realTopics = partitions.stream()
+ .map(idx -> topicName + "-partition-" + idx)
+ .collect(Collectors.toList());
+ Consumer<byte[]> consumer1 = pulsarClient.newConsumer()
+ .topics(realTopics)
Review Comment:
@lordcheng10 `consumer1` could subscribe the partitioned topic. Here is the test result without your patch.
BTW, you can simplify the `consumer2` creation like:
```java
.topics(IntStream.range(0, 3).mapToObj(i -> topicName + "-partition-" + i)
.collect(Collectors.toList()))
```
--
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
[GitHub] [pulsar] lordcheng10 commented on a diff in pull request #18633: [fix][client] For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block
Posted by GitBox <gi...@apache.org>.
lordcheng10 commented on code in PR #18633:
URL: https://github.com/apache/pulsar/pull/18633#discussion_r1035448678
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java:
##########
@@ -539,6 +540,38 @@ public void testTopicNameValid() throws Exception{
}).get();
}
+ @Test
+ public void testExclusiveSubscribe() throws Exception {
+ final String topicName = "persistent://prop/use/ns-abc/testTopicNameValid";
+ TenantInfoImpl tenantInfo = createDefaultTenantInfo();
+ admin.tenants().createTenant("prop", tenantInfo);
+ admin.topics().createPartitionedTopic(topicName, 3);
+ List<Integer> partitions = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ partitions.add(i);
+ }
+ List<String> realTopics = partitions.stream()
+ .map(idx -> topicName + "-partition-" + idx)
+ .collect(Collectors.toList());
+ Consumer<byte[]> consumer1 = pulsarClient.newConsumer()
+ .topics(realTopics)
+ .subscriptionName("subscriptionName")
+ .subscriptionType(SubscriptionType.Exclusive)
+ .subscribeAsync().get(10, TimeUnit.SECONDS);
+
+
+ try {
+ Consumer<byte[]> consumer2 = pulsarClient.newConsumer()
+ .topics(realTopics)
+ .subscriptionName("subscriptionName")
+ .subscriptionType(SubscriptionType.Exclusive)
+ .subscribeAsync().get(10, TimeUnit.SECONDS);
+ } catch (Exception e) {
+ String errorLog = e.getMessage();
+ assertTrue(errorLog != null && errorLog.contains("Exclusive consumer is already connected"));
+ }
Review Comment:
After all execution, I will close consumer1
--
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
[GitHub] [pulsar] WJL3333 commented on a diff in pull request #18633: [fix][client] For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block
Posted by GitBox <gi...@apache.org>.
WJL3333 commented on code in PR #18633:
URL: https://github.com/apache/pulsar/pull/18633#discussion_r1033570970
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java:
##########
@@ -539,6 +540,38 @@ public void testTopicNameValid() throws Exception{
}).get();
}
+ @Test
+ public void testExclusiveSubscribe() throws Exception {
+ final String topicName = "persistent://prop/use/ns-abc/testTopicNameValid";
+ TenantInfoImpl tenantInfo = createDefaultTenantInfo();
+ admin.tenants().createTenant("prop", tenantInfo);
+ admin.topics().createPartitionedTopic(topicName, 3);
+ List<Integer> partitions = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ partitions.add(i);
+ }
+ List<String> realTopics = partitions.stream()
+ .map(idx -> topicName + "-partition-" + idx)
+ .collect(Collectors.toList());
+ Consumer<byte[]> consumer1 = pulsarClient.newConsumer()
+ .topics(realTopics)
+ .subscriptionName("subscriptionName")
+ .subscriptionType(SubscriptionType.Exclusive)
+ .subscribeAsync().get(10, TimeUnit.SECONDS);
+
+
+ try {
+ Consumer<byte[]> consumer2 = pulsarClient.newConsumer()
+ .topics(realTopics)
+ .subscriptionName("subscriptionName")
+ .subscriptionType(SubscriptionType.Exclusive)
+ .subscribeAsync().get(10, TimeUnit.SECONDS);
+ } catch (Exception e) {
+ String errorLog = e.getMessage();
+ assertTrue(errorLog != null && errorLog.contains("Exclusive consumer is already connected"));
+ }
Review Comment:
consumer1 should be closed.
--
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
[GitHub] [pulsar] lordcheng10 commented on pull request #18633: [fix][client] For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block
Posted by GitBox <gi...@apache.org>.
lordcheng10 commented on PR #18633:
URL: https://github.com/apache/pulsar/pull/18633#issuecomment-1328974363
> Could you add a test to avoid regression?
@BewareMyPower Fixed,PTAL, 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
[GitHub] [pulsar] lordcheng10 commented on a diff in pull request #18633: [fix][client] For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block
Posted by GitBox <gi...@apache.org>.
lordcheng10 commented on code in PR #18633:
URL: https://github.com/apache/pulsar/pull/18633#discussion_r1035448678
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java:
##########
@@ -539,6 +540,38 @@ public void testTopicNameValid() throws Exception{
}).get();
}
+ @Test
+ public void testExclusiveSubscribe() throws Exception {
+ final String topicName = "persistent://prop/use/ns-abc/testTopicNameValid";
+ TenantInfoImpl tenantInfo = createDefaultTenantInfo();
+ admin.tenants().createTenant("prop", tenantInfo);
+ admin.topics().createPartitionedTopic(topicName, 3);
+ List<Integer> partitions = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ partitions.add(i);
+ }
+ List<String> realTopics = partitions.stream()
+ .map(idx -> topicName + "-partition-" + idx)
+ .collect(Collectors.toList());
+ Consumer<byte[]> consumer1 = pulsarClient.newConsumer()
+ .topics(realTopics)
+ .subscriptionName("subscriptionName")
+ .subscriptionType(SubscriptionType.Exclusive)
+ .subscribeAsync().get(10, TimeUnit.SECONDS);
+
+
+ try {
+ Consumer<byte[]> consumer2 = pulsarClient.newConsumer()
+ .topics(realTopics)
+ .subscriptionName("subscriptionName")
+ .subscriptionType(SubscriptionType.Exclusive)
+ .subscribeAsync().get(10, TimeUnit.SECONDS);
+ } catch (Exception e) {
+ String errorLog = e.getMessage();
+ assertTrue(errorLog != null && errorLog.contains("Exclusive consumer is already connected"));
+ }
Review Comment:
After all execution, I will close consumer1 and consumer2
--
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
[GitHub] [pulsar] BewareMyPower commented on a diff in pull request #18633: [fix][client] For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block
Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on code in PR #18633:
URL: https://github.com/apache/pulsar/pull/18633#discussion_r1034238241
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java:
##########
@@ -539,6 +540,38 @@ public void testTopicNameValid() throws Exception{
}).get();
}
+ @Test
+ public void testExclusiveSubscribe() throws Exception {
+ final String topicName = "persistent://prop/use/ns-abc/testTopicNameValid";
+ TenantInfoImpl tenantInfo = createDefaultTenantInfo();
+ admin.tenants().createTenant("prop", tenantInfo);
+ admin.topics().createPartitionedTopic(topicName, 3);
+ List<Integer> partitions = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ partitions.add(i);
+ }
+ List<String> realTopics = partitions.stream()
+ .map(idx -> topicName + "-partition-" + idx)
+ .collect(Collectors.toList());
+ Consumer<byte[]> consumer1 = pulsarClient.newConsumer()
+ .topics(realTopics)
+ .subscriptionName("subscriptionName")
+ .subscriptionType(SubscriptionType.Exclusive)
+ .subscribeAsync().get(10, TimeUnit.SECONDS);
+
+
+ try {
+ Consumer<byte[]> consumer2 = pulsarClient.newConsumer()
+ .topics(realTopics)
+ .subscriptionName("subscriptionName")
+ .subscriptionType(SubscriptionType.Exclusive)
+ .subscribeAsync().get(10, TimeUnit.SECONDS);
+ } catch (Exception e) {
Review Comment:
Catch `ExecutionException` to avoid `errorLog != null` check (when `errorLogs` is null, `e` is `TimeoutException`).
```java
fail();
} catch (ExecutionException e) {
assertTrue(e.getCause() instanceof PulsarClientException.ConsumerBusyException);
}
```
--
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
[GitHub] [pulsar] lordcheng10 commented on pull request #18633: [fix][client] For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block
Posted by GitBox <gi...@apache.org>.
lordcheng10 commented on PR #18633:
URL: https://github.com/apache/pulsar/pull/18633#issuecomment-1331544455
fixed, PTAL,thanks! @lifepuzzlefun @BewareMyPower @codelipenghui
--
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
[GitHub] [pulsar] codecov-commenter commented on pull request #18633: [fix][client] For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on PR #18633:
URL: https://github.com/apache/pulsar/pull/18633#issuecomment-1327415818
# [Codecov](https://codecov.io/gh/apache/pulsar/pull/18633?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#18633](https://codecov.io/gh/apache/pulsar/pull/18633?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (355ba60) into [master](https://codecov.io/gh/apache/pulsar/commit/fdf86d3b4e66e25936c2e2143bb5809b079ebc0d?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (fdf86d3) will **decrease** coverage by `4.04%`.
> The diff coverage is `36.76%`.
> :exclamation: Current head 355ba60 differs from pull request most recent head 9f653bb. Consider uploading reports for the commit 9f653bb to get more accurate results
[](https://codecov.io/gh/apache/pulsar/pull/18633?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## master #18633 +/- ##
============================================
- Coverage 45.67% 41.62% -4.05%
+ Complexity 10075 2406 -7669
============================================
Files 693 235 -458
Lines 67940 16666 -51274
Branches 7273 1818 -5455
============================================
- Hits 31030 6937 -24093
+ Misses 33333 8969 -24364
+ Partials 3577 760 -2817
```
| Flag | Coverage Δ | |
|---|---|---|
| unittests | `41.62% <36.76%> (-4.05%)` | :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=The+Apache+Software+Foundation#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://codecov.io/gh/apache/pulsar/pull/18633?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [.../java/org/apache/pulsar/client/impl/ClientCnx.java](https://codecov.io/gh/apache/pulsar/pull/18633/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL0NsaWVudENueC5qYXZh) | `29.97% <0.00%> (-0.20%)` | :arrow_down: |
| [...nt/impl/PulsarClientImplementationBindingImpl.java](https://codecov.io/gh/apache/pulsar/pull/18633/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL1B1bHNhckNsaWVudEltcGxlbWVudGF0aW9uQmluZGluZ0ltcGwuamF2YQ==) | `72.41% <ø> (+2.92%)` | :arrow_up: |
| [...pache/pulsar/client/impl/TableViewBuilderImpl.java](https://codecov.io/gh/apache/pulsar/pull/18633/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL1RhYmxlVmlld0J1aWxkZXJJbXBsLmphdmE=) | `0.00% <0.00%> (ø)` | |
| [...a/org/apache/pulsar/client/impl/TableViewImpl.java](https://codecov.io/gh/apache/pulsar/pull/18633/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL1RhYmxlVmlld0ltcGwuamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [...r/client/impl/UnAckedMessageRedeliveryTracker.java](https://codecov.io/gh/apache/pulsar/pull/18633/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL1VuQWNrZWRNZXNzYWdlUmVkZWxpdmVyeVRyYWNrZXIuamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [...ar/client/impl/conf/ConsumerConfigurationData.java](https://codecov.io/gh/apache/pulsar/pull/18633/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL2NvbmYvQ29uc3VtZXJDb25maWd1cmF0aW9uRGF0YS5qYXZh) | `92.55% <ø> (+2.12%)` | :arrow_up: |
| [...n/java/org/apache/pulsar/client/util/NoOpLock.java](https://codecov.io/gh/apache/pulsar/pull/18633/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC91dGlsL05vT3BMb2NrLmphdmE=) | `0.00% <0.00%> (ø)` | |
| [...va/org/apache/pulsar/client/impl/ConsumerImpl.java](https://codecov.io/gh/apache/pulsar/pull/18633/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL0NvbnN1bWVySW1wbC5qYXZh) | `15.12% <2.50%> (+0.02%)` | :arrow_up: |
| [...he/pulsar/client/impl/MultiTopicsConsumerImpl.java](https://codecov.io/gh/apache/pulsar/pull/18633/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL011bHRpVG9waWNzQ29uc3VtZXJJbXBsLmphdmE=) | `22.98% <7.14%> (+0.02%)` | :arrow_up: |
| [...va/org/apache/pulsar/client/impl/ConsumerBase.java](https://codecov.io/gh/apache/pulsar/pull/18633/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL0NvbnN1bWVyQmFzZS5qYXZh) | `21.93% <11.11%> (-0.02%)` | :arrow_down: |
| ... and [478 more](https://codecov.io/gh/apache/pulsar/pull/18633/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
--
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
[GitHub] [pulsar] codelipenghui commented on a diff in pull request #18633: [fix][client] For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on code in PR #18633:
URL: https://github.com/apache/pulsar/pull/18633#discussion_r1034201729
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java:
##########
@@ -539,6 +540,38 @@ public void testTopicNameValid() throws Exception{
}).get();
}
+ @Test
+ public void testExclusiveSubscribe() throws Exception {
+ final String topicName = "persistent://prop/use/ns-abc/testTopicNameValid";
+ TenantInfoImpl tenantInfo = createDefaultTenantInfo();
+ admin.tenants().createTenant("prop", tenantInfo);
+ admin.topics().createPartitionedTopic(topicName, 3);
+ List<Integer> partitions = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ partitions.add(i);
+ }
+ List<String> realTopics = partitions.stream()
+ .map(idx -> topicName + "-partition-" + idx)
+ .collect(Collectors.toList());
+ Consumer<byte[]> consumer1 = pulsarClient.newConsumer()
+ .topics(realTopics)
+ .subscriptionName("subscriptionName")
+ .subscriptionType(SubscriptionType.Exclusive)
+ .subscribeAsync().get(10, TimeUnit.SECONDS);
+
+
+ try {
+ Consumer<byte[]> consumer2 = pulsarClient.newConsumer()
+ .topics(realTopics)
+ .subscriptionName("subscriptionName")
+ .subscriptionType(SubscriptionType.Exclusive)
+ .subscribeAsync().get(10, TimeUnit.SECONDS);
+ } catch (Exception e) {
+ String errorLog = e.getMessage();
+ assertTrue(errorLog != null && errorLog.contains("Exclusive consumer is already connected"));
Review Comment:
If no exception is thrown, the test can still get passed.
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java:
##########
@@ -539,6 +540,38 @@ public void testTopicNameValid() throws Exception{
}).get();
}
+ @Test
+ public void testExclusiveSubscribe() throws Exception {
+ final String topicName = "persistent://prop/use/ns-abc/testTopicNameValid";
+ TenantInfoImpl tenantInfo = createDefaultTenantInfo();
+ admin.tenants().createTenant("prop", tenantInfo);
+ admin.topics().createPartitionedTopic(topicName, 3);
+ List<Integer> partitions = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ partitions.add(i);
+ }
+ List<String> realTopics = partitions.stream()
+ .map(idx -> topicName + "-partition-" + idx)
+ .collect(Collectors.toList());
+ Consumer<byte[]> consumer1 = pulsarClient.newConsumer()
+ .topics(realTopics)
+ .subscriptionName("subscriptionName")
+ .subscriptionType(SubscriptionType.Exclusive)
+ .subscribeAsync().get(10, TimeUnit.SECONDS);
Review Comment:
```suggestion
.subscribe();
```
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java:
##########
@@ -539,6 +540,38 @@ public void testTopicNameValid() throws Exception{
}).get();
}
+ @Test
+ public void testExclusiveSubscribe() throws Exception {
+ final String topicName = "persistent://prop/use/ns-abc/testTopicNameValid";
Review Comment:
It's better not to use the V1 version topic name. We will remove it in the future.
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java:
##########
@@ -539,6 +540,38 @@ public void testTopicNameValid() throws Exception{
}).get();
}
+ @Test
+ public void testExclusiveSubscribe() throws Exception {
+ final String topicName = "persistent://prop/use/ns-abc/testTopicNameValid";
+ TenantInfoImpl tenantInfo = createDefaultTenantInfo();
+ admin.tenants().createTenant("prop", tenantInfo);
+ admin.topics().createPartitionedTopic(topicName, 3);
+ List<Integer> partitions = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ partitions.add(i);
+ }
+ List<String> realTopics = partitions.stream()
+ .map(idx -> topicName + "-partition-" + idx)
+ .collect(Collectors.toList());
+ Consumer<byte[]> consumer1 = pulsarClient.newConsumer()
+ .topics(realTopics)
Review Comment:
Why not subscribe to the 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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] lordcheng10 commented on a diff in pull request #18633: [fix][client] For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block
Posted by GitBox <gi...@apache.org>.
lordcheng10 commented on code in PR #18633:
URL: https://github.com/apache/pulsar/pull/18633#discussion_r1035445068
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java:
##########
@@ -539,6 +540,38 @@ public void testTopicNameValid() throws Exception{
}).get();
}
+ @Test
+ public void testExclusiveSubscribe() throws Exception {
+ final String topicName = "persistent://prop/use/ns-abc/testTopicNameValid";
+ TenantInfoImpl tenantInfo = createDefaultTenantInfo();
+ admin.tenants().createTenant("prop", tenantInfo);
+ admin.topics().createPartitionedTopic(topicName, 3);
+ List<Integer> partitions = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ partitions.add(i);
+ }
+ List<String> realTopics = partitions.stream()
+ .map(idx -> topicName + "-partition-" + idx)
+ .collect(Collectors.toList());
+ Consumer<byte[]> consumer1 = pulsarClient.newConsumer()
+ .topics(realTopics)
+ .subscriptionName("subscriptionName")
+ .subscriptionType(SubscriptionType.Exclusive)
+ .subscribeAsync().get(10, TimeUnit.SECONDS);
+
+
+ try {
+ Consumer<byte[]> consumer2 = pulsarClient.newConsumer()
+ .topics(realTopics)
+ .subscriptionName("subscriptionName")
+ .subscriptionType(SubscriptionType.Exclusive)
+ .subscribeAsync().get(10, TimeUnit.SECONDS);
+ } catch (Exception e) {
+ String errorLog = e.getMessage();
+ assertTrue(errorLog != null && errorLog.contains("Exclusive consumer is already connected"));
Review Comment:
I will fix
--
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
[GitHub] [pulsar] lordcheng10 commented on a diff in pull request #18633: [fix][client] For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block
Posted by GitBox <gi...@apache.org>.
lordcheng10 commented on code in PR #18633:
URL: https://github.com/apache/pulsar/pull/18633#discussion_r1035450378
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java:
##########
@@ -539,6 +540,38 @@ public void testTopicNameValid() throws Exception{
}).get();
}
+ @Test
+ public void testExclusiveSubscribe() throws Exception {
+ final String topicName = "persistent://prop/use/ns-abc/testTopicNameValid";
Review Comment:
I will fix
--
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
[GitHub] [pulsar] lordcheng10 merged pull request #18633: [fix][client] For exclusive subscriptions, if two consumers are created repeatedly, the second consumer will block
Posted by GitBox <gi...@apache.org>.
lordcheng10 merged PR #18633:
URL: https://github.com/apache/pulsar/pull/18633
--
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