Fix inconsistent prompt message when schema version is empty using AVRO.

[Technoboy-]

Motivation

When create a topic with schema - AVRO, if a producer sends bytes data directly to the topic, a consumer with HTTP and TCP lookup has different prompt message. See below test:

# Consume
Consumer<User> consumer = pulsarClient
                .newConsumer(Schema.AVRO(User.class))
                .topic("persistent://public/default/test")
                .subscriptionName("sub-2")
                .subscriptionType(SubscriptionType.Shared)
                .subscribe();
while(true){
     Message<User> message = consumer.receive();
     System.out.println("received msg : " + message.getValue());
}

# Produce
bin/pulsar-client produce -n 5 -m "hello" test

If the consumer is using TCP lookup, it will get the below message :

Exception in thread "main" com.google.common.util.concurrent.UncheckedExecutionException: org.apache.commons.lang3.SerializationException: Failed at fetching schema info for EMPTY
	at com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2051)
	at com.google.common.cache.LocalCache.get(LocalCache.java:3951)
	at com.google.common.cache.LocalCache.getOrLoad(LocalCache.java:3974)
	at com.google.common.cache.LocalCache$LocalLoadingCache.get(LocalCache.java:4935)
	at org.apache.pulsar.client.impl.schema.reader.AbstractMultiVersionReader.getSchemaReader(AbstractMultiVersionReader.java:83)

But with HTTP lookup, the error message change to ：

com.google.common.util.concurrent.UncheckedExecutionException: com.google.common.util.concurrent.UncheckedExecutionException: java.nio.BufferUnderflowException
	at com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2055)
	at com.google.common.cache.LocalCache.get(LocalCache.java:3966)
	at com.google.common.cache.LocalCache.getOrLoad(LocalCache.java:3989)
	at com.google.common.cache.LocalCache$LocalLoadingCache.get(LocalCache.java:4950)
	at org.apache.pulsar.client.impl.schema.reader.AbstractMultiVersionReader.getSchemaReader(AbstractMultiVersionReader.java:82)

The root cause is that : if producer produce with bytes data, the schema version is empty. When in TCP lookup, the server will throw unchecked exception here(line-244) :

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java

Lines 236 to 245 in 7c1f17a

	public SchemaVersion versionFromBytes(byte[] version) {
	// The schema storage converts the schema from bytes to long
	// so it handles both cases 1) version is 64 bytes long pre 2.4.0;
	// 2) version is 8 bytes long post 2.4.0
	//
	// NOTE: if you are planning to change the logic here. you should consider
	// both 64 bytes and 8 bytes cases.
	ByteBuffer bb = ByteBuffer.wrap(version);
	return new LongSchemaVersion(bb.getLong());
	}

When in HTTP lookup, the error will throw by the client here(line-164):

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

Lines 161 to 165 in 7c1f17a

	if (version != null) {
	path = String.format("admin/v2/schemas/%s/schema/%s",
	schemaName,
	ByteBuffer.wrap(version).getLong());
	}

So in order to keep the consistent prompt message, we will check the empty schema at the client-side and throw the same exception.

Modifications

• Keep checking the empty version at the client-side.
• Using the same exception.
• Add server-side check at ServerCnx#handleGetSchema to help non-java client to print the same error.

Documentation

• no-need-doc

[nodece]

LGTM

[congbobo184]

LGTM

[gaoran10]

LGTM

[BewareMyPower]

When I debugged testAvroSchemaWithTcpLookup and testAvroSchemaWithHttpLookup, it never went here. Then I removed these lines, tests still passed.