[FLINK-8544] JSONKeyValueDeserializationSchema throws NPE when message key is null

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.4.0
• Fix Version/s: 1.6.0
• Component/s: Connectors / Kafka
• Labels: pull-request-available
• Assignee: Unassigned
• Reporter: nicolasyang
• Created: 01/Feb/18 13:37
• Updated: 15/Jul/18 18:46
• Resolved: 15/Jul/18 18:46

Description

JSONKeyValueDeserializationSchema call Jaskon to deserialize the message key without validation.
If a message with key == null is read, flink throws an NPE.

	@Override
	public ObjectNode deserialize(byte[] messageKey, byte[] message, String topic, int partition, long offset) throws IOException {
		if (mapper == null) {
			mapper = new ObjectMapper();
		}
		ObjectNode node = mapper.createObjectNode();
		node.set("key", mapper.readValue(messageKey, JsonNode.class)); // messageKey is not validate against null.
		node.set("value", mapper.readValue(message, JsonNode.class));

The fix is very straightforward.

		if (messageKey == null) {
			node.set("key", null)
		} else {
			node.set("key", mapper.readValue(messageKey, JsonNode.class));
		}

If it is appreciated, I would send a pull request.

Issue Links

• links to: GitHub Pull Request #5516