JsonToRowTest: fix race condition

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;
@@ -34,9 +35,14 @@
 @Internal
 public class RowJsonUtils {
 
+  // The maximum string length for the JSON parser, set to 100 MB.
+  public static final int MAX_STRING_LENGTH = 100 * 1024 * 1024;
+
   //
   private static int defaultBufferLimit;
 
+  private static final boolean STREAM_READ_CONSTRAINTS_AVAILABLE = streamReadConstraintsAvailable();
+
   /**
    * Increase the default jackson-databind stream read constraint.
    *
@@ -63,14 +69,52 @@ public static void increaseDefaultStreamReadConstraints(int newLimit) {
   }
 
   static {
-    increaseDefaultStreamReadConstraints(100 * 1024 * 1024);
+    increaseDefaultStreamReadConstraints(MAX_STRING_LENGTH);
+  }
+
+  private static boolean streamReadConstraintsAvailable() {
+    try {
+      Class.forName("com.fasterxml.jackson.core.StreamReadConstraints");
+      return true;
+    } catch (ClassNotFoundException e) {
+      return false;
+    }
+  }
+
+  private static class StreamReadConstraintsHelper {
+    static void setStreamReadConstraints(JsonFactory jsonFactory, int sizeLimit) {
+      com.fasterxml.jackson.core.StreamReadConstraints streamReadConstraints =
+          com.fasterxml.jackson.core.StreamReadConstraints.builder()
+              .maxStringLength(sizeLimit)
+              .build();
+      jsonFactory.setStreamReadConstraints(streamReadConstraints);
+    }
+  }
+
+  /**
+   * 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.
+   */
+  public static JsonFactory createJsonFactory(int sizeLimit) {
+    sizeLimit = Math.max(sizeLimit, MAX_STRING_LENGTH);
+    JsonFactory jsonFactory = new JsonFactory();
+    if (STREAM_READ_CONSTRAINTS_AVAILABLE) {
+      StreamReadConstraintsHelper.setStreamReadConstraints(jsonFactory, sizeLimit);
+    }
+    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 +124,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;

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowJsonCoder.java

@@ -75,10 +75,8 @@ public long getEncodedElementByteSize(TableRow value) throws Exception {
   private static final TypeDescriptor<TableRow> TYPE_DESCRIPTOR;
 
   static {
-    RowJsonUtils.increaseDefaultStreamReadConstraints(100 * 1024 * 1024);
-
     MAPPER =
-        new ObjectMapper()
+        new ObjectMapper(RowJsonUtils.createJsonFactory(RowJsonUtils.MAX_STRING_LENGTH))
             .registerModule(new JavaTimeModule())
             .registerModule(new JodaModule())
             .disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS)