[fix][common]Fix presto sql avro decode error when publish non-batched msgs

[AnonHxy]

Motivation

Finally I find the bug, which makes org.apache.pulsar.tests.integration.presto.TestBasicPresto#testForSchema fail, But I still don't know why other PRs can run it success

• This PR will fix Flaky-test: org.apache.pulsar.tests.integration.presto.TestBasicPresto.testForSchema #16975 OR [Pulsar SQL] Query key-value schema data error #9704
• The root cause is that:

  • The keyByteBuf we get from currentMessage is not correct (line545 in code block 1), because we didn't decode the key with Base64 when the message is non-batched published. Explain in details:

    • As we know SingleMessageMetadata is only set when publish batched messages. So we will use RawMessageImpl#singleMessageMetadata to decode the key when messages is batched, or use RawMessageImpl#msgMetadata if messages in non-batched, As line174 in code block 2.
    • The problem is that RawMessageImpl#singleMessageMetadata will never be null (line38 code block 3)
    • If we try to decode non-batched message, the line175 in code block 2 will execute and return false.
    • Then line166 in code block 2 will execute and decode the key without Base64

    code block 1: [PulsarRecordCursor.java]:
    

    pulsar/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarRecordCursor.java

    Lines 542 to 547 in 70020f1

    	int valueLength = dataPayload.readInt();
    	valueByteBuf = dataPayload.readSlice(valueLength);
    	} else {
    	keyByteBuf = this.currentMessage.getKeyBytes().get();
    	valueByteBuf = this.currentMessage.getData();
    	}

    code block 2 [step into currentMessage.getKeyBytes() in RawMessageImpl.java]：
    

    pulsar/pulsar-common/src/main/java/org/apache/pulsar/common/api/raw/RawMessageImpl.java

    Lines 161 to 178 in 70020f1

    	public Optional<ByteBuf> getKeyBytes() {
    	if (getKey().isPresent()) {
    	if (hasBase64EncodedKey()) {
    	return Optional.of(Unpooled.wrappedBuffer(Base64.getDecoder().decode(getKey().get())));
    	} else {
    	return Optional.of(Unpooled.wrappedBuffer(getKey().get().getBytes(StandardCharsets.UTF_8)));
    	}
    	}
    	return Optional.empty();
    	}
    	@Override
    	public boolean hasBase64EncodedKey() {
    	if (singleMessageMetadata != null) {
    	return singleMessageMetadata.isPartitionKeyB64Encoded();
    	}
    	return msgMetadata.getMetadata().isPartitionKeyB64Encoded();
    	}

    code block 3 [RawMessageImpl.java]
    

    pulsar/pulsar-common/src/main/java/org/apache/pulsar/common/api/raw/RawMessageImpl.java

    Lines 34 to 41 in 70020f1

    	public class RawMessageImpl implements RawMessage {
    	private final RawMessageIdImpl messageId = new RawMessageIdImpl();
    	private ReferenceCountedMessageMetadata msgMetadata;
    	private final SingleMessageMetadata singleMessageMetadata = new SingleMessageMetadata();
    	private ByteBuf payload;

Modifications

• Add boolean setSingleMessageMetadata in org.apache.pulsar.common.api.raw.RawMessageImpl in order to check if
  singleMessageMetadata is set or not

Verifying this change

• Make sure that the change passes the CI checks.
• org.apache.pulsar.tests.integration.presto.TestBasicPresto#testForSchema has covered this change
• Add UT org.apache.pulsar.common.api.raw.RawMessageImplTest#testNonBatchedMessage
• Add UT org.apache.pulsar.common.api.raw.RawMessageImplTest#testBatchedMessage

Documentation

Check the box below or label this PR directly.

Need to update docs?

• doc-required
  (Your PR needs to update docs and you will update later)

• doc-not-needed
  (Please explain why)

• doc
  (Your PR contains doc changes)

• doc-complete
  (Docs have been already added)

[AnonHxy]

/pulsarbot run-failure-checks

[AnonHxy]

@codelipenghui @gaoran10 PTAL~

[tisonkun]

Perhaps we can avoid introducing a new boolean flag associated:

diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/api/raw/RawMes
sageImpl.java b/pulsar-common/src/main/java/org/apache/pulsar/common/api/raw/Raw
MessageImpl.java
index 9f1d493247f..bc2177a25ed 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/api/raw/RawMessageImp
l.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/api/raw/RawMessageImp
l.java
@@ -36,7 +36,7 @@ public class RawMessageImpl implements RawMessage {
     private final RawMessageIdImpl messageId = new RawMessageIdImpl();
 
     private ReferenceCountedMessageMetadata msgMetadata;
-    private final SingleMessageMetadata singleMessageMetadata = new SingleMessa
geMetadata();
+    private volatile SingleMessageMetadata singleMessageMetadata;
     private ByteBuf payload;
 
     private static final Recycler<RawMessageImpl> RECYCLER = new Recycler<RawMessageImpl>() {
@@ -56,7 +56,7 @@ public class RawMessageImpl implements RawMessage {
     public void release() {
         msgMetadata.release();
         msgMetadata = null;
-        singleMessageMetadata.clear();
+        singleMessageMetadata = null;
 
         payload.release();
         handle.recycle(this);
@@ -71,7 +71,7 @@ public class RawMessageImpl implements RawMessage {
         msg.msgMetadata.retain();
 
         if (singleMessageMetadata != null) {
-            msg.singleMessageMetadata.copyFrom(singleMessageMetadata);
+            msg.singleMessageMetadata = singleMessageMetadata;
         }
         msg.messageId.ledgerId = ledgerId;
         msg.messageId.entryId = entryId;

... while I'm unsure whether we can hold the reference, or we should deep copy. It seems we can barely hold the ref from my first glance, though.

[AnonHxy]

It seems we can barely hold the ref from my first glance, though

We didn't know weather RawMessageImpl#singleMessageMetadata is set value or has been cleared, through SingleMessageMetadata's properties

There is a field SingleMessageMetadata#_bitField0, which value of 0 means SingleMessageMetadata didn't set any value. But it is private accessible, we can not use this flag @tisonkun

[tisonkun]

@AnonHxy please take a closer look at the patch, it sets singleMessageMetadata to null when release. Then you don't have to modify other code in the file and we can use singleMessageMetadata == null checker.

[AnonHxy]

Now I got that @tisonkun . But I think we have to use deep copy here. The reason is that:

• There is only one SingleMessageMetadata instance in the thread, which get from threadlocal(Line163). And the
  processor maybe process in another thread(Line173).

• If we didn't deep copy the SingleMessageMetadata, the SingleMessageMetadata could be modified while
  processor processes the RawMessageImpl.

Although we can create a new SingleMessageMetadata every time when RawMessageImpl#get and deep copy to it, which will save the boolean flag, I think that it is not a good way, because this way will create a lot of SingleMessageMetadata instance.

pulsar/pulsar-common/src/main/java/org/apache/pulsar/common/api/raw/MessageParser.java

Lines 157 to 175 in 70020f1

	private static void receiveIndividualMessagesFromBatch(ReferenceCountedMessageMetadata msgMetadata,
	ByteBuf uncompressedPayload, long ledgerId, long entryId, MessageProcessor processor) {
	int batchSize = msgMetadata.getMetadata().getNumMessagesInBatch();
	try {
	for (int i = 0; i < batchSize; ++i) {
	SingleMessageMetadata singleMessageMetadata = LOCAL_SINGLE_MESSAGE_METADATA.get();
	ByteBuf singleMessagePayload = Commands.deSerializeSingleMessageInBatch(uncompressedPayload,
	singleMessageMetadata, i, batchSize);
	if (singleMessageMetadata.isCompactedOut()) {
	// message has been compacted out, so don't send to the user
	singleMessagePayload.release();
	continue;
	}
	processor.process(RawMessageImpl.get(msgMetadata, singleMessageMetadata, singleMessagePayload,
	ledgerId, entryId, i));
	}

[tisonkun]

Thanks for your explanation @AnonHxy!

LGTM.

Still, I don't like the way introduce an associated boolean flag, so you may think of deep copy based on my suggested patch - that is, set to null when release, create a new SingleMessageMetadata and copyForm the parameter when get. This way may have short coming on object reuse so it's not a net win. I'm OK with either approach.

[codelipenghui]

The fix looks good, could you please help add test to avoid the regression?

[AnonHxy]

The fix looks good, could you please help add test to avoid the regression?

OK, have add UT, PTAL @codelipenghui

[AnonHxy]

/pulsarbot run-failure-checks

[AnonHxy]

/pulsarbot run-failure-checks

[AnonHxy]

@Jason918 @Technoboy- Could you also please help take a look

[gaoran10]

More context

Before this change we can use this singleMessageMetadata != null to judge the message is batched or not, it seems that after this change, the test should be failed. It's strange.

[AnonHxy]

The modification in pulsar-common/src/main/java/org/apache/pulsar/common/api/raw/RawMessageImpl.java as show below changed the behavior.

Before this modification, the singleMessageMetadata could be null if the method param is null. But after this modification, the singleMessageMetadata will never be null, and it uses copyFrom to set value. @gaoran10