ObjectMapper initialized inside JsonDeserializer ignores default configuration

[LudovitVarga]

I am using JsonDeserializer provided by spring-kafka (version 2.2.2.RELEASE) with default ObjectMapper. I find out that configuration made to this ObjectMapper inside JsonDeserializer class is ignored

148:   this.objectMapper.configure(MapperFeature.DEFAULT_VIEW_INCLUSION, false);
149:   this.objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);

For replication i created small application with single JUnit test (attached at the end), using Spring Boot. If you run test with default JsonDeserializer, result is failure with message:

com.fasterxml.jackson.databind.exc.UnrecognizedPropertyException: Unrecognized field "unknownField" (class com.example.MyEvent), not marked as ignorable (one known property: "myField"])

But if i switch to JsonDeserializer with provided ObjectMapper (with same configuration as is done inside JsonDeserializer) then message is properly deserialized and displayed.

MyTests.java

package com.example;

import java.util.Map;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.serializer.JsonDeserializer;
import org.springframework.kafka.test.EmbeddedKafkaBroker;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit4.SpringRunner;

import com.example.MyApplication;
import com.example.MyEvent;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.MapperFeature;
import com.fasterxml.jackson.databind.ObjectMapper;

@RunWith(SpringRunner.class)
@SpringBootTest(classes = MyApplication.class, properties = {
		"spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
		"logging.level.root=WARN" })
@DirtiesContext
@EmbeddedKafka(partitions = 1, topics = { "test-topic" })
public class MyTests {

	private static final String CONSUMER_GROUP_ID = "testConsumer";

	@Autowired
	private EmbeddedKafkaBroker embeddedKafka;

	@Autowired
	private KafkaTemplate<Object, String> kafkaTemplate;

	@Test
	public void test() throws Exception {
		this.kafkaTemplate.send("test-topic", "{ \"myField\" : \"lorem\", \"unknownField\" : \"impsum\"}");
		this.kafkaTemplate.flush();

		Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(CONSUMER_GROUP_ID, "true", this.embeddedKafka);
		consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

		ConsumerFactory<String, MyEvent> cf = new DefaultKafkaConsumerFactory<>(consumerProps, new StringDeserializer(), defaultJsonDeserializer());
		Consumer<String, MyEvent> consumer = cf.createConsumer();
		this.embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "test-topic");
		ConsumerRecords<String, MyEvent> records = KafkaTestUtils.getRecords(consumer, 5000);

		records.forEach(record -> System.out.println(record.value()));
	}

	private JsonDeserializer<MyEvent> defaultJsonDeserializer() {
		 return new JsonDeserializer<>(MyEvent.class);
	}

	private JsonDeserializer<MyEvent> customizedJsonDeserializer() {
		ObjectMapper objectMapper = new ObjectMapper();
		objectMapper.configure(MapperFeature.DEFAULT_VIEW_INCLUSION, false);
		objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
		return new JsonDeserializer<>(MyEvent.class, objectMapper);
	}

}

MyApplication.java

package com.example;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.kafka.annotation.EnableKafka;

@SpringBootApplication
@EnableKafka
public class MyApplication {

	public static void main(String[] args) {
		SpringApplication.run(MyApplication.class, args);
	}

}

MyEvent.class

package com.example;

public class MyEvent {

	private String myField;

	public String getMyField() {
		return myField;
	}

	public void setMyField(String myField) {
		this.myField = myField;
	}

	@Override
	public String toString() {
		return "MyEvent [myField=" + myField + "]";
	}

}

pom.xml

<project xmlns="http://maven.apache.org/POM/4.0.0"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>

	<groupId>com.example</groupId>
	<artifactId>spring-kafka-problem</artifactId>
	<version>0.0.1-SNAPSHOT</version>
	<packaging>jar</packaging>

	<parent>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-parent</artifactId>
		<version>2.1.1.RELEASE</version>
		<relativePath /> <!-- lookup parent from repository -->
	</parent>

	<properties>
		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
		<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
		<java.version>8</java.version>
	</properties>

	<dependencies>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springframework.kafka</groupId>
			<artifactId>spring-kafka</artifactId>
		</dependency>
		<dependency>
			<groupId>com.fasterxml.jackson.core</groupId>
			<artifactId>jackson-databind</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-test</artifactId>
			<scope>test</scope>
		</dependency>
		<dependency>
			<groupId>org.springframework.kafka</groupId>
			<artifactId>spring-kafka-test</artifactId>
			<scope>test</scope>
		</dependency>
	</dependencies>

	<build>
		<plugins>
			<plugin>
				<groupId>org.springframework.boot</groupId>
				<artifactId>spring-boot-maven-plugin</artifactId>
			</plugin>
		</plugins>
	</build>

</project>

[garyrussell]

Confirmed; the problem is that, using that constructor, the reader is created before the options are set.