Description
The JsonConverter.TO_CONNECT_LOGICAL_CONVERTERS map contains LogicalTypeConverter implementations to convert from the raw value into the corresponding logical type value, and they are used during deserialization of message keys and/or values. However, these implementations do not handle the case when the input raw value is null, which can happen when a key or value has a schema that is or contains a field that is optional.
Consider a Kafka Connect schema of type STRUCT that contains a field "date" with an optional schema of type org.apache.kafka.connect.data.Date. When the key or value with this schema contains a null "date" field and is serialized, the logical serializer properly will serialize the null value. However, upon deserialization, the JsonConverter.TO_CONNECT_LOGICAL_CONVERTERS are used to convert the literal value (which is null) to a logical value. All of the JsonConverter.TO_CONNECT_LOGICAL_CONVERTERS implementations will throw a NullPointerException when the input value is null.
For example:
java.lang.NullPointerException at org.apache.kafka.connect.json.JsonConverter$14.convert(JsonConverter.java:224) at org.apache.kafka.connect.json.JsonConverter.convertToConnect(JsonConverter.java:731) at org.apache.kafka.connect.json.JsonConverter.access$100(JsonConverter.java:53) at org.apache.kafka.connect.json.JsonConverter$12.convert(JsonConverter.java:200) at org.apache.kafka.connect.json.JsonConverter.convertToConnect(JsonConverter.java:727) at org.apache.kafka.connect.json.JsonConverter.jsonToConnect(JsonConverter.java:354) at org.apache.kafka.connect.json.JsonConverter.toConnectData(JsonConverter.java:343)