How do you inject the EXISTING object mapper into JsonSerializer and JsonDeserializer

[trajano]

This document talks about creating a NEW (no-arg JsonSerializer) https://docs.spring.io/spring-kafka/docs/current/reference/html/#tip-json but this does not utilize the existing ObjectMapper instance. I'm not saying it doesn't work (it does work), but I am just making sure there's no way of doing it.

#1703 had a bit of discussion but was closed without a working example where it takes the configuration from a @KafkaListener e.g.

  @KafkaListener(
      topics = "${kafka.topic.name}",
      properties = {
        "spring.json.key.default.type: com.example.template.EchoRequest",
        "spring.json.value.default.type: com.example.template.EchoResponse",
      })

And the consumer properties (with the error handling deserializer)

    consumer:
      group-id: template-group
      key-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      properties:
        spring.deserializer.key.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer
        spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer
        spring.json.use.type.headers: false

#680 also had a discussion and was closed with #1053 but that enhanced ObjectMapper is still not the one that was provisioned. It would be preferable that we can simply have it injected somehow rather than it trying to create a new one.

#1782 may have fixed it but I can't find the ObjectMapperCustomizer anymore

[trajano]

Adding these as noted in #680 (comment) does not work, if I looked at the debugger the objectMapper being used for serder is still the enhancedObjectMapper

  @Bean
  @SuppressWarnings({"unchecked", "rawtypes"})
  Serializer<Object> jsonSerializer(ObjectMapper objectMapper) {
    return  new JsonSerializer<>(objectMapper);
  }

  @Bean
  @SuppressWarnings({"unchecked", "rawtypes"})
  Deserializer<Object> jsonDeserializer(ObjectMapper objectMapper) {
    return new JsonDeserializer<>(objectMapper);
  }

[artembilan]

When you add those beans, they have to be used explicitly for the DefaultKafkaProducerFactory:

	/**
	 * Construct a factory with the provided configuration and {@link Serializer}s.
	 * Also configures a {@link #transactionIdPrefix} as a value from the
	 * {@link ProducerConfig#TRANSACTIONAL_ID_CONFIG} if provided.
	 * This config is going to be overridden with a suffix for target {@link Producer} instance.
	 * The serializers' {@code configure()} methods will be called with the
	 * configuration map.
	 * @param configs the configuration.
	 * @param keySerializer the key {@link Serializer}.
	 * @param valueSerializer the value {@link Serializer}.
	 */
	public DefaultKafkaProducerFactory(Map<String, Object> configs,
			@Nullable Serializer<K> keySerializer,
			@Nullable Serializer<V> valueSerializer) {

and DefaultKafkaConsumerFactory, respectively:

	/**
	 * Construct a factory with the provided configuration and deserializers.
	 * The deserializers' {@code configure()} methods will be called with the
	 * configuration map.
	 * @param configs the configuration.
	 * @param keyDeserializer the key {@link Deserializer}.
	 * @param valueDeserializer the value {@link Deserializer}.
	 */
	public DefaultKafkaConsumerFactory(Map<String, Object> configs,
			@Nullable Deserializer<K> keyDeserializer,
			@Nullable Deserializer<V> valueDeserializer) {

If you talk about Spring Boot and its auto-configuration for Spring for Apache Kafka, then you need to provide respective DefaultKafkaProducerFactoryCustomizer and DefaultKafkaConsumerFactoryCustomizer.
From there you call setters for (De)Serializer on those factories:

@Bean 
DefaultKafkaProducerFactoryCustomizer producerCustomizer(Serializer<?> jsonSerializer) {
    return factory -> factory.setValueSerializer(jsonSerializer);
}

and similar for the DefaultKafkaConsumerFactoryCustomizer.

The configuration via properties knows nothing about dependency injection and those classes are instantiated by the Kafka Client via reflection.

[trajano]

You don't have to actually, which is pretty damn nice. Because I have base configuration in application.yml and @KafkaListener #1703 was actually bringing up the issue. Here's a snippet of my application.yml

spring:
  kafka:
    # bootstrap-servers: localhost:9092
    consumer:
      group-id: template-group
      key-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      properties:
        spring.deserializer.key.delegate.class: commons.kafka.ObjectMapperJsonDeserializer
        spring.deserializer.value.delegate.class: commons.kafka.ObjectMapperJsonDeserializer
        spring.json.use.type.headers: false
    producer:
      key-serializer: commons.kafka.ObjectMapperJsonSerializer
      value-serializer: commons.kafka.ObjectMapperJsonSerializer

I have the custom serializer and deserializer and the code is pretty much simple for them

// note not a @Component and is in the same package as the serializer
public class ObjectMapperJsonDeserializer extends JsonDeserializer<Object> {
  public ObjectMapperJsonDeserializer() {
    super(CUSTOM_OBJECT_MAPPER);
  }
}

// note not a @Component
public class ObjectMapperJsonSerializer extends JsonSerializer<Object> {
  static final ObjectMapper CUSTOM_OBJECT_MAPPER;
  static {
    CUSTOM_OBJECT_MAPPER = JacksonUtils.enhancedObjectMapper();
    CUSTOM_OBJECT_MAPPER.registerModule(new ProtobufToJsonModule());
  }
  public ObjectMapperJsonSerializer() {
    super(CUSTOM_OBJECT_MAPPER);
  }
}

It does not solve the problem that we're not getting the already built object mapper but at least we can rebuild it on our own.

[artembilan]

As I said: that reflection configuration style knows nothing about Spring. Therefore existing object mapper cannot be available from just plain properties config.