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