Wrong @body annotation on KafkaBodyBinder

[mfarachepiksel]

Thanks for reporting an issue for Micronaut, please review the task list below before submitting the
issue. Your issue report will be closed if the issue is incomplete and the below tasks not completed.

NOTE: If you are unsure about something and the issue is more of a question a better place to ask questions is on Stack Overflow (http://stackoverflow.com/tags/micronaut) or Gitter (https://gitter.im/micronautfw/). DO NOT use the issue tracker to ask questions.

Task List

• [X ] Steps to reproduce provided
• Stacktrace (if present) provided
• Example that reproduces the problem uploaded to Github
• [X ] Full description of the issue provided (see below)

Steps to Reproduce

I reported in the gitter channel on 7th November, and I got the impresion from the team that I was right on assumption of the possible bug and how to fix it.

I report it here so is not missed

I cannot provide a sample, but can provide some pointers on how to reproduce.

Assume you have a Foo class (Serializable and with empty Contructor)

FooEventListener.java

@KafkaListener
public class FooEventListener {
	
	protected static final Logger LOG = LoggerFactory.getLogger(FooEventListener.class);
	
	@Topic("foo")
	public void consume( @KafkaKey String fooId, @Body FooEvent foo) {
		LOG.debug("KAKFA EVENT RECEIVED AT CUSTOM APPLICATION LISTENER");
	}
}

FooClient

@KafkaClient
public interface FooClient {

    @Topic("foo") 
    void sendFoo(@KafkaKey String fooId, @Body FooEvent Event); 
}

Note: In both example @Body annotation uses

import io.micronaut.messaging.annotation.Body;

Expected Behaviour

The message should be consumed by the KafkaListener class

Actual Behaviour

There is no problem sending Foo object to my kafka , however I see errores while consuming
the record.

18:21:52.173 [pool-1-thread-1] ERROR io.micronaut.configuration.kafka.exceptions.KafkaListenerExceptionHandler - Error processing record [Optional[ConsumerRecord(topic = foo, partition = 0, offset = 0, CreateTime = 1541697712137, serialized key size = 4, serialized value size = 71, headers = RecordHeaders(headers = [], isReadOnly = false), key = 1111, value = micronaut.demo.Foo@115d85d5)]] for Kafka consumer [micronaut.demo.FooEventListener@8dfe921] produced error: Required argument [FooEvent fooEvent] not specified
io.micronaut.core.bind.exceptions.UnsatisfiedArgumentException: Required argument [FooEvent fooEvent] not specified
at io.micronaut.core.bind.DefaultExecutableBinder.bind(DefaultExecutableBinder.java:90)
at io.micronaut.configuration.kafka.processor.KafkaConsumerProcessor.lambda$process$7(KafkaConsumerProcessor.java:418)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
`

Possible reason:

See L21 from KafkaBodyBinder "import io.micronaut.http.annotation.Body;"

micronaut-core/configurations/kafka/src/main/java/io/micronaut/configuration/kafka/bind/KafkaBodyBinder.java

Line 21 in 0caa873

import io.micronaut.http.annotation.Body;

However from the guides, apparently it should be used io.micronaut.messaging.annotation.Body

That would justify why there is no binder associated to the record.

Environment Information

• Operating System: MacOS Mojava
• Micronaut Version: 1.0.0
• JDK Version: 1.8

Example Application

Sorry cannot share as the use case is convoluted.