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/10/25 04:06:32 UTC
[GitHub] [pulsar] congbobo184 opened a new issue, #18182: PIP-218: Consumer batchReceive() single partition every receive
congbobo184 opened a new issue, #18182:
URL: https://github.com/apache/pulsar/issues/18182
### Motivation
Currently, client API does not make the use of `acknowledgeCumulative` simple and easy to understand.
If the user uses the receive() interface to consume multi-partition topic, there are two ways to use `acknowledgeCumulative` correctly :
1. Cache the different topic name messageId locally.
```java
Map<String, MessageId> map = new HashMap<>();
while (true) {
for (int i = 0; i < 1000; i++) {
Message<String> message = consumer.receive();
map.put(message.getTopicName(), message.getMessageId());
}
map.forEach((k, v) -> consumer.acknowledgeCumulativeAsync(v));
map.clear();
}
```
It is complex and challenging for users to understand.
2. Every message invokes `acknowledgeCumulative`.
```java
while (true) {
Message<String> message = consumer.receive();
consumer.acknowledgeCumulativeAsync(message);
}
```
Simple, but does not conform to the definition of 'acknowledgeCumulative', which is difficult for users to understand.
Similarly, using batchReceive() is the same as receive(). Therefore, we need to implement a simple and easy-to-understand usage way.
### Goal
Because receive() cannot reflect the value of 'cumulative ack', so we mainly consider batchReceive(). We can make every batchReceive() receive the same topic messages.
```java
```
// consumer the multi-partition messages
while (true) {
Messages<String> messages = consumer.batchReceive();
Message<String> finalMessage = null;
for (Message<String> message : messages) {
process(message);
finalMessage = message;
}
if (finalMessage != null) {
consumer.acknowledgeCumulativeAsync(finalMessage);
}
}
```
```
### API Changes
### API Changes
Add configuration in `BatchReceivePolicy`.
```java
/**
* whether receive the single topic message every time(default: false).
*/
private final boolean receiveSingleTopicMessageEnabled;
```
### Implementation
Change ConsumerBase batchReceive logical. If peek different topic message, return the batch result.
```java
protected final void notifyPendingBatchReceivedCallBack(CompletableFuture<Messages<T>> batchReceiveFuture) {
MessagesImpl<T> messages = getNewMessagesImpl();
Message<T> msgPeeked = incomingMessages.peek();
String topicName = null;
while (msgPeeked != null && messages.canAdd(msgPeeked)) {
// one batch receive request only can receive the same topic partition
if (receiveSingleTopicMessageEnabled) {
// get the first message's `topicName` to check if the following message peeked is the same topic message.
if (messages.size() == 1) {
topicName = messages.getMessageList().get(0).getTopicName();
}
// if the peeked message is not the same topic as the first message, return the batch result
if (topicName != null && !topicName.equals(msgPeeked.getTopicName())) {
break;
}
}
Message<T> msg = incomingMessages.poll();
if (msg != null) {
messageProcessed(msg);
Message<T> interceptMsg = beforeConsume(msg);
messages.add(interceptMsg);
}
msgPeeked = incomingMessages.peek();
}
completePendingBatchReceive(batchReceiveFuture, messages);
}
```
### Alternatives
Don't add any config, this will cause users to fall back to the old version of the client and cannot use this feature without awareness.
### Anything else?
#### Follow-up Work
optimize performance
#### mail discuss
#### Are you willing to submit a PR?
- [x] I'm willing to submit a PR!
--
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.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] github-actions[bot] commented on issue #18182: PIP-218: Consumer batchReceive() single partition every receive
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on issue #18182:
URL: https://github.com/apache/pulsar/issues/18182#issuecomment-1334669375
The issue 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] codelipenghui commented on issue #18182: PIP-218: Consumer batchReceive() single partition every receive
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on issue #18182:
URL: https://github.com/apache/pulsar/issues/18182#issuecomment-1290652170
Just one suggestion about the new option name
Maybe `messagesFromMultiTopicsEnabled=true` is more understandable.
And add a description of the behavior if the user disables it.
It looks like if `messagesFromMultiTopicsEnabled=true`. The max messages and max size will not be strictly followed.
--
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] congbobo184 closed issue #18182: PIP-218: Consumer batchReceive() single partition every receive
Posted by "congbobo184 (via GitHub)" <gi...@apache.org>.
congbobo184 closed issue #18182: PIP-218: Consumer batchReceive() single partition every receive
URL: https://github.com/apache/pulsar/issues/18182
--
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] congbobo184 commented on issue #18182: PIP-218: Consumer batchReceive() single partition every receive
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on issue #18182:
URL: https://github.com/apache/pulsar/issues/18182#issuecomment-1293303936
> Just one suggestion about the new option name
>
> Maybe `messagesFromMultiTopicsEnabled=true` is more understandable. And add a description of the behavior if the user disables it. It looks like if `messagesFromMultiTopicsEnabled=false`. The max messages and max size will not be strictly followed.
make sense
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on issue #18182: PIP-218: Consumer batchReceive() single partition every receive
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on issue #18182:
URL: https://github.com/apache/pulsar/issues/18182#issuecomment-1291587204
> Just one suggestion about the new option name
>
> Maybe `messagesFromMultiTopicsEnabled=true` is more understandable. And add a description of the behavior if the user disables it. It looks like if `messagesFromMultiTopicsEnabled=true`. The max messages and max size will not be strictly followed.
make sense
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org