update bigquery code

Author: pjfanning

elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/javadsl/ElasticsearchSource.scala

@@ -21,7 +21,7 @@ import pekko.stream.connectors.elasticsearch.{ impl, _ }
 import pekko.stream.javadsl.Source
 import pekko.stream.{ Attributes, Materializer }
 import pekko.util.ccompat.JavaConverters._
-import com.fasterxml.jackson.core.{ JsonFactory, StreamReadConstraints, StreamWriteConstraints }
+import com.fasterxml.jackson.core.{ JsonFactory, JsonFactoryBuilder, StreamReadConstraints, StreamWriteConstraints }
 import com.fasterxml.jackson.databind.ObjectMapper
 import com.fasterxml.jackson.databind.json.JsonMapper
 import com.fasterxml.jackson.databind.node.{ ArrayNode, NumericNode }
@@ -180,12 +180,11 @@ object ElasticsearchSource {
     val streamWriteConstraints = StreamWriteConstraints.builder
       .maxNestingDepth(config.getInt("write.max-nesting-depth"))
       .build
-    val jsonFactory = JsonFactory.builder
+    val jsonFactory = JsonFactory.builder.asInstanceOf[JsonFactoryBuilder]
       .streamReadConstraints(streamReadConstraints)
       .streamWriteConstraints(streamWriteConstraints)
       .build
     new JsonMapper(jsonFactory)
-
   }
 
   private final class JacksonReader[T](mapper: ObjectMapper, clazz: Class[T]) extends impl.MessageReader[T] {

google-cloud-bigquery/src/main/resources/reference.conf

@@ -7,5 +7,23 @@ pekko.connectors.google {
     # BigQuery has a hard limit of 1,500 load jobs per table per day (just over 1 job per minute)
     # This sets the rate limit when loading data via BigQuery.insertAllAsync
     load-job-per-table-quota = 1 minute
+
+    jackson {
+      read {
+        # see https://www.javadoc.io/static/com.fasterxml.jackson.core/jackson-core/2.16.0/com/fasterxml/jackson/core/StreamReadConstraints.html
+        # these defaults are the same as the defaults in `StreamReadConstraints`
+        max-nesting-depth = 1000
+        max-number-length = 1000
+        max-string-length = 20000000
+        max-name-length = 50000
+        # max-document-length of -1 means unlimited
+        max-document-length = -1
+      }
+      write {
+        # see https://www.javadoc.io/static/com.fasterxml.jackson.core/jackson-core/2.16.0/com/fasterxml/jackson/core/StreamWriteConstraints.html
+        # these defaults are the same as the defaults in `StreamWriteConstraints`
+        max-nesting-depth = 1000
+      }
+    }
   }
 }

google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/javadsl/jackson/BigQueryMarshallers.scala

@@ -21,12 +21,15 @@ import pekko.http.javadsl.unmarshalling.Unmarshaller
 import pekko.stream.connectors.googlecloud.bigquery.model.QueryResponse
 import pekko.stream.connectors.googlecloud.bigquery.model.{ TableDataInsertAllRequest, TableDataListResponse }
 import com.fasterxml.jackson.databind.{ JavaType, MapperFeature, ObjectMapper }
+import com.fasterxml.jackson.core.{ JsonFactory, JsonFactoryBuilder, StreamReadConstraints, StreamWriteConstraints }
+import com.fasterxml.jackson.databind.json.JsonMapper
+import com.typesafe.config.ConfigFactory
 
 import java.io.IOException
 
 object BigQueryMarshallers {
 
-  private val defaultObjectMapper = new ObjectMapper().enable(MapperFeature.SORT_PROPERTIES_ALPHABETICALLY)
+  private val defaultObjectMapper = createObjectMapper().enable(MapperFeature.SORT_PROPERTIES_ALPHABETICALLY)
 
   /**
    * [[pekko.http.javadsl.unmarshalling.Unmarshaller]] for [[pekko.stream.connectors.googlecloud.bigquery.model.TableDataListResponse]]
@@ -99,4 +102,23 @@ object BigQueryMarshallers {
       case e: IOException =>
         throw new IllegalArgumentException("Cannot unmarshal JSON as " + expectedType.getTypeName, e)
     }
+
+  private def createObjectMapper(): ObjectMapper = {
+    val config = ConfigFactory.load.getConfig("pekko.connectors.google.bigquery.jackson")
+    val streamReadConstraints = StreamReadConstraints.builder
+      .maxNestingDepth(config.getInt("read.max-nesting-depth"))
+      .maxNumberLength(config.getInt("read.max-number-length"))
+      .maxStringLength(config.getInt("read.max-string-length"))
+      .maxNameLength(config.getInt("read.max-name-length"))
+      .maxDocumentLength(config.getLong("read.max-document-length"))
+      .build
+    val streamWriteConstraints = StreamWriteConstraints.builder
+      .maxNestingDepth(config.getInt("write.max-nesting-depth"))
+      .build
+    val jsonFactory = JsonFactory.builder.asInstanceOf[JsonFactoryBuilder]
+      .streamReadConstraints(streamReadConstraints)
+      .streamWriteConstraints(streamWriteConstraints)
+      .build
+    new JsonMapper(jsonFactory)
+  }
 }

project/Dependencies.scala

@@ -214,9 +214,9 @@ object Dependencies {
       "org.apache.pekko" %% "pekko-http-jackson" % PekkoHttpVersion % Provided,
       "org.apache.pekko" %% "pekko-http-spray-json" % PekkoHttpVersion,
       "io.spray" %% "spray-json" % "1.3.6",
-      "com.fasterxml.jackson.core" % "jackson-annotations" % JacksonDatabindVersion,
-      "com.fasterxml.jackson.datatype" % "jackson-datatype-jsr310" % JacksonDatabindVersion % Test,
-      "io.specto" % "hoverfly-java" % hoverflyVersion % Test) ++ Mockito)
+      "com.fasterxml.jackson.core" % "jackson-annotations" % JacksonDatabindVersion216,
+      "com.fasterxml.jackson.datatype" % "jackson-datatype-jsr310" % JacksonDatabindVersion216 % Test,
+      "io.specto" % "hoverfly-java" % hoverflyVersion % Test) ++ JacksonDatabindDependencies216 ++ Mockito)
   val GoogleBigQueryStorage = Seq(
     // see Pekko gRPC version in plugins.sbt
     libraryDependencies ++= Seq(