Use this factory instead of {@code new ObjectMapper()} for any mapper construction
+ * outside this class. Centralizing the call lets every Jackson consumer pick up future
+ * default tweaks (new modules, new feature flags) without each caller having to
+ * re-discover the right config.
+ *
+ * @return a fresh, base-configured {@link ObjectMapper}
+ */
+ public static ObjectMapper newObjectMapper() {
+ return new ObjectMapper()
+ .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+ }
+
/**
* Deserialize a JSON string into an object with the specified type.
* @param json JSON string
diff --git a/datastream-common/src/test/java/com/linkedin/datastream/common/TestJsonUtils.java b/datastream-common/src/test/java/com/linkedin/datastream/common/TestJsonUtils.java
new file mode 100644
index 000000000..348643d33
--- /dev/null
+++ b/datastream-common/src/test/java/com/linkedin/datastream/common/TestJsonUtils.java
@@ -0,0 +1,69 @@
+/**
+ * Copyright 2026 LinkedIn Corporation. All rights reserved.
+ * Licensed under the BSD 2-Clause License. See the LICENSE file in the project root for license information.
+ * See the NOTICE file in the project root for additional information regarding copyright ownership.
+ */
+package com.linkedin.datastream.common;
+
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+
+/**
+ * Tests for {@link JsonUtils}.
+ */
+public class TestJsonUtils {
+
+ @Test
+ public void testNewObjectMapperReturnsNonNull() {
+ ObjectMapper mapper = JsonUtils.newObjectMapper();
+ Assert.assertNotNull(mapper);
+ }
+
+ @Test
+ public void testNewObjectMapperDisablesFailOnUnknownProperties() {
+ ObjectMapper mapper = JsonUtils.newObjectMapper();
+ Assert.assertFalse(mapper.isEnabled(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES),
+ "JsonUtils.newObjectMapper() must disable FAIL_ON_UNKNOWN_PROPERTIES so consumers don't crash on "
+ + "newer payloads during rolling deploys / schema evolution.");
+ }
+
+ @Test
+ public void testNewObjectMapperToleratesUnknownProperties() throws Exception {
+ ObjectMapper mapper = JsonUtils.newObjectMapper();
+
+ // A producer adds a future field that this consumer's class doesn't know about; deserialization
+ // must succeed rather than fail-loud.
+ String json = "{\"name\":\"alpha\",\"futureField\":\"ignored\"}";
+ SimpleBean bean = mapper.readValue(json, SimpleBean.class);
+ Assert.assertEquals(bean.getName(), "alpha");
+ }
+
+ @Test
+ public void testEveryNewObjectMapperCallReturnsAFreshInstance() {
+ ObjectMapper a = JsonUtils.newObjectMapper();
+ ObjectMapper b = JsonUtils.newObjectMapper();
+ Assert.assertNotSame(a, b,
+ "JsonUtils.newObjectMapper() is a factory; callers should be free to mutate (e.g. registerModule) "
+ + "the returned mapper without affecting siblings.");
+ }
+
+ /**
+ * Plain-old-Java bean used to test unknown-property tolerance. Only declares {@code name};
+ * any extra field in the JSON payload should be silently ignored.
+ */
+ public static final class SimpleBean {
+ private String _name;
+
+ public String getName() {
+ return _name;
+ }
+
+ public void setName(String name) {
+ _name = name;
+ }
+ }
+}
diff --git a/datastream-kafka-connector/src/main/java/com/linkedin/datastream/connectors/kafka/KafkaDatastreamStatesResponse.java b/datastream-kafka-connector/src/main/java/com/linkedin/datastream/connectors/kafka/KafkaDatastreamStatesResponse.java
index 2890d2294..7883288f3 100644
--- a/datastream-kafka-connector/src/main/java/com/linkedin/datastream/connectors/kafka/KafkaDatastreamStatesResponse.java
+++ b/datastream-kafka-connector/src/main/java/com/linkedin/datastream/connectors/kafka/KafkaDatastreamStatesResponse.java
@@ -16,7 +16,6 @@
import com.fasterxml.jackson.core.JsonParser;
import com.fasterxml.jackson.core.Version;
import com.fasterxml.jackson.databind.DeserializationContext;
-import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.JsonDeserializer;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.KeyDeserializer;
@@ -106,9 +105,9 @@ public static KafkaDatastreamStatesResponse fromJson(String json) {
simpleModule.addKeyDeserializer(FlushlessEventProducerHandler.SourcePartition.class, SourcePartitionDeserializer.getInstance());
simpleModule.addKeyDeserializer(TopicPartition.class, TopicPartitionKeyDeserializer.getInstance());
simpleModule.addDeserializer(TopicPartition.class, TopicPartitionDeserializer.getInstance());
- ObjectMapper mapper = new ObjectMapper();
+ // JsonUtils.newObjectMapper() already disables FAIL_ON_UNKNOWN_PROPERTIES.
+ ObjectMapper mapper = JsonUtils.newObjectMapper();
mapper.registerModule(simpleModule);
- mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
return JsonUtils.fromJson(json, KafkaDatastreamStatesResponse.class, mapper);
}
diff --git a/datastream-server/src/main/java/com/linkedin/datastream/server/providers/FileBasedPartitionThroughputProvider.java b/datastream-server/src/main/java/com/linkedin/datastream/server/providers/FileBasedPartitionThroughputProvider.java
index 540abcbc4..f4f77dcb6 100644
--- a/datastream-server/src/main/java/com/linkedin/datastream/server/providers/FileBasedPartitionThroughputProvider.java
+++ b/datastream-server/src/main/java/com/linkedin/datastream/server/providers/FileBasedPartitionThroughputProvider.java
@@ -19,6 +19,7 @@
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.linkedin.datastream.common.JsonUtils;
import com.linkedin.datastream.server.ClusterThroughputInfo;
import com.linkedin.datastream.server.DatastreamGroup;
import com.linkedin.datastream.server.PartitionThroughputInfo;
@@ -82,7 +83,7 @@ private File getThroughputFileFromResources() {
}
private HashMap