springboot-kafka java 8 time serialization

When you set the serializers/deserializers using properties, Kafka instantiates them, not Spring. Kafka knows nothing about Spring or the customized ObjectMapper.

You need to override Boot's default producer/consumer factories and use the alternate constructors (or setters) to add the serializers/deserializers.

See the documentation.

Important

Only simple configuration can be performed with properties; for more advanced configuration (such as using a custom ObjectMapper in the serializer/deserializer), you should use the producer/consumer factory constructors that accept a pre-built serializer and deserializer. For example, with Spring Boot, to override the default factories:

@Bean
public ConsumerFactory<Foo, Bar> kafkaConsumerFactory(KafkaProperties properties,
    JsonDeserializer customDeserializer) {

    return new DefaultKafkaConsumerFactory<>(properties.buildConsumerProperties(),
        customDeserializer, customDeserializer);
}

@Bean
public ProducererFactory<Foo, Bar> kafkaProducerFactory(KafkaProperties properties,
    JsonSserializer customSerializer) {

    return new DefaultKafkaConsumerFactory<>(properties.buildProducerProperties(),
        customSerializer, customSerializer);
}

Setters are also provided, as an alternative to using these constructors.

Using the Json(De)Serializer constructor with the object mapper param worked for me. I was having trouble (de)serializing a pojo that had an java.time.Instant field, so after hours of troubleshooting this same org.apache.kafka.common.errors.SerializationException***, I finally realized (with the help of answers such as those on here) that the issue is not spring, but kafka's own serialization. Given the objectmapper bean I had, I resolved by autowiring this into the JsonSerializer and JsonDeserializer parameters of my kafka producer and consumer set-ups.

@Configuration
public class JacksonConfig {

    @Bean
    @Primary
    public ObjectMapper objectMapper(Jackson2ObjectMapperBuilder builder) {
        ObjectMapper objectMapper = builder.build();
        objectMapper.registerModule(new JavaTimeModule());
        objectMapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
        return objectMapper;
    }
}

@Configuration
public class KafkaProducerConfig {

    @Value(value="${kafka.bootstrapAddress}")
    private String bootstrapAddress;

    @Autowired
    private ObjectMapper objectMapper;

    @Bean
    public KafkaTemplate<String, Order> orderKafkaTemplate(){
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
        ProducerFactory<String, Order> producerFactory = new DefaultKafkaProducerFactory<>(props, new StringSerializer(), new JsonSerializer<Order>(objectMapper));

        return new KafkaTemplate<>(producerFactory);
    }
}

@Configuration
public class KafkaConsumerConfig {

    @Value(value="${kafka.bootstrapAddress}")
    private String bootstrapAddress;

    @Value(value="${kafka.consumer.groupId}")
    private String groupId;

    @Autowired
    private ObjectMapper objectMapper;

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Order> orderKafkaListenerContainerFactory(){
        ConcurrentKafkaListenerContainerFactory<String, Order> factory = new ConcurrentKafkaListenerContainerFactory<>();

        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        ConsumerFactory<String, Order> consumerFactory = new DefaultKafkaConsumerFactory<>(props, new StringDeserializer(), new JsonDeserializer<>(Order.class, objectMapper));

        factory.setConsumerFactory(consumerFactory);
        return factory;
    }

}

(Pojo shown for further clarity)

public class Order {

    private long accountId;
    private long assetId;
    private long quantity;
    private long price;
    private Instant createdOn = Instant.now();

    // no args constructor, constructor with params for all fields except createdOn, and getters/setters for all fields omitted

***often the cause was: com.fasterxml.jackson.databind.exc.InvalidDefinitionException: Cannot construct instance of 'java.time.Instant' (no Creators, like default construct, exist): cannot deserialize from object value (no delegate- or property-based Creator) at [Source: (byte[])"{"accountId":1,"assetId":2,"quantity":100,"price":1000,"createdOn":{"epochSecond":1558570217,"nano":728000000}}"...