fix parseMessageMetadata error cause by not skip broker entry metadata

[aloyszhang]

Fixes #10967

Motivation

fix parseMessageMetadata error cause by not skip broker entry metadata

Modifications

skip broker entry metadata if exist before parsing message metadata

[aloyszhang]

/pulsarbot run-failure-checks

[lhotari]

@aloyszhang Would it be able to add some tests that verify the behavior? A failing test is a good starting point for fixing a bug.

[lhotari]

It seems that skipBrokerEntryMetadataIfExist is called separately in many locations in Pulsar code base, before calling parseMessageMetadata method. For example:

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java

Lines 298 to 313 in 18f2f4a

	public static MessageImpl<byte[]> deserializeSkipBrokerEntryMetaData(
	ByteBuf headersAndPayloadWithBrokerEntryMetadata) throws IOException {
	@SuppressWarnings("unchecked")
	MessageImpl<byte[]> msg = (MessageImpl<byte[]>) RECYCLER.get();
	Commands.skipBrokerEntryMetadataIfExist(headersAndPayloadWithBrokerEntryMetadata);
	Commands.parseMessageMetadata(headersAndPayloadWithBrokerEntryMetadata, msg.msgMetadata);
	msg.payload = headersAndPayloadWithBrokerEntryMetadata;
	msg.messageId = null;
	msg.topic = null;
	msg.cnx = null;
	msg.properties = Collections.emptyMap();
	msg.brokerEntryMetadata = null;
	return msg;
	}

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java

Lines 2454 to 2455 in 202da11

	Commands.skipBrokerEntryMetadataIfExist(metadataAndPayload);
	MessageMetadata metadata = Commands.parseMessageMetadata(metadataAndPayload);

pulsar/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java

Lines 1670 to 1671 in f25b4a5

	skipBrokerEntryMetadataIfExist(metadataAndPayload);
	MessageMetadata metadata = Commands.parseMessageMetadata(metadataAndPayload);

pulsar/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java

Lines 1685 to 1686 in f25b4a5

	skipBrokerEntryMetadataIfExist(metadataAndPayload);
	MessageMetadata metadata = Commands.parseMessageMetadata(metadataAndPayload);

It seems that this inconsistency would have to be resolved.

[BewareMyPower]

I agree with @lhotari, the extra call of skipBrokerEntryMetadataIfExist should be removed.

Regarding to the unit tests, I think the existed tests already cover the cases when broker entry metadata is not enabled. For the case that broker entry metadata is enabled, some tests have been added to BrokerEntryMetadataE2ETest that could also cover the case. IMO, you can only add tests to cover #10950 since this PR could solve the problem completely.

/cc @wuzhanpeng

[aloyszhang]

@BewareMyPower @lhotari I'll remove the extra call of skipBrokerEntryMetadataIfExist and add some tests.

[aloyszhang]

/pulsarbot run-failure-checks

[aloyszhang]

@lhotari @BewareMyPower PTAL

[lhotari]

LGTM