diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/RowJsonUtils.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/RowJsonUtils.java index 408143fb1ebe..c83048ca8def 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/RowJsonUtils.java +++ b/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. + * + *

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; diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowJsonCoder.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowJsonCoder.java index 8cf3eeb479c0..f8e877fe98e6 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowJsonCoder.java +++ b/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 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)