Add a Message.getSchema() in the client API

[vroyer]

When the schema is not known at compile time, a Consumer<GenericObject> should be able to get the message schema.

Currently, a Consumer<GenericRecord> can get the AVRO message and retreive its schema, but if the message is a KeyValue for example, there is no way to get the actual Schema.

The idea is to introduce a CompletableFuture<Schema<?>> getSchema() in the Message API, returning null by default.

The implementation classes (MessageImpl, TopicMessageImpl) could then fetch the Schema if available, and eventually keep it cached as a class member.

[eolivelli]

@congbobo184 here https://github.com/apache/pulsar/pull/9970/files#diff-955419b4b0ad976e96f9f7595989e79c391109aeaa304bd286a80fc6eb9360c7R324 you are adding MessageImpl#getSchema() with CompletableFuture

but why can't we use directly the schema instance that is cached inside the Message ?

[eolivelli]

Related work:
Here in this existing test we have a Consumer with AUTO_CONSUME schema
the tests tests that it is possible to access the "current" AvroSchema using Schema.getNativeSchema()

pulsar/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleSchemaTest.java

Line 530 in 7c0a798

org.apache.avro.Schema avroSchema = (org.apache.avro.Schema) impl.getSchema().getNativeSchema().get();

[eolivelli]

The problem with AUTO_CONSUME and especially while dealing with multi-version schemas, is that Message.getSchema() should return the exact schema applied to the message, that is the schema identified by "topic" + "schemaVersion".

[eolivelli]

As discussed at the community meeting I am going to send a PIP

[eolivelli]

This is the PIP
https://docs.google.com/document/d/1VWi5LHP44V31nP4bCui9d5RXwH6xc_phrUes6tvNguk

[codelipenghui]

#10476 is merged, so close this issue.