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 2021/05/11 23:46:40 UTC
[GitHub] [pulsar] complone opened a new pull request #10544: [plusar-10479] Resolved Allow Java client to create consumer with rec…
complone opened a new pull request #10544:
URL: https://github.com/apache/pulsar/pull/10544
<!--
### Contribution Checklist
- Name the pull request in the form "[Issue XYZ][component] Title of the pull request", where *XYZ* should be replaced by the actual issue number.
Skip *Issue XYZ* if there is no associated github issue for this pull request.
Skip *component* if you are unsure about which is the best component. E.g. `[docs] Fix typo in produce method`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
*(If this PR fixes a github issue, please add `Fixes #<xyz>`.)*
Fixes #<xyz>
*(or if this PR is one task of a github issue, please add `Master Issue: #<xyz>` to link to the master issue.)*
Master Issue: #<xyz>
### Motivation
*Explain here the context, and why you're making that change. What is the problem you're trying to solve.*
### Modifications
Modify ConsumerBase#receive to allow Java client to create consumer with receiverQueueSize(0) when receiving with timeout
### Verifying this change
Demo sample is here
[plusar-10479](https://github.com/apache/pulsar/issues/10479)
*(Please pick either of the following options)*
This change is a trivial rework / code cleanup without any test coverage.
*(or)*
This change is already covered by existing tests, such as *(please describe tests)*.
*(or)*
This change added tests and can be verified as follows:
*(example:)*
- *Added integration tests for end-to-end deployment with large payloads (10MB)*
- *Extended integration test for recovery after broker failure*
### Does this pull request potentially affect one of the following parts:
*If `yes` was chosen, please highlight the changes*
- Dependencies (does it add or upgrade a dependency): (yes / no)
- The public API: (yes / no)
- The schema: (yes / no / don't know)
- The default values of configurations: (yes / no)
- The wire protocol: (yes / no)
- The rest endpoints: (yes / no)
- The admin cli options: (yes / no)
- Anything that affects deployment: (yes / no / don't know)
### Documentation
- Does this pull request introduce a new feature? (yes / no)
- If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
- If a feature is not applicable for documentation, explain why?
- If a feature is not documented yet in this PR, please create a followup issue for adding the documentation
This abstract method A has a zero queue implementation
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] complone commented on a change in pull request #10544: [plusar-10479] Resolved Allow Java client to create consumer with rec…
Posted by GitBox <gi...@apache.org>.
complone commented on a change in pull request #10544:
URL: https://github.com/apache/pulsar/pull/10544#discussion_r637244804
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
##########
@@ -172,8 +172,17 @@ protected ConsumerBase(PulsarClientImpl client, String topic, ConsumerConfigurat
@Override
public Message<T> receive(int timeout, TimeUnit unit) throws PulsarClientException {
if (conf.getReceiverQueueSize() == 0) {
- throw new PulsarClientException.InvalidConfigurationException(
- "Can't use receive with timeout, if the queue size is 0");
+
+ if(!(getState() == State.Connecting)){
Review comment:
Good two, this function has been refactored, can you take time to have a look？
@linlinnn @zymap
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on pull request #10544: [plusar-10479] Resolved Allow Java client to create consumer with rec…
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on pull request #10544:
URL: https://github.com/apache/pulsar/pull/10544#issuecomment-869128136
@complone Could you please add some description for this PR? From the conversation, looks like support zero queue consumer for the multiple topic consumer?
--
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] complone commented on a change in pull request #10544: [plusar-10479] Resolved Allow Java client to create consumer with rec…
Posted by GitBox <gi...@apache.org>.
complone commented on a change in pull request #10544:
URL: https://github.com/apache/pulsar/pull/10544#discussion_r646169083
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
##########
@@ -184,6 +191,50 @@ protected ConsumerBase(PulsarClientImpl client, String topic, ConsumerConfigurat
return internalReceive(timeout, unit);
}
+ @Override
+ public CompletableFuture<Message<T>> receiveAsync(int timeout, TimeUnit unit) {
+
+ if (listener != null) {
+ return FutureUtil.failedFuture(new PulsarClientException.InvalidConfigurationException(
+ "Cannot use receive() when a listener has been set"));
+ }
+ try {
+ verifyConsumerState();
+ } catch (PulsarClientException e) {
+ return FutureUtil.failedFuture(e);
+ }
+
+ ScheduledExecutorService messageScheduledFuture = Executors.newSingleThreadScheduledExecutor(
Review comment:
> can you please clarify the description of this work ?
> it looks interesting but I do not fully understand the goal
@eolivelli In order to achieve the collection of a message within a given time as you mentioned, allowing users to use the Java client to use receiverQueueSize(0) to create consumers when receiving overtime
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] linlinnn commented on a change in pull request #10544: [plusar-10479] Resolved Allow Java client to create consumer with rec…
Posted by GitBox <gi...@apache.org>.
linlinnn commented on a change in pull request #10544:
URL: https://github.com/apache/pulsar/pull/10544#discussion_r637725474
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
##########
@@ -184,6 +191,50 @@ protected ConsumerBase(PulsarClientImpl client, String topic, ConsumerConfigurat
return internalReceive(timeout, unit);
}
+ @Override
+ public CompletableFuture<Message<T>> receiveAsync(int timeout, TimeUnit unit) {
+
+ if (listener != null) {
+ return FutureUtil.failedFuture(new PulsarClientException.InvalidConfigurationException(
+ "Cannot use receive() when a listener has been set"));
+ }
+ try {
+ verifyConsumerState();
+ } catch (PulsarClientException e) {
+ return FutureUtil.failedFuture(e);
+ }
+
+ ScheduledExecutorService messageScheduledFuture = Executors.newSingleThreadScheduledExecutor(
Review comment:
@complone I left some comments, please take a look.
and then you can ping me again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] linlinnn commented on a change in pull request #10544: [plusar-10479] Resolved Allow Java client to create consumer with rec…
Posted by GitBox <gi...@apache.org>.
linlinnn commented on a change in pull request #10544:
URL: https://github.com/apache/pulsar/pull/10544#discussion_r631520251
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
##########
@@ -172,8 +172,17 @@ protected ConsumerBase(PulsarClientImpl client, String topic, ConsumerConfigurat
@Override
public Message<T> receive(int timeout, TimeUnit unit) throws PulsarClientException {
if (conf.getReceiverQueueSize() == 0) {
- throw new PulsarClientException.InvalidConfigurationException(
- "Can't use receive with timeout, if the queue size is 0");
+
+ if(!(getState() == State.Connecting)){
Review comment:
We already check the state of consumer by `verifyConsumerState`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] complone commented on a change in pull request #10544: [plusar-10479] Resolved Allow Java client to create consumer with rec…
Posted by GitBox <gi...@apache.org>.
complone commented on a change in pull request #10544:
URL: https://github.com/apache/pulsar/pull/10544#discussion_r640314823
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
##########
@@ -184,6 +191,50 @@ protected ConsumerBase(PulsarClientImpl client, String topic, ConsumerConfigurat
return internalReceive(timeout, unit);
}
+ @Override
+ public CompletableFuture<Message<T>> receiveAsync(int timeout, TimeUnit unit) {
+
+ if (listener != null) {
+ return FutureUtil.failedFuture(new PulsarClientException.InvalidConfigurationException(
+ "Cannot use receive() when a listener has been set"));
+ }
+ try {
+ verifyConsumerState();
+ } catch (PulsarClientException e) {
+ return FutureUtil.failedFuture(e);
+ }
+
+ ScheduledExecutorService messageScheduledFuture = Executors.newSingleThreadScheduledExecutor(
Review comment:
Okay, thank you for your attention to this issue. There are some things that need to be dealt with recently, it will be done as soon as possible this week
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] complone edited a comment on pull request #10544: [plusar-10479] Resolved Allow Java client to create consumer with rec…
Posted by GitBox <gi...@apache.org>.
complone edited a comment on pull request #10544:
URL: https://github.com/apache/pulsar/pull/10544#issuecomment-839556400
> Thanks for the contribution. Add a test for this?
using ZeroQueueSizeTest#zeroQueueSizeReceieveAsyncInCompatibility
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] complone commented on a change in pull request #10544: [plusar-10479] Resolved Allow Java client to create consumer with rec…
Posted by GitBox <gi...@apache.org>.
complone commented on a change in pull request #10544:
URL: https://github.com/apache/pulsar/pull/10544#discussion_r631444735
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
##########
@@ -172,8 +172,17 @@ protected ConsumerBase(PulsarClientImpl client, String topic, ConsumerConfigurat
@Override
public Message<T> receive(int timeout, TimeUnit unit) throws PulsarClientException {
if (conf.getReceiverQueueSize() == 0) {
- throw new PulsarClientException.InvalidConfigurationException(
- "Can't use receive with timeout, if the queue size is 0");
+
+ if(!(getState() == State.Connecting)){
Review comment:
I think this way. The meaning of this state is whether the connection between each consumer and the topic is alive. If the task is cancelled or interrupted during the process of receiving the message, the downstream consumer should be notified in time
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] complone edited a comment on pull request #10544: [plusar-10479] Resolved Allow Java client to create consumer with rec…
Posted by GitBox <gi...@apache.org>.
complone edited a comment on pull request #10544:
URL: https://github.com/apache/pulsar/pull/10544#issuecomment-869140532
> @complone你能为这个 PR 添加一些描述吗？从对话中，看起来像多主题消费者支持零队列消费者？
@codelipenghui This PR is aimed at the problem raised by [issues-10479](https://github.com/apache/pulsar/issues/10479), that is, when the parameter ```receiverQueueSize``` configured by ```PulsarClient``` is set to 0, the ```ZeroQueueConsumerImpl#receive``` method will be called to schedule the ```ZeroQueueConsumerImpl#fetchSingleMessageFromBroker``` task, but it does not consider the task timeout.
--
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] linlinnn commented on pull request #10544: [plusar-10479] Resolved Allow Java client to create consumer with rec…
Posted by GitBox <gi...@apache.org>.
linlinnn commented on pull request #10544:
URL: https://github.com/apache/pulsar/pull/10544#issuecomment-855532002
@eolivelli
Currently, the user case receiving messages with the config `receiverQueueSize = 0` will throw exception directly.
This PR is intended to support it by scheduling the task that `fetchSingleMessageFromBroker` with timeout in `ZeroQueueConsumerImpl`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] complone commented on a change in pull request #10544: [plusar-10479] Resolved Allow Java client to create consumer with rec…
Posted by GitBox <gi...@apache.org>.
complone commented on a change in pull request #10544:
URL: https://github.com/apache/pulsar/pull/10544#discussion_r631444735
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
##########
@@ -172,8 +172,17 @@ protected ConsumerBase(PulsarClientImpl client, String topic, ConsumerConfigurat
@Override
public Message<T> receive(int timeout, TimeUnit unit) throws PulsarClientException {
if (conf.getReceiverQueueSize() == 0) {
- throw new PulsarClientException.InvalidConfigurationException(
- "Can't use receive with timeout, if the queue size is 0");
+
+ if(!(getState() == State.Connecting)){
Review comment:
I think this way. The meaning of this state is whether the connection between each consumer and the topic is alive. If the connection is closed or interrupted during the process of receiving the message, the downstream consumer should be notified in time
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] linlinnn commented on a change in pull request #10544: [plusar-10479] Resolved Allow Java client to create consumer with rec…
Posted by GitBox <gi...@apache.org>.
linlinnn commented on a change in pull request #10544:
URL: https://github.com/apache/pulsar/pull/10544#discussion_r643693410
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
##########
@@ -184,6 +191,50 @@ protected ConsumerBase(PulsarClientImpl client, String topic, ConsumerConfigurat
return internalReceive(timeout, unit);
}
+ @Override
+ public CompletableFuture<Message<T>> receiveAsync(int timeout, TimeUnit unit) {
+
+ if (listener != null) {
+ return FutureUtil.failedFuture(new PulsarClientException.InvalidConfigurationException(
+ "Cannot use receive() when a listener has been set"));
+ }
+ try {
+ verifyConsumerState();
+ } catch (PulsarClientException e) {
+ return FutureUtil.failedFuture(e);
+ }
+
+ ScheduledExecutorService messageScheduledFuture = Executors.newSingleThreadScheduledExecutor(
Review comment:
Yes, we should let it run as user expected, look forward to it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] complone commented on pull request #10544: [plusar-10479] Resolved Allow Java client to create consumer with rec…
Posted by GitBox <gi...@apache.org>.
complone commented on pull request #10544:
URL: https://github.com/apache/pulsar/pull/10544#issuecomment-846273832
Good two, this function has been refactored, can you take time to have a look？
@linlinnn @zymap
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] complone edited a comment on pull request #10544: [plusar-10479] Resolved Allow Java client to create consumer with rec…
Posted by GitBox <gi...@apache.org>.
complone edited a comment on pull request #10544:
URL: https://github.com/apache/pulsar/pull/10544#issuecomment-855440649
> @complone I think you are working on a right way, good, but we still need to do more work for completing it.
@linlinnn Okay. At the moment, I am also working on enhancements to the company’s data distribution system. I will complete this demand improvement as soon as possible
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] linlinnn commented on a change in pull request #10544: [plusar-10479] Resolved Allow Java client to create consumer with rec…
Posted by GitBox <gi...@apache.org>.
linlinnn commented on a change in pull request #10544:
URL: https://github.com/apache/pulsar/pull/10544#discussion_r630865925
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
##########
@@ -172,8 +172,17 @@ protected ConsumerBase(PulsarClientImpl client, String topic, ConsumerConfigurat
@Override
public Message<T> receive(int timeout, TimeUnit unit) throws PulsarClientException {
if (conf.getReceiverQueueSize() == 0) {
- throw new PulsarClientException.InvalidConfigurationException(
- "Can't use receive with timeout, if the queue size is 0");
+
+ if(!(getState() == State.Connecting)){
Review comment:
Thanks for your contribution.
But I think this is not a right way to fix this.
for example, you catch the case not connecting, and throw exception indicate "Can't use receive with timeout, if the queue size is 0"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on pull request #10544: [plusar-10479] Resolved Allow Java client to create consumer with rec…
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on pull request #10544:
URL: https://github.com/apache/pulsar/pull/10544#issuecomment-1058890511
The pr had no activity for 30 days, mark with Stale label.
--
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] linlinnn commented on a change in pull request #10544: [plusar-10479] Resolved Allow Java client to create consumer with rec…
Posted by GitBox <gi...@apache.org>.
linlinnn commented on a change in pull request #10544:
URL: https://github.com/apache/pulsar/pull/10544#discussion_r645986900
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/ZeroQueueConsumerImpl.java
##########
@@ -56,6 +62,45 @@ public ZeroQueueConsumerImpl(PulsarClientImpl client, String topic, ConsumerConf
super(client, topic, conf, executorProvider, partitionIndex, hasParentConsumer, subscribeFuture,
startMessageId, 0 /* startMessageRollbackDurationInSec */, schema, interceptors,
createTopicIfDoesNotExist);
+ //Actually,i want to achieve to
+ this.messageScheduledFuture = Executors.newSingleThreadScheduledExecutor(
+ new DefaultThreadFactory(getClass().getSimpleName()+"-receive-message"));
+ }
+
+
+ @Override
+ protected Message<T> internalReceive(int timeout, TimeUnit unit) throws PulsarClientException {
+ Message<T> tMessage = null;
+
+ if (Objects.nonNull(conf.getReceiveThreads())){
+ messageScheduledFuture = Executors.newScheduledThreadPool(
Review comment:
in this way, every call will create a new instance if we have this conf.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] complone commented on a change in pull request #10544: [plusar-10479] Resolved Allow Java client to create consumer with rec…
Posted by GitBox <gi...@apache.org>.
complone commented on a change in pull request #10544:
URL: https://github.com/apache/pulsar/pull/10544#discussion_r642739810
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
##########
@@ -184,6 +191,50 @@ protected ConsumerBase(PulsarClientImpl client, String topic, ConsumerConfigurat
return internalReceive(timeout, unit);
}
+ @Override
+ public CompletableFuture<Message<T>> receiveAsync(int timeout, TimeUnit unit) {
+
+ if (listener != null) {
+ return FutureUtil.failedFuture(new PulsarClientException.InvalidConfigurationException(
+ "Cannot use receive() when a listener has been set"));
+ }
+ try {
+ verifyConsumerState();
+ } catch (PulsarClientException e) {
+ return FutureUtil.failedFuture(e);
+ }
+
+ ScheduledExecutorService messageScheduledFuture = Executors.newSingleThreadScheduledExecutor(
Review comment:
@linlinnn Hi
I noticed that user @eolivelli mentioned that we should provide users with options for their own configuration. I think we should refer to the implementation of pulsar's other timing thread pools and let users set how long they need to wait when the elements of the receiving queue are 0.
originally posted by @eolivelli in https://github.com/apache/pulsar/issues/10479#issuecomment-831996071_
Relevant patches will be submitted in the next few days, thank you for your guidance
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] linlinnn commented on a change in pull request #10544: [plusar-10479] Resolved Allow Java client to create consumer with rec…
Posted by GitBox <gi...@apache.org>.
linlinnn commented on a change in pull request #10544:
URL: https://github.com/apache/pulsar/pull/10544#discussion_r637719012
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
##########
@@ -32,6 +32,9 @@
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
+import java.time.Duration;
+import java.util.*;
Review comment:
please avoid import *
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
##########
@@ -184,6 +191,50 @@ protected ConsumerBase(PulsarClientImpl client, String topic, ConsumerConfigurat
return internalReceive(timeout, unit);
}
+ @Override
+ public CompletableFuture<Message<T>> receiveAsync(int timeout, TimeUnit unit) {
+
+ if (listener != null) {
+ return FutureUtil.failedFuture(new PulsarClientException.InvalidConfigurationException(
+ "Cannot use receive() when a listener has been set"));
+ }
+ try {
+ verifyConsumerState();
+ } catch (PulsarClientException e) {
+ return FutureUtil.failedFuture(e);
+ }
+
+ ScheduledExecutorService messageScheduledFuture = Executors.newSingleThreadScheduledExecutor(
Review comment:
1. In this way, you break the former logic handling the case when queue size is not zero.
2. We just need remove the code below:
```
if (conf.getReceiverQueueSize() == 0) {
throw new PulsarClientException.InvalidConfigurationException(
"Can't use receive with timeout, if the queue size is 0");
}
```
and then override the method `internalReceive(timeout, unit)` in `ZeroQueueConsumerImpl`
3. use single instance of `ScheduledExecutorService`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] linlinnn commented on a change in pull request #10544: [plusar-10479] Resolved Allow Java client to create consumer with rec…
Posted by GitBox <gi...@apache.org>.
linlinnn commented on a change in pull request #10544:
URL: https://github.com/apache/pulsar/pull/10544#discussion_r630865925
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
##########
@@ -172,8 +172,17 @@ protected ConsumerBase(PulsarClientImpl client, String topic, ConsumerConfigurat
@Override
public Message<T> receive(int timeout, TimeUnit unit) throws PulsarClientException {
if (conf.getReceiverQueueSize() == 0) {
- throw new PulsarClientException.InvalidConfigurationException(
- "Can't use receive with timeout, if the queue size is 0");
+
+ if(!(getState() == State.Connecting)){
Review comment:
Thanks for your contribution.
But I think this is not a right way to fix 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] complone commented on pull request #10544: [plusar-10479] Resolved Allow Java client to create consumer with rec…
Posted by GitBox <gi...@apache.org>.
complone commented on pull request #10544:
URL: https://github.com/apache/pulsar/pull/10544#issuecomment-839556400
> Thanks for the contribution. Add a test for this?
using ZeroQueueSizeTest#zeroQueueSizeReceieveAsyncInCompatibility
> Thanks for the contribution. Add a test for this?
ZeroQueueSizeTest#zeroQueueSizeReceieveAsyncInCompatibility
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] complone commented on pull request #10544: [plusar-10479] Resolved Allow Java client to create consumer with rec…
Posted by GitBox <gi...@apache.org>.
complone commented on pull request #10544:
URL: https://github.com/apache/pulsar/pull/10544#issuecomment-855440649
> @complone I think you are working on a right way, good, but we still need to do more work for completing it.
Okay. At the moment, I am also working on enhancements to the company’s data distribution system. I will complete this demand improvement as soon as possible
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] complone commented on a change in pull request #10544: [plusar-10479] Resolved Allow Java client to create consumer with rec…
Posted by GitBox <gi...@apache.org>.
complone commented on a change in pull request #10544:
URL: https://github.com/apache/pulsar/pull/10544#discussion_r646169809
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/ZeroQueueConsumerImpl.java
##########
@@ -56,6 +62,45 @@ public ZeroQueueConsumerImpl(PulsarClientImpl client, String topic, ConsumerConf
super(client, topic, conf, executorProvider, partitionIndex, hasParentConsumer, subscribeFuture,
startMessageId, 0 /* startMessageRollbackDurationInSec */, schema, interceptors,
createTopicIfDoesNotExist);
+ //Actually,i want to achieve to
+ this.messageScheduledFuture = Executors.newSingleThreadScheduledExecutor(
+ new DefaultThreadFactory(getClass().getSimpleName()+"-receive-message"));
+ }
+
+
+ @Override
+ protected Message<T> internalReceive(int timeout, TimeUnit unit) throws PulsarClientException {
+ Message<T> tMessage = null;
+
+ if (Objects.nonNull(conf.getReceiveThreads())){
+ messageScheduledFuture = Executors.newScheduledThreadPool(
+ conf.getReceiveThreads(),new DefaultThreadFactory(getClass().getSimpleName()+"-receive-message")
+ );
+ }
+
+ long time1 = System.nanoTime();
+
+ Supplier<Message<T>> asyncTask = this::doZeroQueue;
+ CompletableFuture<Message<T>> future = new CompletableFuture<>();
+ try {
+ future = FutureUtil.schedule(messageScheduledFuture, asyncTask,
+ conf.getReceiveInterval(), TimeUnit.SECONDS);
+ //Determine whether the task is over, and then get the result
+ if (future.isDone()) {
+
+ tMessage = future.get();
+ }
+ }catch (ExecutionException | InterruptedException ignore){
+ long time2 = System.nanoTime();
+ System.out.println("No timeout after " +
+ (time2-time1)/1000000000.0 + " seconds");
+ }finally {
+ //When the captured futrue throws an exception, the worker thread clears the interruption status in time
+ future.cancel(true);
Review comment:
@linlinnn Because we did not start a thread to complete a task execution, I think we should close this futrue in time, and the task call has been completed
enmm and the premise implemented here is to allocate a thread pool instance in advance
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] complone commented on a change in pull request #10544: [plusar-10479] Resolved Allow Java client to create consumer with rec…
Posted by GitBox <gi...@apache.org>.
complone commented on a change in pull request #10544:
URL: https://github.com/apache/pulsar/pull/10544#discussion_r646170170
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/ZeroQueueConsumerImpl.java
##########
@@ -81,6 +126,17 @@ public ZeroQueueConsumerImpl(PulsarClientImpl client, String topic, ConsumerConf
return future;
}
+ private Message<T> doZeroQueue() {
+ Message<T> result = null;
+ try {
+ result = internalReceive();
Review comment:
I think a semaphore is needed here to complete the blocking
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] linlinnn commented on a change in pull request #10544: [plusar-10479] Resolved Allow Java client to create consumer with rec…
Posted by GitBox <gi...@apache.org>.
linlinnn commented on a change in pull request #10544:
URL: https://github.com/apache/pulsar/pull/10544#discussion_r645986900
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/ZeroQueueConsumerImpl.java
##########
@@ -56,6 +62,45 @@ public ZeroQueueConsumerImpl(PulsarClientImpl client, String topic, ConsumerConf
super(client, topic, conf, executorProvider, partitionIndex, hasParentConsumer, subscribeFuture,
startMessageId, 0 /* startMessageRollbackDurationInSec */, schema, interceptors,
createTopicIfDoesNotExist);
+ //Actually,i want to achieve to
+ this.messageScheduledFuture = Executors.newSingleThreadScheduledExecutor(
+ new DefaultThreadFactory(getClass().getSimpleName()+"-receive-message"));
+ }
+
+
+ @Override
+ protected Message<T> internalReceive(int timeout, TimeUnit unit) throws PulsarClientException {
+ Message<T> tMessage = null;
+
+ if (Objects.nonNull(conf.getReceiveThreads())){
+ messageScheduledFuture = Executors.newScheduledThreadPool(
Review comment:
in this way, every call will create a new instance.
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/ZeroQueueConsumerImpl.java
##########
@@ -81,6 +126,17 @@ public ZeroQueueConsumerImpl(PulsarClientImpl client, String topic, ConsumerConf
return future;
}
+ private Message<T> doZeroQueue() {
+ Message<T> result = null;
+ try {
+ result = internalReceive();
Review comment:
internalReceive will hold a lock, we cannot call it directly when receiving message with timeout.
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/ZeroQueueConsumerImpl.java
##########
@@ -56,6 +62,45 @@ public ZeroQueueConsumerImpl(PulsarClientImpl client, String topic, ConsumerConf
super(client, topic, conf, executorProvider, partitionIndex, hasParentConsumer, subscribeFuture,
startMessageId, 0 /* startMessageRollbackDurationInSec */, schema, interceptors,
createTopicIfDoesNotExist);
+ //Actually,i want to achieve to
+ this.messageScheduledFuture = Executors.newSingleThreadScheduledExecutor(
+ new DefaultThreadFactory(getClass().getSimpleName()+"-receive-message"));
+ }
+
+
+ @Override
+ protected Message<T> internalReceive(int timeout, TimeUnit unit) throws PulsarClientException {
+ Message<T> tMessage = null;
+
+ if (Objects.nonNull(conf.getReceiveThreads())){
+ messageScheduledFuture = Executors.newScheduledThreadPool(
+ conf.getReceiveThreads(),new DefaultThreadFactory(getClass().getSimpleName()+"-receive-message")
+ );
+ }
+
+ long time1 = System.nanoTime();
+
+ Supplier<Message<T>> asyncTask = this::doZeroQueue;
+ CompletableFuture<Message<T>> future = new CompletableFuture<>();
+ try {
+ future = FutureUtil.schedule(messageScheduledFuture, asyncTask,
+ conf.getReceiveInterval(), TimeUnit.SECONDS);
+ //Determine whether the task is over, and then get the result
+ if (future.isDone()) {
+
+ tMessage = future.get();
+ }
+ }catch (ExecutionException | InterruptedException ignore){
+ long time2 = System.nanoTime();
Review comment:
we need to interrupt current thread when catch InterruptedException
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/ZeroQueueConsumerImpl.java
##########
@@ -56,6 +62,45 @@ public ZeroQueueConsumerImpl(PulsarClientImpl client, String topic, ConsumerConf
super(client, topic, conf, executorProvider, partitionIndex, hasParentConsumer, subscribeFuture,
startMessageId, 0 /* startMessageRollbackDurationInSec */, schema, interceptors,
createTopicIfDoesNotExist);
+ //Actually,i want to achieve to
+ this.messageScheduledFuture = Executors.newSingleThreadScheduledExecutor(
Review comment:
we need add some shutdown logic when we close the consumer.
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/ZeroQueueConsumerImpl.java
##########
@@ -56,6 +62,45 @@ public ZeroQueueConsumerImpl(PulsarClientImpl client, String topic, ConsumerConf
super(client, topic, conf, executorProvider, partitionIndex, hasParentConsumer, subscribeFuture,
startMessageId, 0 /* startMessageRollbackDurationInSec */, schema, interceptors,
createTopicIfDoesNotExist);
+ //Actually,i want to achieve to
+ this.messageScheduledFuture = Executors.newSingleThreadScheduledExecutor(
+ new DefaultThreadFactory(getClass().getSimpleName()+"-receive-message"));
+ }
+
+
+ @Override
+ protected Message<T> internalReceive(int timeout, TimeUnit unit) throws PulsarClientException {
+ Message<T> tMessage = null;
+
+ if (Objects.nonNull(conf.getReceiveThreads())){
+ messageScheduledFuture = Executors.newScheduledThreadPool(
+ conf.getReceiveThreads(),new DefaultThreadFactory(getClass().getSimpleName()+"-receive-message")
+ );
+ }
+
+ long time1 = System.nanoTime();
Review comment:
please rename it like startTime or beginTime.
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/ZeroQueueConsumerImpl.java
##########
@@ -56,6 +62,45 @@ public ZeroQueueConsumerImpl(PulsarClientImpl client, String topic, ConsumerConf
super(client, topic, conf, executorProvider, partitionIndex, hasParentConsumer, subscribeFuture,
startMessageId, 0 /* startMessageRollbackDurationInSec */, schema, interceptors,
createTopicIfDoesNotExist);
+ //Actually,i want to achieve to
+ this.messageScheduledFuture = Executors.newSingleThreadScheduledExecutor(
+ new DefaultThreadFactory(getClass().getSimpleName()+"-receive-message"));
+ }
+
+
+ @Override
+ protected Message<T> internalReceive(int timeout, TimeUnit unit) throws PulsarClientException {
+ Message<T> tMessage = null;
+
+ if (Objects.nonNull(conf.getReceiveThreads())){
+ messageScheduledFuture = Executors.newScheduledThreadPool(
+ conf.getReceiveThreads(),new DefaultThreadFactory(getClass().getSimpleName()+"-receive-message")
+ );
+ }
+
+ long time1 = System.nanoTime();
+
+ Supplier<Message<T>> asyncTask = this::doZeroQueue;
+ CompletableFuture<Message<T>> future = new CompletableFuture<>();
+ try {
+ future = FutureUtil.schedule(messageScheduledFuture, asyncTask,
+ conf.getReceiveInterval(), TimeUnit.SECONDS);
+ //Determine whether the task is over, and then get the result
+ if (future.isDone()) {
+
+ tMessage = future.get();
+ }
+ }catch (ExecutionException | InterruptedException ignore){
+ long time2 = System.nanoTime();
+ System.out.println("No timeout after " +
+ (time2-time1)/1000000000.0 + " seconds");
+ }finally {
+ //When the captured futrue throws an exception, the worker thread clears the interruption status in time
+ future.cancel(true);
Review comment:
finally means that we will always close the future.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] complone commented on pull request #10544: [plusar-10479] Resolved Allow Java client to create consumer with rec…
Posted by GitBox <gi...@apache.org>.
complone commented on pull request #10544:
URL: https://github.com/apache/pulsar/pull/10544#issuecomment-869140532
> @complone你能为这个 PR 添加一些描述吗？从对话中，看起来像多主题消费者支持零队列消费者？
This PR is aimed at the problem raised by [issues-10479](https://github.com/apache/pulsar/issues/10479), that is, when the parameter ```receiverQueueSize``` configured by ```PulsarClient``` is set to 0, the ```ZeroQueueConsumerImpl#receive``` method will be called to schedule the ```ZeroQueueConsumerImpl#fetchSingleMessageFromBroker``` task, but it does not consider the task timeout.
--
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] linlinnn commented on pull request #10544: [plusar-10479] Resolved Allow Java client to create consumer with rec…
Posted by GitBox <gi...@apache.org>.
linlinnn commented on pull request #10544:
URL: https://github.com/apache/pulsar/pull/10544#issuecomment-839549813
Emm.. plusar -> pulsar
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] complone edited a comment on pull request #10544: [plusar-10479] Resolved Allow Java client to create consumer with rec…
Posted by GitBox <gi...@apache.org>.
complone edited a comment on pull request #10544:
URL: https://github.com/apache/pulsar/pull/10544#issuecomment-839556400
> Thanks for the contribution. Add a test for this?
ZeroQueueSizeTest#zeroQueueSizeReceieveAsyncInCompatibility
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] linlinnn commented on pull request #10544: [plusar-10479] Resolved Allow Java client to create consumer with rec…
Posted by GitBox <gi...@apache.org>.
linlinnn commented on pull request #10544:
URL: https://github.com/apache/pulsar/pull/10544#issuecomment-855241414
@complone I think you are working on a right way, good, but we still need to do more work for completing it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] complone commented on pull request #10544: [plusar-10479] Resolved Allow Java client to create consumer with rec…
Posted by GitBox <gi...@apache.org>.
complone commented on pull request #10544:
URL: https://github.com/apache/pulsar/pull/10544#issuecomment-839557614
> Emm.. plusar -> pulsar
The version reported by users is 2.7.1, so I merged to branch-2.7.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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] complone commented on pull request #10544: [plusar-10479] Resolved Allow Java client to create consumer with rec…
Posted by GitBox <gi...@apache.org>.
complone commented on pull request #10544:
URL: https://github.com/apache/pulsar/pull/10544#issuecomment-839396549
@sijie Can you help review？
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org