[RIP-37] Add new APIs for consumer #4019
Conversation
apis/src/main/java/org/apache/rocketmq/apis/consumer/PullConsumer.java
Outdated
Show resolved
Hide resolved
| * <p>Once push consumer is closed, <strong>it could not be started once again.</strong> we maintained an FSM | ||
| * (finite-state machine) to record the different states for each producer, which is similar to | ||
| * {@link Service.State}. | ||
| */ |
There was a problem hiding this comment.
Maybe we need to get the MessageListenner.
There was a problem hiding this comment.
Can you explain the exact scene that need get MessageListener.
There was a problem hiding this comment.
When we use subscribe mode, client will rebalance and poll messages from messageQueue into local buffer. MessageListener will be invoked after this relationship change, and start task to pull message.
On the other hand, app wants to be notify once the queue it consumes is determined, init app with messageQueue, messageListener will be used in this scene.
apis/src/main/java/org/apache/rocketmq/apis/consumer/SimpleConsumer.java
Show resolved
Hide resolved
apis/src/main/java/org/apache/rocketmq/apis/consumer/PullConsumer.java
Outdated
Show resolved
Hide resolved
apis/src/main/java/org/apache/rocketmq/apis/consumer/PullConsumer.java
Outdated
Show resolved
Hide resolved
apis/src/main/java/org/apache/rocketmq/apis/consumer/MessageListener.java
Outdated
Show resolved
Hide resolved
apis/src/main/java/org/apache/rocketmq/apis/consumer/PullConsumer.java
Outdated
Show resolved
Hide resolved
apis/src/main/java/org/apache/rocketmq/apis/consumer/PullConsumer.java
Outdated
Show resolved
Hide resolved
apis/src/main/java/org/apache/rocketmq/apis/consumer/PullConsumer.java
Outdated
Show resolved
Hide resolved
apis/src/main/java/org/apache/rocketmq/apis/ClientServiceProvider.java
Outdated
Show resolved
Hide resolved
apis/src/main/java/org/apache/rocketmq/apis/consumer/PullConsumer.java
Outdated
Show resolved
Hide resolved
apis/src/main/java/org/apache/rocketmq/apis/consumer/PullConsumerBuilder.java
Outdated
Show resolved
Hide resolved
apis/src/main/java/org/apache/rocketmq/apis/consumer/PushConsumer.java
Outdated
Show resolved
Hide resolved
apis/src/main/java/org/apache/rocketmq/apis/consumer/PullConsumerBuilder.java
Outdated
Show resolved
Hide resolved
apis/src/main/java/org/apache/rocketmq/apis/consumer/PullConsumerBuilder.java
Outdated
Show resolved
Hide resolved
apis/src/main/java/org/apache/rocketmq/apis/consumer/PullConsumer.java
Outdated
Show resolved
Hide resolved
| * SimpleConsumer is a thread-safe rocketmq client which is used to consume message by group. | ||
| * | ||
| * <p>Simple consumer is lightweight consumer , if you want fully control the message consumption operation by yourself, | ||
| * simple consumer should be your first consideration. |
There was a problem hiding this comment.
Consider adding the difference between SimpleConsumer and PullConsumer.
apis/src/main/java/org/apache/rocketmq/apis/consumer/SimpleConsumer.java
Outdated
Show resolved
Hide resolved
| * @return map of messageQueue to messageViews. | ||
| */ | ||
| Map<MessageQueue, Collection<MessageView>> pull(int maxMessageNum) throws ClientException; | ||
|
|
There was a problem hiding this comment.
sync pull can use pullAsync().get
There was a problem hiding this comment.
But it's not convenient to catch the real exception(ClientException) when using pullAsync.get.
| * @param messageQueue the specified messageQueue to commit offset | ||
| * @param committedOffset the specified offset commit to server | ||
| */ | ||
| void commit(MessageQueue messageQueue, long committedOffset) throws ClientException; |
There was a problem hiding this comment.
need commit a collection of messageQueue interface;
and commitedOffset can not find in the return of pull method.
| * @param subscriptionExpression new subscription expression to add. | ||
| * @return pull consumer instance. | ||
| */ | ||
| PullConsumer subscribe(SubscriptionExpression subscriptionExpression) throws ClientException; |
There was a problem hiding this comment.
subscribe(HashSet subs) is need.
apis/src/main/java/org/apache/rocketmq/apis/consumer/FilterExpression.java
Show resolved
Hide resolved
| * @param messageViews are batch of messages which need consume. this message in the collection may come from different topics and the collection size control by setBatchSize(int batchSize) in {@link PushConsumerBuilder} | ||
| * @param ackList are collection of messages which already consume success and need ack to server. | ||
| */ | ||
| void consume(Collection<MessageView> messageViews, Collection<MessageView> ackList); |
There was a problem hiding this comment.
Do we consider replacing ackList with failedList? The code will look cleaner if all messages succeed.
There was a problem hiding this comment.
Do we consider replacing
ackListwithfailedList? The code will look cleaner if all messages succeed.
How do we define this behavior that exception is thrown during the invocation of MessageListener? Agree with your opinion to some degree, I feel anxious about user forget to put message into the ackList explicitly, which may cause a bulk of retry-messages.
|
Let's merge this PR to the 5.0 branch first and polish the APIs continuously. |
This reverts commit df5e885.
This reverts commit df5e885.
As we mentioned in RIP-37 New and unified APIs, establish a new APIs specifications. we divide it into two independent pull requests.
The producer part.
The consumer part.
Similar to the previous part #3987, this pull request is about the consumer part.
related issue: #3973