Add MediaMessage annotation support.

[scruel]

Supports the ability to define the content type of Message from the client side for resolving.

Usage example:

@Configuration
public class MediaMessageConfig implements KafkaListenerConfigurer {
  public CompositeMessageConverter createMessageConverter() {
    List<MessageConverter> converters = new ArrayList<>();
    converters.add(new StringMessageConverter());
    converters.add(new ByteArrayMessageConverter());
    converters.add(createJacksonConverter());
    return new CompositeMessageConverter(converters);
  }

  protected MappingJackson2MessageConverter createJacksonConverter() {
    MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
    converter.setContentTypeResolver(new DefaultContentTypeResolver());
    return converter;
  }

  @Override
  public void configureKafkaListeners(@JsonDeserialize KafkaListenerEndpointRegistrar registrar) {
    registrar.setCustomMethodArgumentResolvers(new MediaMessageResolver(createMessageConverter()));
  }
}


public class ClientConsumer {

@KafkaListener(topics = <topics>, groupId = <groupId>)
  public void onMessage(@MediaMessage BusinessDTO info) {
    ...
  }
}

[garyrussell]

This looks interesting, but @MediaMessage doesn't seem to be correct. Perhaps something like @ResolvableType?

Also, this needs test coverage.

[scruel]

This looks interesting, but @MediaMessage doesn't seem to be correct. Perhaps something like @ResolvableType?

Also, this needs test coverage.

Done, for the name part, maybe @MeidaResolvableType would be better?

[garyrussell]

Thanks; it needs some docs; probably in this section https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/asciidoc/kafka.adoc#kafkalistener-annotation

Also, maybe a "how to use" here https://docs.spring.io/spring-kafka/docs/current/reference/html/#tips-n-tricks

[garyrussell]

@artembilan Any thoughts on the annotation name?

[artembilan]

There is already a org.springframework.core.ResolvableType.
So, this one is going to confuse.
Plus it doesn't look like we really resolve a type, but talk more about a content-type.

I'm not fully sure that I follow with the goal of this feature, but probably better to have it as an @ExpectedContentType since there are many ContentType classes around 😸 .
I would say this is similar to consumes on the @RequestMapping, but you really have here a fallback to the expected if no one provided in the record.
Anyway I'd like to understand better what is going on.
Looks like you want to have a common composite converter and choose an appropriate one according the content-type header or fallback.
How about have an attribute on the @KafkaListener instead? Kinda fallbackContentType?
Populate it to the message (if needed) and still rely on that composite ability to choose an expected converter.

Just thinking aloud...

Thanks

[garyrussell]

Thanks, Artem, for your perspective; I agree coercing the content type in the listener adapter via a @KafkaListener property would be better than polluting the method signature.

[scruel]

Thanks, Artem, for your perspective; I agree coercing the content type in the listener adapter via a @KafkaListener property would be better than polluting the method signature.

In spring, annotation on parameters is very common, I think it can not treat as polluting the method signature, for example @RequestBody.

[artembilan]

Yes, but we already have enough param annotations to pollute enough.
And your idea makes me think that this kind of fallbackContentType should go to the existing @Payload annotation which we rely in Spring Kafka as well.
But again: your current goal could be achieved with an explicit converter for the particular @KakfaListener. The rest I believe is already covered by the composition if contentType header is present.

[scruel]

Hah, @Payload annotation is in org.springframework.messaging package but not this project, I treat the @MediaMessage as an enhanced annotation for spring-kafka.
I know it was covered if contentType header is present, but sometimes, you can not control the producers for which message they will send.

[scruel]

Thanks; it needs some docs; probably in this section https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka-docs/src/main/asciidoc/kafka.adoc#kafkalistener-annotation

Also, maybe a "how to use" here https://docs.spring.io/spring-kafka/docs/current/reference/html/#tips-n-tricks

I will fulfill the docs after you make your decision.

[scruel]

Thanks, Artem, for your perspective; I agree coercing the content type in the listener adapter via a @KafkaListener property would be better than polluting the method signature.

Would you need me also to implement this idea?

[artembilan]

Well, and why we should implement such a feature only in Spring for Apache Kafka?
Why other projects cannot benefit from the feature? For example Spring AMQP, Spring Cloud AWS. Spring Integration at all, at last. And so on. The same @MessageMapping for WebScokets and RSockets.
That's why I'm talking about that @Payload as a general place for such a fallback option.
But again: I believe even fallback option can be implemented right now with just a composition with desired converter in the end of chain.
I even think something like this is resent in Spring Cloud Stream with its JSON converter as a fallback.

I really against new annotations because end-user needs to know what to look for. With just a new attribute on the well-known @KafkaListener he/she gets a clue what is possible.
However independently of the end-user hook for the method, they still need to support an infrastructure for converter composition and so on.

I guess we need to think more what and how could be done and where.
Right now I'm not fully convinced in a new annotation, but I'm opened for arguments anyway.

Thanks for understanding

[scruel]

Thanks, I agree that the other projects have the rights to benefit from this.

they still need to support an infrastructure for converter composition and so on.

I do want to add this in default configuration, but maybe we should consider how to implement this first, then consider should or not should to add this support feature as a default.
I will wait for your decision, feel free to comment your ideas.