Centralize ObjectMapper construction via JsonUtils.newObjectMapper()

checkstyle/checkstyle.xml

@@ -226,6 +226,17 @@ LinkedIn Java style.
     <property name="message" value="Correct misspelled Javadoc tag"/>
   </module>
 
+  <!-- Force every Jackson mapper through JsonUtils.newObjectMapper() so the codebase has a
+       single source of truth for ObjectMapper configuration (FAIL_ON_UNKNOWN_PROPERTIES
+       disabled by default, with room to grow as the Jackson defaults need to evolve).
+       The factory itself (JsonUtils.java) is the only legitimate `new ObjectMapper(` site;
+       it's suppressed via checkstyle/suppressions.xml. -->
+  <module name="RegexpSingleline">
+    <property name="id" value="RawObjectMapperConstruction"/>
+    <property name="format" value="new\s+ObjectMapper\s*\("/>
+    <property name="message" value="Use JsonUtils.newObjectMapper() instead of constructing a raw ObjectMapper. The factory applies the codebase-wide Jackson defaults (FAIL_ON_UNKNOWN_PROPERTIES disabled) so rolling deploys and schema evolution don't crash older consumers."/>
+  </module>
+
   <!-- Read checker suppressions from a file -->
   <module name="SuppressionFilter">
     <property name="file" value="${config_loc}/suppressions.xml"/>

checkstyle/suppressions.xml

@@ -3,4 +3,9 @@
     "-//Puppy Crawl//DTD Suppressions 1.1//EN"
     "http://www.puppycrawl.com/dtds/suppressions_1_1.dtd">
 <suppressions>
+  <!-- JsonUtils is the canonical factory for ObjectMapper instances; the RawObjectMapperConstruction
+       rule (RegexpSingleline) intentionally exists to send everyone else to JsonUtils.newObjectMapper()
+       instead. -->
+  <suppress checks="RegexpSingleline" id="RawObjectMapperConstruction"
+            files="[\\/]JsonUtils\.java$"/>
 </suppressions>

datastream-common/src/main/java/com/linkedin/datastream/avrogenerator/AvroJson.java

@@ -19,6 +19,8 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.base.CaseFormat;
 
+import com.linkedin.datastream.common.JsonUtils;
+
 
 /**
  * This is a helper class to build all the components of the
@@ -154,7 +156,7 @@ public Map<String, Object> info() {
    */
   public Schema toSchema() throws SchemaGenerationException {
     try {
-      ObjectMapper mapper = new ObjectMapper();
+      ObjectMapper mapper = JsonUtils.newObjectMapper();
       JsonFactory factory = new JsonFactory();
       StringWriter writer = new StringWriter();
       JsonGenerator jgen = factory.createJsonGenerator(writer);

datastream-common/src/main/java/com/linkedin/datastream/common/JsonUtils.java

@@ -34,15 +34,35 @@
 public final class JsonUtils {
   private static final Logger LOG = LoggerFactory.getLogger(JsonUtils.class.getName());
 
-  private static final ObjectMapper MAPPER = new ObjectMapper();
+  private static final ObjectMapper MAPPER = newObjectMapper();
   static {
-    MAPPER.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
-
     MAPPER.addMixIn(Datastream.class, IgnoreDatastreamSetPausedMixIn.class);
     MAPPER.addMixIn(DatastreamSource.class, IgnoreDatastreamSourceSetPartitionsMixIn.class);
     MAPPER.addMixIn(DatastreamDestination.class, IgnoreDatastreamDestinationSetPartitionsMixIn.class);
   }
 
+  private JsonUtils() {
+    // utility class
+  }
+
+  /**
+   * Create a new {@link ObjectMapper} configured with the defaults the Brooklin codebase
+   * applies uniformly: {@link DeserializationFeature#FAIL_ON_UNKNOWN_PROPERTIES} disabled,
+   * so consumers don't crash when newer producers introduce additional JSON fields during
+   * rolling deploys, schema evolution, or cross-version coordination.
+   *
+   * <p>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

datastream-common/src/test/java/com/linkedin/datastream/common/TestJsonUtils.java

@@ -0,0 +1,68 @@
+/**
+ *  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 com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+
+/**
+ * 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;
+    }
+  }
+}

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);
   }
 

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<String, ClusterThroughputInfo> readThroughputInfoFromFile(File file) {
-    ObjectMapper mapper = new ObjectMapper();
+    ObjectMapper mapper = JsonUtils.newObjectMapper();
     HashMap<String, ClusterThroughputInfo> clusterInfoMap = new HashMap<>();
 
     try {
@@ -102,7 +103,7 @@ private HashMap<String, ClusterThroughputInfo> readThroughputInfoFromFile(File f
   }
 
   private ClusterThroughputInfo readThroughputInfoFromFile(File file, String clusterName) {
-    ObjectMapper mapper = new ObjectMapper();
+    ObjectMapper mapper = JsonUtils.newObjectMapper();
     ClusterThroughputInfo clusterInfo = null;
 
     try {

datastream-tools/src/main/java/com/linkedin/datastream/tools/DatastreamRestClientCli.java

@@ -57,7 +57,7 @@ private enum Operation {
   }
 
   private static void printDatastreams(boolean noformat, List<Datastream> streams) {
-    ObjectMapper mapper = new ObjectMapper();
+    ObjectMapper mapper = JsonUtils.newObjectMapper();
 
     streams.forEach(s -> {
       try {