[DRAFT] upgrade jackson used in elasticsearch connector

elasticsearch/src/main/resources/reference.conf

@@ -0,0 +1,28 @@
+# SPDX-License-Identifier: Apache-2.0
+
+##############################################
+# Pekko Connectors ElasticSearch Config File #
+##############################################
+
+# This is the reference config file that contains all the default settings.
+# Make your edits/overrides in your application.conf.
+
+pekko.connectors.elasticsearch {
+  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
+    }
+  }
+}

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

@@ -21,8 +21,11 @@ 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, JsonFactoryBuilder, StreamReadConstraints, StreamWriteConstraints }
 import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.databind.json.JsonMapper
 import com.fasterxml.jackson.databind.node.{ ArrayNode, NumericNode }
+import com.typesafe.config.ConfigFactory
 
 import scala.concurrent.ExecutionContext
 
@@ -38,7 +41,7 @@ object ElasticsearchSource {
   def create(elasticsearchParams: ElasticsearchParams,
       query: String,
       settings: SourceSettingsBase[_, _]): Source[ReadResult[java.util.Map[String, Object]], NotUsed] =
-    create(elasticsearchParams, query, settings, new ObjectMapper())
+    create(elasticsearchParams, query, settings, createObjectMapper())
 
   /**
    * Creates a [[pekko.stream.javadsl.Source]] from Elasticsearch that streams [[ReadResult]]s of [[java.util.Map]].
@@ -105,7 +108,7 @@ object ElasticsearchSource {
       query: String,
       settings: SourceSettingsBase[_, _],
       clazz: Class[T]): Source[ReadResult[T], NotUsed] =
-    typed[T](elasticsearchParams, query, settings, clazz, new ObjectMapper())
+    typed[T](elasticsearchParams, query, settings, clazz, createObjectMapper())
 
   /**
    * Creates a [[pekko.stream.javadsl.Source]] from Elasticsearch that streams [[ReadResult]]s of type `T`.
@@ -165,6 +168,26 @@ object ElasticsearchSource {
       }
       .mapMaterializedValue(_ => NotUsed)
 
+  private def createObjectMapper(): ObjectMapper = {
+    val config = ConfigFactory.load.getConfig("pekko.connectors.elasticsearch.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)
+
+  }
+
   private final class JacksonReader[T](mapper: ObjectMapper, clazz: Class[T]) extends impl.MessageReader[T] {
 
     override def convert(json: String): impl.ScrollResponse[T] = {

project/Dependencies.scala

@@ -92,6 +92,11 @@ object Dependencies {
     "com.fasterxml.jackson.core" % "jackson-core" % JacksonDatabindVersion,
     "com.fasterxml.jackson.core" % "jackson-databind" % JacksonDatabindVersion)
 
+  val JacksonDatabindVersion216 = "2.16.0"
+  val JacksonDatabindDependencies216 = Seq(
+    "com.fasterxml.jackson.core" % "jackson-core" % JacksonDatabindVersion216,
+    "com.fasterxml.jackson.core" % "jackson-databind" % JacksonDatabindVersion216)
+
   val Amqp = Seq(
     libraryDependencies ++= Seq(
       "com.rabbitmq" % "amqp-client" % "5.14.2") ++ Mockito)
@@ -152,7 +157,7 @@ object Dependencies {
     libraryDependencies ++= Seq(
       "org.apache.pekko" %% "pekko-http" % PekkoHttpVersion,
       "org.apache.pekko" %% "pekko-http-spray-json" % PekkoHttpVersion,
-      "org.slf4j" % "jcl-over-slf4j" % jclOverSlf4jVersion % Test) ++ JacksonDatabindDependencies)
+      "org.slf4j" % "jcl-over-slf4j" % jclOverSlf4jVersion % Test) ++ JacksonDatabindDependencies216)
 
   val File = Seq(
     libraryDependencies ++= Seq(