[Bug] The RawMessageImpl.getProperties throw `Duplicate key` IllegalStateException.
- Dominant language
- Java
- Stars
- 15.3k
- Forks
- 3.8k
- Avg merge
- 1d 14h
- Merged PRs (30d)
- 160
Description
### Search before asking
- [x] I searched in the [issues](https://github.com/apache/pulsar/issues) and found nothing similar.
### Read release policy
- [x] I understand that unsupported versions don't get bug fixes. I will attempt to reproduce the issue on a supported version of Pulsar client and Pulsar broker.
### Version
master
### Minimal reproduce step
You can run this code to reproduce it.
```
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import java.util.Map;
import org.apache.pulsar.common.api.proto.KeyValue;
import org.apache.pulsar.common.api.proto.MessageMetadata;
import org.apache.pulsar.common.api.raw.RawMessage;
import org.apache.pulsar.common.api.raw.RawMessageImpl;
import org.apache.pulsar.common.api.raw.ReferenceCountedMessageMetadata;
import org.apache.pulsar.shade.io.netty.buffer.ByteBufAllocator;
public class Test {
public static void main(String[] args) {
byte[] messageContent = "Hello, Pulsar!".getBytes();
ByteBuf payload = Unpooled.wrappedBuffer(messageContent);
org.apache.pulsar.shade.io.netty.buffer.ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer();
ReferenceCountedMessageMetadata referenceCountedMessageMetadata = ReferenceCountedMessageMetadata.get(buffer);
MessageMetadata metadata1 = referenceCountedMessageMetadata.getMetadata();
KeyValue keyValue = metadata1.addProperty();
keyValue.setKey("aa");
keyValue.setValue("bb");
KeyValue keyValue1 = metadata1.addProperty();
keyValue1.setKey("aa");
keyValue1.setValue("cc");
RawMessage rawMessage = RawMessageImpl.get(referenceCountedMessageMetadata, null, null, 0, 0, 0);
Map properties = rawMessage.getProperties();
}
}
```
### What did you expect to see?
Do not throw the exception.
### What did you see instead?
It will throw the following exception.
```
Exception in thread "main" java.lang.IllegalStateException: Duplicate key aa (attempted merging values bb and cc)
at java.base/java.util.stream.Collectors.duplicateKeyException(Collectors.java:135)
at java.base/java.util.stream.Collectors.lambda$uniqKeysMapAccumulator$1(Collectors.java:182)
at java.base/java.util.stream.ReduceOps$3ReducingSink.accept(ReduceOps.java:169)
at java.base/java.util.ArrayList$SubList$2.forEachRemaining(ArrayList.java:1481)
at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509)
at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499)
at java.base/java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:921)
at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
at java.base/java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:682)
at org.apache.pulsar.common.api.raw.RawMessageImpl.getProperties(RawMessageImpl.java:102)
at com.zy.Test.main(Test.java:38)
```
### Anything else?
_No response_
### Are you willing to submit a PR?
- [x] I'm willing to submit a PR!
Contributor guide
Research direction
Start at RawMessageImpl.getProperties(), identified at line 102 in the stack trace, and run the provided Java reproducer with duplicate metadata keys. Check the existing raw-message behavior and add coverage for this case; done means getProperties() no longer throws for the reproduced input.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- distributed-systems
- Issue type
- Bug
- Difficulty
- 2/5
- Estimated time
- 1-3 hours
- Activity status
- Stale
- Clarity
- Clearly specified
- Newbie friendliness
- 48/100