Pulsar IO: Allow to develop Sinks that support Schema but without setting it at build time (Sink<GenericObject>)

[eolivelli]

Motivation

Support users that want to implement a Sink that is aware of the Schema but it is not tied to a specific schema type at compile time.
With this patch you can implement a Sink that is aware of the Schema but it is not tied to a particular schema type at compile time.

public class ObjectSink implements Sink<GenericObject> {

    @Override
    public void write(Record<GenericObject> record) throws Exception {
        try {
            Schema<?> schema = record.getSchema();
            SchemaType type = record.getValue().getType();
            Object value = record.getValue().getNativeObject();
            log.info("Record {} type {} schema {} value {}", record, type, schema, value);
            record.ack();
        } catch (Throwable t) {
            .....
        }
    }

Modifications

We are adding support in TopicSchema.java to deal with GenericObject, the same way we did with GenericRecord. GenericRecord is a subtype of GenericObject, and AutoConsumeSchema now returns GenericObject instances that wrap primitive data types

Verifying this change

I will add test cases and integration tests as soon as the community accepts this feature.

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

With this feature Sink developers will be able to write advanced Sinks

Documentation

• Does this pull request introduce a new feature? yes
• If yes, how is the feature documented? docs

I will be happy to provide documentation as soon as this feature lands to Pulsar

[eolivelli]

More references here
https://docs.google.com/document/d/117gr6hG3N46MlkqENRKRtLejb_e_K0cZagQZv3jbwvQ/edit?ts=605a14f6#heading=h.gm9efz639pj1

[eolivelli]

@codelipenghui I have reverted the conf file.

[eolivelli]

@codelipenghui @jerrypeng CI finally passed