Pulsar schema doesn't support null value

[sijie]

Describe the bug

If a Pulsar producer sends messages without payload, Pulsar consumer is not able to consume the messages without payload. Because the payload passed to schema is an empty array not null. Hence exceptions are thrown. An example exception is shown as below:

org.apache.pulsar.client.api.SchemaSerializationException: Size of data received by BooleanSchema is not 1

	at org.apache.pulsar.client.impl.schema.BooleanSchema.validate(BooleanSchema.java:44)
	at org.apache.pulsar.client.impl.schema.BooleanSchema.decode(BooleanSchema.java:62)
	at org.apache.pulsar.client.impl.schema.BooleanSchema.decode(BooleanSchema.java:29)
	at org.apache.pulsar.client.impl.MessageImpl.getValue(MessageImpl.java:273)
	at org.apache.pulsar.client.impl.TopicMessageImpl.getValue(TopicMessageImpl.java:143)

To Reproduce

1. Use a typed producer to send messages without setting value.
2. Use a typed consumer to receive messages.

Expected behavior

The typed consumer should receive messages with a null value.

[sijie]

In event-sourcing use cases, a null value is usually used for representing a delete operation. Hence we should return null if the message doesn't have any payload.

[jiazhai]

move this into 2.5.0, since PR has already moved.

[fmiguelez]

There is a workaround with direct consumer/reader interface since you can always verify if data is empty before calling getValue():
Message<GenericRecord> lastMsg = reader.readNext(); lastEntity = lastMsg.getData().length > 0 ? lastMsg.getValue() : null;

However there is not such option when working with functions. E.g. A Sink<GenericRecord>implementation will fail before calling write(Record<GenericRecord> record) if a message with no payload (valid tombstone) is received with this error:

10:02:27.911 [dbus/test/rtsink-0] ERROR org.apache.pulsar.functions.instance.JavaInstanceRunnable - [dbus/test/rtsink:0] Uncaught exception in Java Instance
org.apache.pulsar.client.api.SchemaSerializationException: java.io.EOFException
at org.apache.pulsar.client.impl.schema.generic.GenericAvroReader.read(GenericAvroReader.java:80) ~[org.apache.pulsar-pulsar-client-original-2.5.1.jar:2.5.1]
at org.apache.pulsar.client.impl.schema.generic.GenericAvroReader.read(GenericAvroReader.java:42) ~[org.apache.pulsar-pulsar-client-original-2.5.1.jar:2.5.1]
at org.apache.pulsar.client.api.schema.SchemaReader.read(SchemaReader.java:36) ~[java-instance.jar:?]
at org.apache.pulsar.client.impl.schema.StructSchema.decode(StructSchema.java:100) ~[org.apache.pulsar-pulsar-client-original-2.5.1.jar:2.5.1]
at org.apache.pulsar.client.impl.schema.AutoConsumeSchema.decode(AutoConsumeSchema.java:96) ~[org.apache.pulsar-pulsar-client-original-2.5.1.jar:2.5.1]
at org.apache.pulsar.client.impl.schema.AutoConsumeSchema.decode(AutoConsumeSchema.java:39) ~[org.apache.pulsar-pulsar-client-original-2.5.1.jar:2.5.1]
at org.apache.pulsar.client.impl.MessageImpl.getValue(MessageImpl.java:276) ~[org.apache.pulsar-pulsar-client-original-2.5.1.jar:2.5.1]
at org.apache.pulsar.client.impl.TopicMessageImpl.getValue(TopicMessageImpl.java:143) ~[org.apache.pulsar-pulsar-client-original-2.5.1.jar:2.5.1]
at org.apache.pulsar.functions.source.PulsarRecord.getValue(PulsarRecord.java:74) ~[org.apache.pulsar-pulsar-functions-instance-2.5.1.jar:2.5.1]
at org.apache.pulsar.functions.instance.JavaInstanceRunnable.readInput(JavaInstanceRunnable.java:472) ~[org.apache.pulsar-pulsar-functions-instance-2.5.1.jar:?]
at org.apache.pulsar.functions.instance.JavaInstanceRunnable.run(JavaInstanceRunnable.java:246) [org.apache.pulsar-pulsar-functions-instance-2.5.1.jar:?]
at java.lang.Thread.run(Thread.java:748) [?:1.8.0_242]
Caused by: java.io.EOFException
at org.apache.avro.io.BinaryDecoder.ensureBounds(BinaryDecoder.java:509) ~[org.apache.avro-avro-1.9.1.jar:1.9.1]
at org.apache.avro.io.BinaryDecoder.readLong(BinaryDecoder.java:181) ~[org.apache.avro-avro-1.9.1.jar:1.9.1]
at org.apache.avro.io.BinaryDecoder.readString(BinaryDecoder.java:279) ~[org.apache.avro-avro-1.9.1.jar:1.9.1]
at org.apache.avro.io.ResolvingDecoder.readString(ResolvingDecoder.java:209) ~[org.apache.avro-avro-1.9.1.jar:1.9.1]
at org.apache.avro.generic.GenericDatumReader.readString(GenericDatumReader.java:467) ~[org.apache.avro-avro-1.9.1.jar:1.9.1]
at org.apache.avro.generic.GenericDatumReader.readString(GenericDatumReader.java:458) ~[org.apache.avro-avro-1.9.1.jar:1.9.1]
at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:182) ~[org.apache.avro-avro-1.9.1.jar:1.9.1]
at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:151) ~[org.apache.avro-avro-1.9.1.jar:1.9.1]
at org.apache.avro.generic.GenericDatumReader.readField(GenericDatumReader.java:248) ~[org.apache.avro-avro-1.9.1.jar:1.9.1]
at org.apache.avro.generic.GenericDatumReader.readRecord(GenericDatumReader.java:237) ~[org.apache.avro-avro-1.9.1.jar:1.9.1]
at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:170) ~[org.apache.avro-avro-1.9.1.jar:1.9.1]
at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:151) ~[org.apache.avro-avro-1.9.1.jar:1.9.1]
at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:144) ~[org.apache.avro-avro-1.9.1.jar:1.9.1]
at org.apache.pulsar.client.impl.schema.generic.GenericAvroReader.read(GenericAvroReader.java:75) ~[org.apache.pulsar-pulsar-client-original-2.5.1.jar:2.5.1]
... 11 more
10:02:27.943 [dbus/test/rtsink-0] INFO org.apache.pulsar.functions.instance.JavaInstanceRunnable - Closing instance
10:02:27.964 [dbus/test/rtsink-0] INFO org.apache.pulsar.client.impl.ConsumerImpl - [persistent://dbus/test/dummy-objects] [dbus/test/rtsink] Closed consumer