[fix][connector] Kinesis sink: fix NPE with KeyValue schema and no value

[nicoloboschi]

Motivation

In the Kinesis sink with FULL_MESSAGE_IN_JSON_EXPAND_VALUE, a message with KeyValue schema and with value=NULL you get this NPE

2022-10-07T07:27:33,698+0000 [public/default/kinesis-sink-0] INFO  org.apache.pulsar.functions.instance.JavaInstanceRunnable - Encountered exception in sink write: 
java.lang.NullPointerException: null
	at org.apache.pulsar.io.kinesis.Utils.toJsonSerializable(Utils.java:250) ~[HsqgvoaXuefCgCSXYBQtdA/:?]
	at org.apache.pulsar.io.kinesis.Utils.serializeRecordToJsonExpandingValue(Utils.java:222) ~[HsqgvoaXuefCgCSXYBQtdA/:?]
	at org.apache.pulsar.io.kinesis.KinesisSink.createKinesisMessage(KinesisSink.java:308) ~[HsqgvoaXuefCgCSXYBQtdA/:?]
	at org.apache.pulsar.io.kinesis.KinesisSink.write(KinesisSink.java:135) ~[HsqgvoaXuefCgCSXYBQtdA/:?]
	at org.apache.pulsar.functions.instance.JavaInstanceRunnable.sendOutputMessage(JavaInstanceRunnable.java:424) ~[?:?]
	at org.apache.pulsar.functions.instance.JavaInstanceRunnable.handleResult(JavaInstanceRunnable.java:394) ~[?:?]
	at org.apache.pulsar.functions.instance.JavaInstanceRunnable.run(JavaInstanceRunnable.java:335) ~[?:?]
	at java.lang.Thread.run(Thread.java:829) ~[?:?]
2022-10-07T07:27:33,712+0000 [public/default/kinesis-sink-0] ERROR org.apache.pulsar.functions.instance.JavaInstanceRunnable - [public/default/kinesis-sink:0] Uncaught exception in Java Instance
java.lang.RuntimeException: Failed to process message: 31:4:-1:0
	at org.apache.pulsar.functions.source.PulsarSource.lambda$buildRecord$1(PulsarSource.java:152) ~[com.datastax.oss-pulsar-functions-instance-2.10.2.1.jar:2.10.2.1]
	at org.apache.pulsar.functions.source.PulsarRecord.fail(PulsarRecord.java:117) ~[com.datastax.oss-pulsar-functions-instance-2.10.2.1.jar:2.10.2.1]
	at org.apache.pulsar.functions.instance.JavaInstanceRunnable.sendOutputMessage(JavaInstanceRunnable.java:429) ~[com.datastax.oss-pulsar-functions-instance-2.10.2.1.jar:2.10.2.1]
	at org.apache.pulsar.functions.instance.JavaInstanceRunnable.handleResult(JavaInstanceRunnable.java:394) ~[com.datastax.oss-pulsar-functions-instance-2.10.2.1.jar:2.10.2.1]
	at org.apache.pulsar.functions.instance.JavaInstanceRunnable.run(JavaInstanceRunnable.java:335) ~[com.datastax.oss-pulsar-functions-instance-2.10.2.1.jar:2.10.2.1]
	at java.lang.Thread.run(Thread.java:829) ~[?:?]

Modifications

• Added NPE check for the KeyValue schema when serializing to JSON. If the key/value is null, the "key"/"value" entry is omitted from the final payload

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

[eolivelli]

LGTM

[tisonkun]

/pulsarbot run-failure-checks