[Bug] The RawMessageImpl.getProperties throw Duplicate key IllegalStateException.

[horizonzy]

Search before asking

• I searched in the issues and found nothing similar.

Read release policy

• 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<String, String> 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?

• I'm willing to submit a PR!