[Java Client] Fix messages sent by producers without schema cannot be decoded

[BewareMyPower]

Motivation

When I tried to consume a topic via a consumer with Avro schema while
the topic was produced by a producer without schema, the consumption
failed. It's because MultiVersionSchemaInfoProvider#getSchemaByVersion
doesn't check if schemaVersion is an empty byte array. If yes, a
BytesSchemaVersion of an empty array will be passed to cache.get and
then passed to loadSchema.

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/MultiVersionSchemaInfoProvider.java

Lines 94 to 98 in f90ef9c

	private CompletableFuture<SchemaInfo> loadSchema(byte[] schemaVersion) {
	return pulsarClient.getLookup()
	.getSchema(topicName, schemaVersion)
	.thenApply(o -> o.orElse(null));
	}

However, LookupService#getSchema cannot accept an empty byte array as
the version, so loadSchema failed.

The root cause is that the schema version was set unexpectly when
messages were sent by a producer without schema. At broker side, the
returned schema version is never null. If the schema version was an
empty array, then it means the message doesn't have schema. However, at
Java client side, the empty byte array is treated as an existing schema
and the schema version field will be set. When consumer receives the
message, it will try to load schema whose version is an empty array.

Modifications

• When a producer receives a response whose schema version is an empty
  byte array, just ignore it.
• Make MesasgeImpl#getSchemaVersion return null if the schema version
  is an empty byte array so that the consumer can consume messages
  produced by older version producers without schema. And return the
  internal schema for getRegetReaderSchema when getSchemaVersion
  returns null.
• Fix the existing tests. Since producer without schema won't set the
  schema_version field after this patch, some tests that rely on the
  precise stats should be modified.
• Add testConsumeAvroMessagesWithoutSchema to cover the case that
  messages without schema are compatible with the schema.

This patch also modifies the existing behavior when
schemaValidationEnforced is false and messages are produced by a
producer without schema and consumed by a consumer with schema.

1. If the message is incompatible with the schema

   • Before: getSchemaVersion returns an empty array and getValue
     fails with SerializationException:

     org.apache.commons.lang3.SerializationException: Failed at fetching schema info for EMPTY

   • After: getSchemaVersion returns null and getValue fails with
     SchemaSerializationException.

2. Otherwise (the message is compatible with the schema)

   • Before: getSchemaVersion returns an empty array and getValue
     fails with SerializationException.
   • After: getSchemaVersion returns null and getValue returns the
     correctly decoded object.

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)

• no-need-doc
  (Please explain why)

• doc
  (Your PR contains doc changes)

• doc-added
  (Docs have been already added)

[BewareMyPower]

I'm not sure whether this fix should be cherry-picked to older branches since it changes the current behavior (even if I think it's correct).

Since it's a fix at producer side, for messages produced by older version producer, even if they are compatible with the schema, the consumer still cannot consume them. In this PR, I also checked schema.length == 0 in MultiVersionSchemaInfoProvider#getSchemaByVersion. However, it means if the changes of producer were reverted, the tests could still pass.

[BewareMyPower]

I found #14626 also tried to fix the similar issue but it didn't fix in a correct way. Since #14626 has already been cherry-picked to branch-2.9 and branch-2.10, I also added the same labels.

[BewareMyPower]

        if (version != null && version.length == 0) {
            schemaFuture.completeExceptionally(new SchemaSerializationException("Empty schema version"));
            return schemaFuture;
        }

This error message added in #14626 could make users more confused. A SchemaSerializationException should represents the error when a byte array is serialized by an Avro schema. "Empty schema version" doesn't indicate anything because schema version is not a part of the message itself.

[Technoboy-]

I found #14626 also tried to fix the similar issue but it didn't fix in a correct way. Since #14261 has already been cherry-picked to branch-2.9 and branch-2.10, I also added the same labels.

Why it didn't fix in a correct way ?

[codelipenghui]

@BewareMyPower @Technoboy- I think this one is related to the discussion in the mailing list?

[BewareMyPower]

@codelipenghui Thanks, I will reply in the email.

@Technoboy- Because it still tried to perform topic look up with an empty byte array schema version, which always throws an exception before decoding the message value. #14626 just modifies the error message caused by the unexpected look up. See more details in the PR description and my previous explanation.

[BewareMyPower]

There are still other failed tests caused by this change, I'll fix them soon.

[BewareMyPower]

@codelipenghui @congbobo184 @Technoboy- @mattisonchao Now all required tests passed, PTAL.