fix race condition

Author: derrickaw

sdks/java/core/src/main/java/org/apache/beam/sdk/util/RowJsonUtils.java

@@ -17,6 +17,7 @@
  */
 package org.apache.beam.sdk.util;
 
+import com.fasterxml.jackson.core.JsonFactory;
 import com.fasterxml.jackson.core.JsonParseException;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.JsonMappingException;
@@ -45,32 +46,41 @@ public class RowJsonUtils {
    * overwrite the default buffer size limit to 100 MB, and exposes this interface for higher limit.
    * If needed, call this method during pipeline run time, e.g. in DoFn.setup.
    */
-  public static void increaseDefaultStreamReadConstraints(int newLimit) {
-    if (newLimit <= defaultBufferLimit) {
-      return;
-    }
-    try {
-      Class<?> unused = Class.forName("com.fasterxml.jackson.core.StreamReadConstraints");
+  // The maximum string length for the JSON parser, set to 100 MB.
+  private static final int MAX_STRING_LENGTH = 100 * 1024 * 1024;
 
-      com.fasterxml.jackson.core.StreamReadConstraints.overrideDefaultStreamReadConstraints(
+  /**
+   * Creates a thread-safe JsonFactory with custom stream read constraints.
+   *
+   * <p>This method encapsulates the logic to increase the default jackson-databind stream read
+   * constraint to 100MB. This functionality was introduced in Jackson 2.15 causing string > 20MB
+   * (5MB in <2.15.0) parsing failure. This has caused regressions in its dependencies including
+   * Beam. Here we create a streamReadConstraints minimum size limit set to 100MB and exposing the
+   * factory to higher limits. If needed, call this method during pipeline run time, e.g. in
+   * DoFn.setup. This avoids a data race caused by modifying the global default settings.
+   */
+  private static JsonFactory createJsonFactory(int sizeLimit) {
+    sizeLimit = Math.max(sizeLimit, MAX_STRING_LENGTH);
+    JsonFactory jsonFactory = new JsonFactory();
+    try {
+      // Check if StreamReadConstraints is available (Jackson 2.15+)
+      Class.forName("com.fasterxml.jackson.core.StreamReadConstraints");
+      com.fasterxml.jackson.core.StreamReadConstraints streamReadConstraints =
           com.fasterxml.jackson.core.StreamReadConstraints.builder()
-              .maxStringLength(newLimit)
-              .build());
+              .maxStringLength(MAX_STRING_LENGTH)
+              .build();
+      jsonFactory.setStreamReadConstraints(streamReadConstraints);
     } catch (ClassNotFoundException e) {
-      // <2.15, do nothing
+      // If the class is not found (i.e., Jackson version < 2.15), do nothing.
     }
-    defaultBufferLimit = newLimit;
-  }
-
-  static {
-    increaseDefaultStreamReadConstraints(100 * 1024 * 1024);
+    return jsonFactory;
   }
 
   public static ObjectMapper newObjectMapperWith(RowJson.RowJsonDeserializer deserializer) {
     SimpleModule module = new SimpleModule("rowDeserializationModule");
     module.addDeserializer(Row.class, deserializer);
 
-    ObjectMapper objectMapper = new ObjectMapper();
+    ObjectMapper objectMapper = new ObjectMapper(createJsonFactory(MAX_STRING_LENGTH));
     objectMapper.registerModule(module);
 
     return objectMapper;
@@ -80,7 +90,7 @@ public static ObjectMapper newObjectMapperWith(RowJson.RowJsonSerializer seriali
     SimpleModule module = new SimpleModule("rowSerializationModule");
     module.addSerializer(Row.class, serializer);
 
-    ObjectMapper objectMapper = new ObjectMapper();
+    ObjectMapper objectMapper = new ObjectMapper(createJsonFactory(MAX_STRING_LENGTH));
     objectMapper.registerModule(module);
 
     return objectMapper;