This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-connectors.git
The following commit(s) were added to refs/heads/main by this push:
new b7288eeac elasticsearch: escape names used in JSON requests (#1817)
b7288eeac is described below
commit b7288eeac2d360d847c41191dfc7d3db74e443b1
Author: PJ Fanning <[email protected]>
AuthorDate: Wed Aug 5 09:49:16 2026 +0100
elasticsearch: escape names used in JSON requests (#1817)
* elasticsearch: escape names used in JSON requests
* Update ElasticsearchSourceStage.scala
---
.../impl/ElasticsearchSourceStage.scala | 30 +++++++---
.../impl/ElasticsearchSearchBodySpec.scala | 65 ++++++++++++++++++++++
2 files changed, 88 insertions(+), 7 deletions(-)
diff --git
a/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/ElasticsearchSourceStage.scala
b/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/ElasticsearchSourceStage.scala
index 75056d867..8c7f71a87 100644
---
a/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/ElasticsearchSourceStage.scala
+++
b/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/ElasticsearchSourceStage.scala
@@ -77,6 +77,28 @@ object ElasticsearchSourceStage {
def validate(indexName: String): Unit = {
require(indexName != null, "You must define an index name")
}
+
+ /**
+ * INTERNAL API
+ *
+ * Construct a JSON search body from parameters. Keys are properly escaped
+ * using spray-json; values are treated as raw JSON.
+ */
+ @InternalApi
+ private[elasticsearch] def buildSearchBody(params: Map[String, String]):
String = {
+ val sb = new StringBuilder()
+ sb.append('{')
+ var first = true
+ params.foreach {
+ case (name, json) =>
+ if (first) first = false else sb.append(',')
+ sb.append(name.toJson.compactPrint)
+ sb.append(':')
+ sb.append(json)
+ }
+ sb.append('}')
+ sb.toString
+ }
}
/**
@@ -145,13 +167,7 @@ private[elasticsearch] final class
ElasticsearchSourceLogic[T](
val queryParams = baseMap ++ routingQueryParam ++ sortQueryParam
val completeParams = searchParams ++ extraParams.flatten - "routing"
- val searchBody = "{" +
- completeParams
- .map {
- case (name, json) =>
- "\"" + name + "\":" + json
- }
- .mkString(",") + "}"
+ val searchBody =
ElasticsearchSourceStage.buildSearchBody(completeParams)
val endpoint: String = settings.apiVersion match {
case ApiVersion.V5 =>
s"/${elasticsearchParams.indexName}/${elasticsearchParams.typeName.get}/_search"
diff --git
a/elasticsearch/src/test/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/ElasticsearchSearchBodySpec.scala
b/elasticsearch/src/test/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/ElasticsearchSearchBodySpec.scala
new file mode 100644
index 000000000..e0b5ded04
--- /dev/null
+++
b/elasticsearch/src/test/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/ElasticsearchSearchBodySpec.scala
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.connectors.elasticsearch.impl
+
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+
+class ElasticsearchSearchBodySpec extends AnyWordSpec with Matchers {
+
+ "ElasticsearchSourceStage.buildSearchBody" should {
+
+ "produce valid JSON for simple keys" in {
+ val body = ElasticsearchSourceStage.buildSearchBody(
+ Map("query" -> """{"match_all":{}}""", "size" -> "10"))
+ body should include(""""query":{"match_all":{}}""")
+ body should include(""""size":10""")
+ body should startWith("{")
+ body should endWith("}")
+ }
+
+ "escape double quotes in keys" in {
+ val body = ElasticsearchSourceStage.buildSearchBody(
+ Map("key\"injected" -> "true"))
+ body shouldBe """{"key\"injected":true}"""
+ }
+
+ "escape backslashes in keys" in {
+ val body = ElasticsearchSourceStage.buildSearchBody(
+ Map("key\\name" -> "true"))
+ body shouldBe """{"key\\name":true}"""
+ }
+
+ "escape newlines in keys" in {
+ val body = ElasticsearchSourceStage.buildSearchBody(
+ Map("key\nname" -> "true"))
+ body shouldBe """{"key\nname":true}"""
+ }
+
+ "handle empty map" in {
+ ElasticsearchSourceStage.buildSearchBody(Map.empty) shouldBe "{}"
+ }
+
+ "handle multiple entries" in {
+ val body = ElasticsearchSourceStage.buildSearchBody(
+ Map("a" -> "1", "b" -> "2"))
+ body should include(""""a":1""")
+ body should include(""""b":2""")
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
