[fix][client] Cache empty schema version in ProducerImpl schemaCache.

[lifepuzzlefun]

Fixes #19928

Motivation

When use Schema.BYTES consumer and enable retry and dlq policy.

need cache the empty schema version to avoid send a lot of GetOrCreateSchema request to retrieve the schema version.

Modifications

1. Cache empty schema version in ProducerImpl schemaCache.

2. When read from schemaCache, check if the schema version means empty.

Verifying this change

• Make sure that the change passes the CI checks.

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository:

[lifepuzzlefun]

Why the BYTES schema consumer's retry producer and dlq consumer keeps sending GetOrCreateSchema to broker?

1. the retry producer create with Schema.AUTO_PRODUCE_BYTES(Schema.BYTES)
   

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

   Lines 613 to 619 in ec102fb

   	if (retryLetterProducer == null) {
   	retryLetterProducer = client.newProducer(Schema.AUTO_PRODUCE_BYTES(schema))
   	.topic(this.deadLetterPolicy.getRetryLetterTopic())
   	.enableBatching(false)
   	.blockIfQueueFull(false)
   	.create();
   	}

2. when message send use Schema.AUTO_PRODUCE_BYTES(retryMessage.getReaderSchema()) schema.
   in this case retryMessage.getReaderSchema() is Schema.BYTES

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

Lines 653 to 660 in ec102fb

	deadLetterProducer.thenAccept(dlqProducer -> {
	TypedMessageBuilder<byte[]> typedMessageBuilderNew =
	dlqProducer.newMessage(Schema.AUTO_PRODUCE_BYTES(retryMessage.getReaderSchema().get()))
	.value(retryMessage.getData())
	.properties(propertiesMap);
	typedMessageBuilderNew.sendAsync().thenAccept(msgId -> {
	doAcknowledge(finalMessageId, ackType, Collections.emptyMap(), null).thenAccept(v -> {
	result.complete(null);

3. the method check both schema not equal. so an GetOrCreateSchema rpc is send to broker.
   

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

   Lines 713 to 721 in ec102fb

   	@VisibleForTesting
   	boolean populateMessageSchema(MessageImpl msg, SendCallback callback) {
   	MessageMetadata msgMetadataBuilder = msg.getMessageBuilder();
   	if (msg.getSchemaInternal() == schema) {
   	schemaVersion.ifPresent(v -> msgMetadataBuilder.setSchemaVersion(v));
   	msg.setSchemaState(MessageImpl.SchemaState.Ready);
   	return true;
   	}
   	// If the message is from the replicator and without replicated schema

4. broker return schemaVersion is an empty byte array (BYTES schema not support version).

5. client has logic when broker return schemaVersion is an empty byte array. not cache it in ProducerImpl.schemaCache.
   

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

   Lines 770 to 780 in ec102fb

   	} else {
   	log.info("[{}] [{}] GetOrCreateSchema succeed", topic, producerName);
   	// In broker, if schema version is an empty byte array, it means the topic doesn't have schema. In this
   	// case, we should not cache the schema version so that the schema version of the message metadata will
   	// be null, instead of an empty array.
   	if (v.length != 0) {
   	schemaCache.putIfAbsent(msg.getSchemaHash(), v);
   	msg.getMessageBuilder().setSchemaVersion(v);
   	}
   	msg.setSchemaState(MessageImpl.SchemaState.Ready);
   	}

so when each message is send to broker. an GetOrCreateSchema rpc is also send to broker.

[lifepuzzlefun]

@BewareMyPower @poorbarcode @Technoboy- can you take a look ? this is a problem occur on master code and in our prod environment.

[BewareMyPower]

Could you also add a unit test for it? I think we can use a mocked ServerCnx to record the number of the CommandGetOrCreateSchema requests received.

[Technoboy-]

Please also add the test mentioned in #19928

[lifepuzzlefun]

@BewareMyPower @Technoboy- unit test added, please take a look

[lifepuzzlefun]

ping @BewareMyPower @Technoboy- @congbobo184 @liangyepianzhou can you take a look ? thank you : - )

[poorbarcode]

Since we will start the RC version of 3.0.0 on 2023-04-11, I will change the label/milestone of PR who have not been merged.

• The PR of type feature is deferred to 3.1.0
• The PR of type fix is deferred to 3.0.1

So drag this PR to 3.0.1

[codecov-commenter]

Codecov Report

Merging #19929 (cec104b) into master (329e80b) will increase coverage by 11.35%.
The diff coverage is 100.00%.

@@              Coverage Diff              @@
##             master   #19929       +/-   ##
=============================================
+ Coverage     61.50%   72.85%   +11.35%     
- Complexity    25755    31623     +5868     
=============================================
  Files          1859     1861        +2     
  Lines        136792   137503      +711     
  Branches      15043    15143      +100     
=============================================
+ Hits          84129   100177    +16048     
+ Misses        44857    29366    -15491     
- Partials       7806     7960      +154     

Flag	Coverage Δ
inttests	24.39% <20.00%> (-0.09%)	⬇️
systests	25.06% <80.00%> (+0.04%)	⬆️
unittests	72.12% <100.00%> (+13.48%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
...va/org/apache/pulsar/client/impl/ProducerImpl.java	83.16% <100.00%> (+9.18%)	⬆️

... and 637 files with indirect coverage changes