Make copies of thread local MessageMetadata when it might be shared - #14556
Make copies of thread local MessageMetadata when it might be shared#14556lhotari wants to merge 4 commits into
Conversation
…ight be shared to other threads - sharing the thread local MessageMetadata instance to other threads will cause issues - Thread local variable is org.apache.pulsar.common.protocol.Commands#LOCAL_MESSAGE_METADATA
|
@lhotari:Thanks for your contribution. For this PR, do we need to update docs? |
|
@lhotari:Thanks for providing doc info! |
| msgMetadata = msgMetadata == null | ||
| ? Commands.peekMessageMetadata(metadataAndPayload, subscription.toString(), -1) | ||
| : msgMetadata; | ||
| if (msgMetadata == null) { |
There was a problem hiding this comment.
Looks like we will not share msgMetadata to other threads? I have checked where used the msgMetadata, they should run in the same thread. Maybe I missed something.
It's a good motivation, is it better to avoid passing the msgMetadata? Instead, we can create local variable, looks like
boolean hasTxnidMostBits = msgMetadata.hasTxnidMostBits();
int batchSize = msgMetadata.getNumMessagesInBatch();
msgMetadata.clear();
// The followings should use the `hasTxnidMostBits` and `batchSize` directly.
There was a problem hiding this comment.
It's a good motivation, is it better to avoid passing the msgMetadata? Instead, we can create local variable, looks like
yes, avoiding the instance is a better choice. I'll take a look in the refactoring.
There was a problem hiding this comment.
I took a look in the possible refactoring. there's a number of fields that are used from MessageMetadata, so it would turn out to be an ugly solution.
It's possible that a copy isn't necessary in this case, but it's hard to tell without checking all accesses. That's why a "defensive copy" is reasonable.
There was a problem hiding this comment.
Ok, looks like we should try our best to avoid passing the MessageMetadata to another method, If we have to pass it to another method, we should make a copy.
For example
Use deliverAtTime is enough.
There was a problem hiding this comment.
If we have to pass it to another method, we should make a copy.
I guess that would be one way to define a rule when to make a copy. (btw. I wonder if it's intentional that the generated copyFrom code doesn't call "clear" as the first step internally. )
- this would fail with NPE if metadata ever was null
There was a problem hiding this comment.
when #14436 has been fixed. I will approve this. because I'm worried about the new problem will be happend. So I leave request changes, prevent it from being merged
|
|
||
| public static final FilterContext FILTER_CONTEXT_DISABLED = new FilterContext(); | ||
|
|
||
| public MessageMetadata getMsgMetadata() { |
There was a problem hiding this comment.
This may be problematic because the filter may want to hold a reference to this object (I don't know why but you know users sometimes do silly things)
There was a problem hiding this comment.
Please elaborate about the problem. What do you suggest as a solution?
| context.consumerEpoch = consumerEpoch; | ||
| context.brokerEntryMetadata = brokerEntryMetadata; | ||
| context.messageMetadata = messageMetadata; | ||
| context.messageMetadata.copyFrom(messageMetadata); |
There was a problem hiding this comment.
This may be problematic because the filter may want to hold a reference to this object (I don't know why but you know users sometimes do silly things)
@congbobo184 What new problems could you see happening with these changes? |
I don't find new problem and #14436 always exist.I don't know what time the msgMetadata is null. |
|
The pr had no activity for 30 days, mark with Stale label. |
|
The pr had no activity for 30 days, mark with Stale label. |
|
It seems #15983 has addressed the issue. |
8 participants
Motivation
Additional context
Modifications