[schema] Introduce multi version generic record schema

[sijie]

Motivation

Currently AUTO_CONSUME only supports decoding records from latest schema.
All the schema versions are lost. It makes AUTO_CONSUME less useful in some use cases,
such as CDC. Because there is no way for the applications to know which version of schema
that a message is using.

In order to support multi-version schema, we need to propagate schema version from
message header through schema#decode method to the decoded record.

Modifications

• Introduce a new decode method decode(byte[] data, byte[] schemaVersion). This allows the implementation
  to leverage the schema version.
• Introduce a method supportSchemaVersioning to tell which decode methods to use. Because most of the schema
  implementations such as primitive schemas and POJO based schema doesn't make any sense to use schema version.
• Introduce a SchemaProvider which returns a specific schema instance for a given schema version
• Implement a MultiVersionGenericRecordSchema which decode the messages based on schema version. All the records
  decoded by this schema will have schema version and its corresponding schema definitions.

NOTES

This implementation only introduce the mechanism. But it doesn't wire the multi-version schema
with auto_consume schema. There will be a subsequent pull request on implementing a schema provider
that fetches and caches schemas from brokers.

[ivankelly]

To clarify, this is only ever used for AUTO_CONSUME, no? When a client passes a schema like Schema.AVRO(MyPojo.class), it won't actually know the schema version. So the schema version is only used here for the client to request a specific version of the schema from the broker?

[sijie]

@ivankelly that's correct. for pojo and primitive schemas, they will not know the schema version. for any other schema implementations that returns GenericRecord, the schema version will be used for the client to request a specific version of the schema from the broker. Currently the schema implementation returning GenericRecord is AUTO_CONSUME.

[codelipenghui]

already get the specific schema, is it necessary to provide schemaVersion when call decode()?

[sijie]

@codelipenghui good point. I think it is a good practice to pass the schema version down to the specific implementation and let the implementation decide whether to use this schema version or not.