KafkaNull and custom deserializer confusing stacktrace (documentation)

[jorgheymans]

I have a custom deserializer that returns null when the payload is somehow not what we expect it to be. spring-kafka converts the null to KafkaNull and I guess then tries to match the @KafkaListener annotated method with this object. This then prints below confusing stacktrace about not being able to invoke my listener methods:

org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message
Endpoint handler details:
Method [public void MyService.onCommand(my.CCMessage)]
Bean [MyService@2689b752]; nested exception is org.springframework.messaging.handler.annotation.support.MethodArgumentNotValidException: Could not resolve method parameter at index 0 in public void MyService.onCommand(my.CCMessage): 1 error(s): [Error in object 'command': codes []; arguments []; default message [Payload value must not be empty]] , failedMessage=GenericMessage [payload=org.springframework.kafka.support.KafkaNull@70b6d3ac, headers={kafka_offset=353, kafka_consumer=brave.kafka.clients.TracingConsumer@75b04b84, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=null, kafka_receivedPartitionId=4, kafka_receivedTopic=command, kafka_receivedTimestamp=1530011220100}]
	at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:262) ~[spring-kafka-2.1.7.RELEASE.jar:2.1.7.RELEASE]
	at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:80) ~[spring-kafka-2.1.7.RELEASE.jar:2.1.7.RELEASE]
	at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:51) ~[spring-kafka-2.1.7.RELEASE.jar:2.1.7.RELEASE]
	at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter$$FastClassBySpringCGLIB$$cde8c01d.invoke(<generated>) ~[spring-kafka-2.1.7.RELEASE.jar:2.1.7.RELEASE]
	at org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:204) ~[spring-core-5.0.7.RELEASE.jar:5.0.7.RELEASE]
	at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:746) ~[spring-aop-5.0.7.RELEASE.jar:5.0.7.RELEASE]
	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163) ~[spring-aop-5.0.7.RELEASE.jar:5.0.7.RELEASE]
	at org.springframework.cloud.sleuth.instrument.messaging.MessageListenerMethodInterceptor.invoke(TraceMessagingAutoConfiguration.java:256) ~[spring-cloud-sleuth-core-2.0.0.RELEASE.jar:2.0.0.RELEASE]
	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:185) ~[spring-aop-5.0.7.RELEASE.jar:5.0.7.RELEASE]
	at org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:688) ~[spring-aop-5.0.7.RELEASE.jar:5.0.7.RELEASE]
	at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter$$EnhancerBySpringCGLIB$$910b2d50.onMessage(<generated>) ~[spring-kafka-2.1.7.RELEASE.jar:2.1.7.RELEASE]
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:1071) [spring-kafka-2.1.7.RELEASE.jar:2.1.7.RELEASE]
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeWithRecords(KafkaMessageListenerContainer.java:1051) [spring-kafka-2.1.7.RELEASE.jar:2.1.7.RELEASE]
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:998) [spring-kafka-2.1.7.RELEASE.jar:2.1.7.RELEASE]
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:866) [spring-kafka-2.1.7.RELEASE.jar:2.1.7.RELEASE]
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:724) [spring-kafka-2.1.7.RELEASE.jar:2.1.7.RELEASE]
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [na:1.8.0_152]
	at java.util.concurrent.FutureTask.run(FutureTask.java:266) [na:1.8.0_152]
	at java.lang.Thread.run(Thread.java:748) [na:1.8.0_152]
Caused by: org.springframework.messaging.handler.annotation.support.MethodArgumentNotValidException: Could not resolve method parameter at index 0 in public void MyService.onCommand(my.CCMessage): 1 error(s): [Error in object 'command': codes []; arguments []; default message [Payload value must not be empty]] 
	at org.springframework.messaging.handler.annotation.support.PayloadArgumentResolver.resolveArgument(PayloadArgumentResolver.java:121) ~[spring-messaging-5.0.7.RELEASE.jar:5.0.7.RELEASE]
	at org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolverComposite.resolveArgument(HandlerMethodArgumentResolverComposite.java:116) ~[spring-messaging-5.0.7.RELEASE.jar:5.0.7.RELEASE]
	at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.getMethodArgumentValues(InvocableHandlerMethod.java:137) ~[spring-messaging-5.0.7.RELEASE.jar:5.0.7.RELEASE]
	at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:109) ~[spring-messaging-5.0.7.RELEASE.jar:5.0.7.RELEASE]
	at org.springframework.kafka.listener.adapter.HandlerAdapter.invoke(HandlerAdapter.java:48) ~[spring-kafka-2.1.7.RELEASE.jar:2.1.7.RELEASE]
	at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:248) ~[spring-kafka-2.1.7.RELEASE.jar:2.1.7.RELEASE]
	... 18 common frames omitted

[garyrussell]

This is a bug; the argument resolvers need to coerce KafkaNull to null for such situations.

[jorgheymans]

thanks for confirming, FWIW i just found out that when i annotate my listener method payload with @Payload(required = false) the stacktrace disappears.

[garyrussell]

Doh - sorry - yes; that's the proper solution - it's not a bug; it was designed that way:

		if (isEmptyPayload(payload)) {
			if (ann == null || ann.required()) {
				String paramName = getParameterName(parameter);
				BindingResult bindingResult = new BeanPropertyBindingResult(payload, paramName);
				bindingResult.addError(new ObjectError(paramName, "Payload value must not be empty"));
				throw new MethodArgumentNotValidException(message, parameter, bindingResult);
			}
			else {
				return null;
			}
		}

isEmptyPayload() returns true for KafkaNull.

We need to improve the documentation; it talks about it, but the examples actually receive KafkaNull which is incorrect.

[jorgheymans]

😎 Thanks for the clarification