nicoloboschi opened a new pull request, #17959:
URL: https://github.com/apache/pulsar/pull/17959
### 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
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update
later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
