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/07/19 09:53:10 UTC
[GitHub] [pulsar] AnonHxy opened a new issue, #16680: PIP-191: Support batched message using entry filter
AnonHxy opened a new issue, #16680:
URL: https://github.com/apache/pulsar/issues/16680
## Motivation
Before We already haved a plug-in way to filter entries in broker, aka PIP-105 https://github.com/apache/pulsar/issues/12269. But this way only checks at the batch header, without digging into the individual messages properties. Of course it provides a interface to deserialize the entire Entry to the heap, But this will bring some memory and cpu workload. And in most scenario we only need partial of the properties to do some filter.
This proposal brings a method to make PIP-105 support batched entry without having to deserialize the entire Entry to the heap
## API Changes
- Add a producer config to specialize the key, of which properties will be added to the batched entry metadata, for example:
```
org.apache.pulsar.client.impl.conf.ProducerConfigurationData#batchedFilterProperties
```
The `batchedFilterProperties` type is `List<String>` with default value is emtpy list. For empty list, it means that the properties of entry's metadata is empty, and the `EntryFilter` will not take effect.
## Implementation
- When call `org.apache.pulsar.client.impl.BatchMessageContainerImpl#add`, we extract the message properties and add it to `metadata`:
```
public boolean add(MessageImpl<?> msg, SendCallback callback) {
if (log.isDebugEnabled()) {
log.debug("[{}] [{}] add message to batch, num messages in batch so far {}", topicName, producerName,
numMessagesInBatch);
}
if (++numMessagesInBatch == 1) {
try {
// some properties are common amongst the different messages in the batch, hence we just pick it up from
// the first message
messageMetadata.setSequenceId(msg.getSequenceId());
List<KeyValue> filterProperties = getProperties(msg);
if (!filterProperties.isEmpty()) {
messageMetadata.addAllProperties(filterProperties); // and message properties here
}
```
- Also we need add a method `hasSameProperties` like `hasSameSchema`. Messages with same properties can be added to the same batch:
```
private boolean canAddToCurrentBatch(MessageImpl<?> msg) {
return batchMessageContainer.haveEnoughSpace(msg)
&& (!isMultiSchemaEnabled(false)
|| batchMessageContainer.hasSameSchema(msg)
|| batchMessageContainer.hasSameProperties(msg))
&& batchMessageContainer.hasSameTxn(msg);
}
```
## Reject Alternatives
- Implement a `AbstractBatchMessageContainer` , saying `BatchMessagePropertiesBasedContainer`, keeping messages with same properties in a single `hashmap` entry, like `BatchMessageKeyBasedContainer`.
Rejection reason: This will publish messages out of order
--
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] asafm commented on issue #16680: PIP-191: Support batched message using entry filter
Posted by GitBox <gi...@apache.org>.
asafm commented on issue #16680:
URL: https://github.com/apache/pulsar/issues/16680#issuecomment-1200959212
Can you explain how in your proposal, you keep the exact ordering of the messages as they were sent to the producer while separating them into different batches based on properties?
--
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] AnonHxy commented on issue #16680: PIP-191: Support batched message using entry filter
Posted by GitBox <gi...@apache.org>.
AnonHxy commented on issue #16680:
URL: https://github.com/apache/pulsar/issues/16680#issuecomment-1192619832
> 1) It is not clear to me if you want to push all the messages metadata
in the main header of the entry or only the metatadata of the first
entry, or only the KEY.
I just want to copy the partial message properties(we call it batch properties here) to the main header of the entry, without modifying the payload or single message metadata. All messages in one batch must have the same batch properties (both key and value).
> 2) we have to clarify the new message format of the entry and update
the protocol documents
It seems that the protocol documents has not described the batch message main header. It just describes the `SingleMessageMetadata`. I agree with that It's better to update the protocol documents to describe the batch message main header.
> 3) existing EntryFilters won't be able to work well with the new
format, we must find a way to make them fail and not process garbage
It seems that there is no compatibility problems. Because I just add properties to the main header , which is empty before, and there is no modification for the payload and single message metadata. The existing EntryFilters can be able to work well with the new
format, the same as the existing Protocol Handlers. :)
--
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] AnonHxy commented on issue #16680: PIP-191: Support batched message using entry filter
Posted by GitBox <gi...@apache.org>.
AnonHxy commented on issue #16680:
URL: https://github.com/apache/pulsar/issues/16680#issuecomment-1265496662
> What ended up with this feature?
There was a VOTE[1] about this PIP and end up with 1 bingding -1 and 1 non-bing +1. So maybe this feature need more discussion or be canceled @asafm
* [1] https://lists.apache.org/thread/21900d06plmjx23z5ww8kkqkgdlwszh5
--
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] AnonHxy commented on issue #16680: PIP-191: Support batched message using entry filter
Posted by GitBox <gi...@apache.org>.
AnonHxy commented on issue #16680:
URL: https://github.com/apache/pulsar/issues/16680#issuecomment-1201245652
> Can you explain how in your proposal, you keep the exact ordering of the messages as they were sent to the producer while separating them into different batches based on properties?
Sure. There is no difference with normal batched sending, except we have to check `hasSameProperties` in `ProducerImpl#canAddToCurrentBatch`. I think the following legend can help explain this more clearly
<img width="905" alt="image" src="https://user-images.githubusercontent.com/10233437/182164813-16b1cf7a-024c-401f-8c54-862f9684013a.png">
Let me know if it is not clear enough or this doesn't explain the question correctly:) @asafm
--
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] asafm commented on issue #16680: PIP-191: Support batched message using entry filter
Posted by GitBox <gi...@apache.org>.
asafm commented on issue #16680:
URL: https://github.com/apache/pulsar/issues/16680#issuecomment-1195377352
Let me see if I understand correctly.
Today, when an entry filter receives an entry, it gets an Entry that has:
```java
public interface Entry {
/**
* @return the data
*/
byte[] getData();
byte[] getDataAndRelease();
/**
* @return the entry length in bytes
*/
int getLength();
/**
* @return the data buffer for the entry
*/
ByteBuf getDataBuffer();
/**
* @return the position at which the entry was stored
*/
Position getPosition();
/**
* @return ledgerId of the position
*/
long getLedgerId();
/**
* @return entryId of the position
*/
long getEntryId();
/**
* Release the resources (data) allocated for this entry and recycle if all the resources are deallocated (ref-count
* of data reached to 0).
*/
boolean release();
}
```
The Entry interface doesn't let you know if this is Batched Entry.
You also get FilterContext:
```java
@Data
public class FilterContext {
private Subscription subscription;
private MessageMetadata msgMetadata;
private Consumer consumer;
```
and in `MessageMetadata`, you have
```protobuf
// differentiate single and batch message metadata
optional int32 num_messages_in_batch = 11 [default = 1];
```
Which enables you to know this entry is batched.
The developer can determine what class would deserialize the entry byte array into a list of separate messages.
So currently, given the entry is batched, the filter author can act on it only by paying the cost of deserializing it, right?
You're saying we can alter the clients (all clients) to extract specific properties from each message and place those properties values in the message metadata of the Batched Entry. The filter can then use the values to decide if to reject/accept.
The problem is that if you have different values for a given property for each message in the batch, then the filter author can't provide a reject or accept for this entry since some messages are rejected, and some are accepted.
So the only solution offered in this suggestion is to change the way messages are batched and collect the records into a batch only if they have the same values for the properties configured to be extracted.
If this is ok and correct, I have notes on it:
1. All I wrote above is not clearly stated in the PIP. IMO the PIP needs to be modified to reflect that explanation.
2. As I wrote before, you are changing the core batching behavior of the client - the user needs to be fully aware of this and its implications. The user configures `batchedFilterProperties` without understanding the consequences of the altering of the batching behavior. One option may be to rename `batchedFilterProperties` to `batchGroupByProperties` so they will know the batching behavior is changing. I wouldn't specify the term `filter` here since the filter at this stage has no direct link.
3. Don't you need to introduce new knobs to control the memory? Up until now, you collected records into a batch and sent them. Now you collect into multiple batches until a certain threshold - won't this consume more memory? How can I control this as a user?
--
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] github-actions[bot] commented on issue #16680: PIP-191: Support batched message using entry filter
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on issue #16680:
URL: https://github.com/apache/pulsar/issues/16680#issuecomment-1301569291
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] AnonHxy commented on issue #16680: PIP-191: Support batched message using entry filter
Posted by GitBox <gi...@apache.org>.
AnonHxy commented on issue #16680:
URL: https://github.com/apache/pulsar/issues/16680#issuecomment-1193317816
Hi Asaf @asafm
> Is this true now or do you plan to enforce it?
This is what I plan to enforce it. The batch entry metadata properties is empty now
> What does this contain? Property names to extract from the messages to be placed in the batch entry metadata properties?
It contains the properties keys that will be put into the batch entry metadata properties.
For example:
- Let's set `batchedFilterProperties`=`<region, version>`
This means only key named `region` and `version`will be extracted to the batch meta properties
- Then we have a producer that sends the messges below in order:
- `msg1` with properties: `<region: eu>`
- `msg2` with properties: `<region: eu>`
- `msg3` with properties: `<region: eu, version:1, tag:a>`
- `msg4` with properties: `<region: eu, version:1>`
- `msg5` with properties: `<region: us, version:1>`
- `msg6` with properties: `<region: us, version:2>`
- The process of properties extraction will be:
- msg1 and msg2 have the same properties: <region: eu>, so they will put into the same batch
- msg3 and msg4 have the same properties: <region: eu, version:1>. tag:a in msg3 will be ignored because the `batchedFilterProperties` doesn't contains 'tag'. So msg3 and msg4 will put into the same batch.
- msg5 and msg6 have different properties, because the value of version is different. So we publish msg5 and msg6 with different batch.
- Just to summarize, the result will be:
| |batch meta properties | single meta properties | payload | single meta properties | payload |
|------- |---------------------- |------------------------------- |------ |-------------------------|------ |
|batch1 | <region: eu> | <region: eu> | msg1 | <region: eu> | msg2 |
|batch2 | <region: eu, version:1> | <region: eu, version:1, tag:a> | msg3 | <region: eu, version:1> | msg4 |
|batch3 | <region: us, version:1> | <region: us, version:1> | msg5 | | |
|batch4 | <region: us, version:2> | <region: us, version:2> | msg6 | | |
--
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] AnonHxy commented on issue #16680: PIP-191: Support batched message using entry filter
Posted by GitBox <gi...@apache.org>.
AnonHxy commented on issue #16680:
URL: https://github.com/apache/pulsar/issues/16680#issuecomment-1195274822
> What I am suggesting is to turn this idea a bit: Instead of batching per the same values of keys, batch all messages, and for each property save a list of values in the metadata. This way you are not affecting the latency on the client side or batch size, yet you still have the ability to filter batches based on metadata without deserializing it.
Thanks for your suggestion @asafm . If I understand correctly, it means that all the properties, which need to be filted, will be put into the entry header meta. So the protocol will be like below:
| | batch meta properties | single meta properties | payload | single meta properties | payload|
| -- | -- | -- | -- | --|---|
| batch | <region: eu,us; version:1> | <region: eu> | msg1 | <region: us; version:1> | msg2|
But this will not work will for entry filter I think, the reason is that:
For example, if our filter's condition is that :
```
if region==eu
return reject
else
return accept
```
so the expected result is that msg1 will be rejected and msg2 will be accepted. However because msg1 and msg2 are in the same entry, so the entry filter will not work well. In other words, entry filter doesn't support batched message now.
--
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] asafm commented on issue #16680: PIP-191: Support batched message using entry filter
Posted by GitBox <gi...@apache.org>.
asafm commented on issue #16680:
URL: https://github.com/apache/pulsar/issues/16680#issuecomment-1200406660
Ok, regarding the memory I see that `canEnqueueRequest` makes sure we're not breaching memory limits for the client prior to insertion.
Regarding the actual implementation, I'm a bit puzzled. Why re-use BatchMessageContainerImpl which was geared towards a single batch and make it multi batch? Why not do like the BatchMessageKeyBasedContainer and create your own? You wrote that it will send messages out of order, but once you batch them in separate batches concurrently , this by design will make them out of order no?
--
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] asafm commented on issue #16680: PIP-191: Support batched message using entry filter
Posted by GitBox <gi...@apache.org>.
asafm commented on issue #16680:
URL: https://github.com/apache/pulsar/issues/16680#issuecomment-1193305602
> All messages in one batch must have the same batch properties (both key and value
Is this true now or do you plan to enforce it?
> batchedFilterProperties
What does this contain? Property names to extract from the messages to be placed in the batch entry metadata properties?
Regarding the extraction. Say I have Message 1 with property region=eu, and Message 2 with property region=us. The end result of batch metadata properties will be region=eu, us?
The reason I'm asking is: Can you please clarify in the doc the process of properties extraction exactly.
--
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] asafm commented on issue #16680: PIP-191: Support batched message using entry filter
Posted by GitBox <gi...@apache.org>.
asafm commented on issue #16680:
URL: https://github.com/apache/pulsar/issues/16680#issuecomment-1265313811
What ended up with this feature?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] PIP-191: Support batched message using entry filter [pulsar]
Posted by "merlimat (via GitHub)" <gi...@apache.org>.
merlimat closed issue #16680: PIP-191: Support batched message using entry filter
URL: https://github.com/apache/pulsar/issues/16680
--
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] asafm commented on issue #16680: PIP-191: Support batched message using entry filter
Posted by GitBox <gi...@apache.org>.
asafm commented on issue #16680:
URL: https://github.com/apache/pulsar/issues/16680#issuecomment-1195218495
> > Might generate smaller batches or increase latency?
>
> Yes. This will generate smaller batches if `batchedFilterProperties ` is not empty. But the advantage is that there is no need to deserialize the entry in the broker or just publish non-batched messages if users want to fliter entries.
>
> > Why not have list of values for property: region=us,eu ?
>
> Good idea. Users should know which values they will use. It seems reasonable.
What I am suggesting is to turn this idea a bit: Instead of batching per the same values of keys, batch all messages, and for each property save a list of values in the metadata. This way you are not affecting the latency on the client side or batch size, yet you still have the ability to filter batches based on metadata without deserializing 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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] AnonHxy commented on issue #16680: PIP-191: Support batched message using entry filter
Posted by GitBox <gi...@apache.org>.
AnonHxy commented on issue #16680:
URL: https://github.com/apache/pulsar/issues/16680#issuecomment-1193334204
> Might generate smaller batches or increase latency?
Yes. This will generate smaller batches if `batchedFilterProperties ` is not empty. But the advantage is that there is no need to deserialize the entry in the broker or just publish non-batched messages if users want to fliter entries.
> Why not have list of values for property: region=us,eu ?
Good idea. Users should know which values they will use. It seems reasonable.
--
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] github-actions[bot] commented on issue #16680: PIP-191: Support batched message using entry filter
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on issue #16680:
URL: https://github.com/apache/pulsar/issues/16680#issuecomment-1242849902
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] AnonHxy commented on issue #16680: PIP-191: Support batched message using entry filter
Posted by GitBox <gi...@apache.org>.
AnonHxy commented on issue #16680:
URL: https://github.com/apache/pulsar/issues/16680#issuecomment-1200433052
> Why not do like the BatchMessageKeyBasedContainer and create your own
- I have considered this solution before. `BatchMessageKeyBasedContainer` doesn't keep message's order because it is just used for key_shared subscription mode. And the document about `BatchMessageKeyBasedContainer` describes it's batch action and we can see that it sends batched message out of the incoming messages order:
https://github.com/apache/pulsar/blob/eec0eb50074d54284e66f8c2f73d864691f05ec2/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageKeyBasedContainer.java#L30-L39
- If we implement a `BatchMessagePropertiesBasedContainer` like `BatchMessageKeyBasedContainer`, it has the same problem I think. For example:
```
* incoming single messages:
* (<version:1>, v1), (<region:2>, v1), (<region:3> v1), (<region:1>, v2), (<region:2>, v2), (<region:3>, v2), (<region:1>, v3), (<region:2>, v3), (<region:3>, v3)
*
* batched into multiple batch messages:
* [(<version:1>, v1), (<version:1>, v2), (<version:1>, v3)], [(<version:2>, v1), (<version:2>, v2), (<version:2>, v3)], [(<version:3>, v1), (<version:3>, v2), (<version:3>, v3)]
```
--
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] asafm commented on issue #16680: PIP-191: Support batched message using entry filter
Posted by GitBox <gi...@apache.org>.
asafm commented on issue #16680:
URL: https://github.com/apache/pulsar/issues/16680#issuecomment-1202678136
Excellent explanation @AnonHxy - I finally understand the whole mechanism.
So in effect, you're adding another trigger condition to send the batch, on top of the current max count, max size, and max delay: Once a message is requested to be added to a batch of its properties (as defined in the configuration) values are different from the records in the batch (i.e. 1st record properties values) than you trigger the batch flush (i.e send and clear).
So the side effect of this behavior is that you can easily end up with tiny batches, perhaps even 1 record per batch. There is a good chance once they turn this feature on, they will lose all performance benefits of batching since the batches will be very small. It completely depends on the distribution of values. It might be a big trade-off you're asking from the user: You might trade off the performance of write and perhaps read, for getting the ability to have the server-side filter work for batches.
1. I would for sure document that trade-off very clearly in the PIP and in the configuration page of the producer.
2. I would rephrase the explanation in this PIP to document the behavior:
> So the only solution is to change the way messages are batched and collect the records into a batch only if they have the same values for the properties configured to be extracted
-->
"So the only solution is to change the way messages are batched and collect the records into a batch only if they have the same values for the properties configured to be extracted. If a message is added to the producer and the properties are not the same as the batched records, it will trigger a send of this batch and start a new batch with that message. "
3. You need to emphasize throughout the document the trigger condition for sending once a message with different properties is added.
4. Now that I understand, the name I suggested does not fit since `batchGroupByProperties` makes you think you are grouping records in several batches by those properties, which you don't since you have only 1 in-flight batch. Maybe `restrictSameValuesInBatchProperties` ?
--
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] asafm commented on issue #16680: PIP-191: Support batched message using entry filter
Posted by GitBox <gi...@apache.org>.
asafm commented on issue #16680:
URL: https://github.com/apache/pulsar/issues/16680#issuecomment-1201289792
I'm genuinely asking again, as I still don't understand.
Say I send M1(`region=eu`) to the producer.
It creates a batch and adds M1 to it.
Now I send M2(`region=us`) to the producer.
It checks current batch 1st record properties are different from M2 properties; thus, it goes left (No) on your diagram. This part I don't understand.
I expected it at this stage to create a batch for `region=us` and add M2 to it.
But I see written in your diagram that you "batched msgs in msgsContainer" (which container? for which properties? Which msgs are you referring to?) and you "and send them" (why? what is the trigger to send them? Size/time?) and then you clear and add M2 to the batch.
I reasoned that you will a batch container per properties values set, so if I also send M3(`region=us`) and M4(`region=eu`), I will get:
Batch 1 (`region=eu`): M1, M4
Batch 2 (`region=us`): M2, M3
When either batch reaches the trigger point (size/time), they will be sent, no?
So once I understand the diagram, I might be able to figure out how it maintains order.
In general, once understood, I believe that these question marks need to be explained in the PIP.
--
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] AnonHxy commented on issue #16680: PIP-191: Support batched message using entry filter
Posted by GitBox <gi...@apache.org>.
AnonHxy commented on issue #16680:
URL: https://github.com/apache/pulsar/issues/16680#issuecomment-1203507452
> I would for sure document that trade-off very clearly in the PIP and in the configuration page of the producer.
OK. Thanks for your suggestions @asafm . I have update the document and add the "Trade-off" part. And I also clarify in this part that this proposal gives a big chance that batched message can also using entry filter, especially when you have konw the distrbution of values. It will bring benfit because entry filter dosen't support batch message currently.
--
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] AnonHxy commented on issue #16680: PIP-191: Support batched message using entry filter
Posted by GitBox <gi...@apache.org>.
AnonHxy commented on issue #16680:
URL: https://github.com/apache/pulsar/issues/16680#issuecomment-1202172999
> I expected it at this stage to create a batch for region=us and add M2 to it.
But I see written in your diagram that you "batched msgs in msgsContainer" (which container? for which properties? Which msgs are you referring to?) and you "and send them" (why? what is the trigger to send them? Size/time?) and then you clear and add M2 to the batch.
Let me try to explain the current batch process first. @asafm
`msgsContainer` in the above diagram refers to the `ProducerImpl#batchMessageContainer`, which has a list named `BatchMessageContainerImpl#messages`. And the `msgList` in the diagram refers to the `BatchMessageContainerImpl#messages`.
Also the `msgsContainer` has a `BatchMessageContainerImpl#messageMetadata`, which will save the first messages metadata in `msgList`. Note that the properties field of `BatchMessageContainerImpl#messageMetada` is always empty currently.
A `ProducerImpl` instance has only one `msgsContainer`. When a producer publishs messages with batch enable, the messages will first be put into the `msgList`. When the `msgsContainer` doesn't has enougth space(size or number of messages threshold) or flush task scheduled (time threshold), the the `msgsContainer` will be triggered to batch the messages in `msgList` and send them. And then the `msgList` will be cleared to prepare the next batch.
The "size or number of messages threshold" check action happens at `ProducerImpl#canAddToCurrentBatch` method, before put message into the `msgList`. If this check result is false, the `msgsContainer` batch action will be triggered. See line636 and line668
https://github.com/apache/pulsar/blob/eddbdd8c5b2bde868758b7cbb3c4e7c11e87e097/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java#L636-L669
All above is current batch sending process.
The key of keeping order is that, there will be only one batch exist at the same time, and the batch will be send to channel in a single thread pool. The single thread pool is the `SingleThreadEventExecutor` attach to the eventloop: line2154
https://github.com/apache/pulsar/blob/eddbdd8c5b2bde868758b7cbb3c4e7c11e87e097/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java#L2145-L2155
What I will do in this proposal just is:
- Extract the first message properties in the batch and fill into the `BatchMessageContainerImpl#messageMetada`
- Check if the sending message has same properties with the properties in `BatchMessageContainerImpl#messageMetada` additionally in `ProducerImpl#canAddToCurrentBatch` method.
So regarding the case you gave above, the process will be:
- Sending M1(region=eu):
- `ProducerImpl#canAddToCurrentBatch` return true
- Add M1 to `msgList`, and fill `(region=eu):` to `BatchMessageContainerImpl#messageMetada`.
- Sending M2(region=us)
- `ProducerImpl#canAddToCurrentBatch` return false. Because it has different properties with `BatchMessageContainerImpl#messageMetada`.
- This will trigger `doBatchSendAndAdd`
- The `msgContainer` will batch M1(`msgList` only has M1) and send it, then clear the `msgList` to prepare next batch.
- Add M2(region=us) to `msgList`, and fill `(region=us):` to `BatchMessageContainerImpl#messageMetada`.
- Sending .....
--
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] AnonHxy commented on issue #16680: PIP-191: Support batched message using entry filter
Posted by GitBox <gi...@apache.org>.
AnonHxy commented on issue #16680:
URL: https://github.com/apache/pulsar/issues/16680#issuecomment-1196210743
Thank you for your patience @asafm . Yes, what you described above is correctly.
> All I wrote above is not clearly stated in the PIP. IMO the PIP needs to be modified to reflect that explanation.
I will give a more detailed explanation in the PIP
> One option may be to rename `batchedFilterProperties` to `batchGroupByProperties` so they will know the batching behavior is changing
I agree with you. `batchGroupByProperties` is better, which indicates batching behavior.
> Don't you need to introduce new knobs to control the memory?
No. There is no need to take care of controling the memory in this PIP. Because the groupBy behavior will be done in the `BatchMessageContainerImpl`, which will control the memory.
For details:
We can add a method named `hasSameProperties` in `BatchMessageContainerImpl`, which will looks like the exsited method `BatchMessageContainerImpl#hasSameSchema`:
```
public boolean hasSameProperties(MessageImpl<?> msg) {
if (numMessagesInBatch == 0) {
return true;
}
if (!messageMetadata.getPropertiesList().isEmpty()) {
return getProperties(msg).isEmpty();
}
return getProperties(msg).equals(messageMetadata.getPropertiesList());
}
```
And `hasSameProperties` will be invoked when producer try to add messages in the `BatchMessageContainerImpl`:
```
private boolean canAddToCurrentBatch(MessageImpl<?> msg) {
return batchMessageContainer.haveEnoughSpace(msg) // memory control is here
&& (!isMultiSchemaEnabled(false) || batchMessageContainer.hasSameSchema(msg))
&& batchMessageContainer.hasSameProperties(msg) // invoke it here
&& batchMessageContainer.hasSameTxn(msg);
}
```
As we see above the `batchMessageContainer` has already checked the space.
--
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] eolivelli commented on issue #16680: PIP-191: Support batched message using entry filter
Posted by GitBox <gi...@apache.org>.
eolivelli commented on issue #16680:
URL: https://github.com/apache/pulsar/issues/16680#issuecomment-1191420830
I understand the problem, and I have already thought about it, because I am the author of some filters (especially the JMS Selectors Filter) but we have to clarify more this PIP.
1. It is not clear to me if you want to push all the messages metadata in the main header of the entry or only the metatadata of the first entry, or only the KEY.
2. we have to clarify the new message format of the entry and update the protocol documents
3. existing EntryFilters won't be able to work well with the new format, we must find a way to make them fail and not process garbage
4. the same problem applies to Interceptors and Protocol Handlers (like KOP), we must make it clear in this PIP what is the upgrade path and give some suggestions to the developers of such components
--
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] asafm commented on issue #16680: PIP-191: Support batched message using entry filter
Posted by GitBox <gi...@apache.org>.
asafm commented on issue #16680:
URL: https://github.com/apache/pulsar/issues/16680#issuecomment-1193321676
Ok, what you just wrote, if I understand correctly is pretty big and must be well documented in the proposal. You're saying that you will change the way batching works today. If today I have certain knobs which controls how batching is preparing: max batch size, timeout, max number of records and that's it, now you're changing it such that the batching will grouped by the properties defined (`batchedFilterProperties`). This might have serious performance hit not? Might generate smaller batches or increase latency?
Why not have list of values for property: region=us,eu ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] PIP-191: Support batched message using entry filter [pulsar]
Posted by "merlimat (via GitHub)" <gi...@apache.org>.
merlimat commented on issue #16680:
URL: https://github.com/apache/pulsar/issues/16680#issuecomment-6377570930
Thanks @AnonHxy for the proposal and for working through the review. The PIP-191 vote (https://lists.apache.org/thread/21900d06plmjx23z5ww8kkqkgdlwszh5) got a binding -1 and didn't pass, and there's been no activity since 2022, so I'm closing this as not planned. Anyone is welcome to restart it as a pip/ document following pip/README.md, taking the review feedback here and on the vote thread into account.
--
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