Support KafkaNull in payload conversion

See https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/pull/539

We need to map payloads of type `KafkaNull` to `null` when mapping method
arguments (`@Payload(required = false) Foo foo`.

Resolves #1602

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/SmartPayloadArgumentResolver.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2018 the original author or authors.
+ * Copyright 2016-2019 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -37,7 +37,9 @@ import org.springframework.validation.Validator;
 
 /**
  * @author Oleg Zhurakousky
+ * @author Gary Russell
  * @deprecated will be removed once https://jira.spring.io/browse/SPR-17503 is addressed
+ * (but see note about KafkaNull below).
  */
 @Deprecated
 class SmartPayloadArgumentResolver extends PayloadArgumentResolver {
@@ -68,6 +70,16 @@ class SmartPayloadArgumentResolver extends PayloadArgumentResolver {
 				&& !parameter.hasParameterAnnotation(Headers.class));
 	}
 
+	/**
+	 * Needed to support mapping KafkaNull to null in method invocation.
+	 */
+	@Override
+	protected boolean isEmptyPayload(Object payload) {
+		return super.isEmptyPayload(payload)
+				|| "org.springframework.kafka.support.KafkaNull"
+						.equals(payload.getClass().getName());
+	}
+
 	@Override
 	@Nullable
 	public Object resolveArgument(MethodParameter parameter, Message<?> message)