ParseStringDeserializer throws NullPointerException when message is null

[lagweasel]

In what version(s) of Spring for Apache Kafka are you seeing this issue?

3.0.7

Describe the bug

When deserialize() method is called with a null (bytes) message a NullPointerException is thrown. This is thrown from the line that attempts to create a new String from the bytes:

	@Override
	public T deserialize(String topic, Headers headers, byte[] data) {
		return this.parser.apply(new String(data, this.charset), headers);
	}

To Reproduce

Publish a null value to a kafka topic where the consumer uses ParseStringDeserializer for deserialization (e.g. via ToFromStringSerde).

Expected behavior

Underlying 'parser' function expected to be invoked with null String value. This expectation is supported by the javadoc on Deserializer interface (that ParseStringDeserializer implements).

    /**
     * Deserialize a record value from a byte array into a value or object.
     * @param topic topic associated with the data
     * @param headers headers associated with the record; may be empty.
     * @param data serialized bytes; may be null; implementations are recommended to handle null by returning a value or null rather than throwing an exception.
     * @return deserialized typed data; may be null
     */
    default T deserialize(String topic, Headers headers, byte[] data) {
        return deserialize(topic, data);
    }

[garyrussell]

Thanks; a simple work around is to subclass the deserializer and call the super method if the value is not null.

[lagweasel]

I notice that the fix has:

	public T deserialize(String topic, Headers headers, byte[] data) {
		if (data == null) {
			return null;
		}
		return this.parser.apply(new String(data, this.charset), headers);
	}

But wondering whether it should instead invoke the parser passing a null value instead. (There may be some reason for the parser to handle null and return something different?) e.g.

	public T deserialize(String topic, Headers headers, byte[] data) {
                String stringData = (data == null ? null : new String(data, this.charset));
		return this.parser.apply(stringData, headers);
	}

[artembilan]

... and cherry-picked to 2.9.x