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 d7c2df39a Elasticsearch: test Opensearch against 2.x and 3.x (#1933)
d7c2df39a is described below
commit d7c2df39a6eff87fa37b940ae24a79199856cbfd
Author: PJ Fanning <[email protected]>
AuthorDate: Fri Sep 18 09:03:05 2026 +0100
Elasticsearch: test Opensearch against 2.x and 3.x (#1933)
* Elasticsearch: test Opensearch against 2.x and 3.x
Motivation:
The Opensearch integration tests only ran against Opensearch 1.3.20.
The 1.x line is end-of-life (1.3.20 was its final release, May 2025), so
the connector had no coverage against any currently supported Opensearch.
Modification:
Replaced the `opensearch1` compose service with `opensearch2`
(2.19.6, port 9203) and `opensearch3` (3.8.0, port 9204), both with the
demo security config disabled and a 512m heap so the extra node does not
grow the CI footprint. Updated the elasticsearch CI pre_cmd to start
both. Turned `OpensearchV1Spec` into an abstract `OpensearchSpec` taking
the base URL, with `Opensearch2Spec` and `Opensearch3Spec` running the
whole spec against each server, and added port 9204 to
`OpensearchParameterizedTest`.
Result:
The Opensearch connector is exercised against both supported Opensearch
major lines instead of an end-of-life one. `OpensearchApiVersion.V1`
remains the only API version - 2.x and 3.x keep the Elasticsearch 7.x
compatible bulk and search API that it targets.
Tests:
- `sbt "elasticsearch/Test/compile"` - success
- `sbt "elasticsearch/scalafmtAll" "elasticsearch/javafmtAll"` - success
- Integration run not done locally: the containers start but this machine
cannot run the suite; relying on the `elasticsearch` CI job.
References:
None - keeping test dependencies on supported versions
* fix build
* Elasticsearch: bump test server to 7.17.29
Motivation:
The Elasticsearch 7 test server was pinned to 7.17.28, one patch behind
the latest 7.17 release.
Modification:
Bumped the `elasticsearch7` compose service to 7.17.29.
Result:
The connector is tested against the current 7.17 patch. `elasticsearch6`
stays on 6.8.23, which is the final 6.8 release. The test servers cannot
move to 8.x or 9.x without a new `ApiVersion`: those removed the mapping
types that `ApiVersion.V5` depends on.
Tests:
- `sbt "elasticsearch/Test/compile"` - success
- Integration run relies on the `elasticsearch` CI job.
References:
Refs #1933
---
.github/workflows/check-build-test.yml | 2 +-
docker-compose.yml | 17 ++++++++++--
docs/src/main/paradox/opensearch.md | 32 +++++++++++-----------
...{OpensearchV1Test.java => Opensearch2Test.java} | 2 +-
.../docs/javadsl/OpensearchParameterizedTest.java | 3 +-
...OpensearchV1Spec.scala => OpensearchSpec.scala} | 12 +++++---
6 files changed, 42 insertions(+), 26 deletions(-)
diff --git a/.github/workflows/check-build-test.yml
b/.github/workflows/check-build-test.yml
index 01052616b..5f58910a2 100644
--- a/.github/workflows/check-build-test.yml
+++ b/.github/workflows/check-build-test.yml
@@ -109,7 +109,7 @@ jobs:
- { connector: couchbase3, pre_cmd: 'docker
compose up -d couchbase8_prep' }
- { connector: csv }
- { connector: dynamodb, pre_cmd: 'docker
compose up -d dynamodb' }
- - { connector: elasticsearch, pre_cmd: 'docker
compose up -d elasticsearch6 elasticsearch7 opensearch1' }
+ - { connector: elasticsearch, pre_cmd: 'docker
compose up -d elasticsearch6 elasticsearch7 opensearch2 opensearch3' }
- { connector: file }
- { connector: ftp, pre_cmd:
'./scripts/ftp-servers.sh' }
- { connector: geode, pre_cmd: 'docker
compose up -d geode' }
diff --git a/docker-compose.yml b/docker-compose.yml
index 7e891fdff..694a36b56 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -39,18 +39,29 @@ services:
environment:
- "discovery.type=single-node"
elasticsearch7:
- image: docker.elastic.co/elasticsearch/elasticsearch:7.17.28
+ image: docker.elastic.co/elasticsearch/elasticsearch:7.17.29
ports:
- "9202:9200"
environment:
- "discovery.type=single-node"
- opensearch1:
- image: opensearchproject/opensearch:1.3.20
+ opensearch2:
+ image: opensearchproject/opensearch:2.19.6
ports:
- "9203:9200"
environment:
- "discovery.type=single-node"
+ - "DISABLE_INSTALL_DEMO_CONFIG=true"
- "DISABLE_SECURITY_PLUGIN=true"
+ - "OPENSEARCH_JAVA_OPTS=-Xms512m -Xmx512m"
+ opensearch3:
+ image: opensearchproject/opensearch:3.8.0
+ ports:
+ - "9204:9200"
+ environment:
+ - "discovery.type=single-node"
+ - "DISABLE_INSTALL_DEMO_CONFIG=true"
+ - "DISABLE_SECURITY_PLUGIN=true"
+ - "OPENSEARCH_JAVA_OPTS=-Xms512m -Xmx512m"
ftp:
image: stilliard/pure-ftpd:latest
ports:
diff --git a/docs/src/main/paradox/opensearch.md
b/docs/src/main/paradox/opensearch.md
index f266015c1..7510faea5 100644
--- a/docs/src/main/paradox/opensearch.md
+++ b/docs/src/main/paradox/opensearch.md
@@ -74,20 +74,20 @@ Use `ElasticsearchSource.typed` and
`ElasticsearchSink.create` to create source
@java[The data is converted to and from JSON by Jackson's ObjectMapper.]
Scala
-: @@snip
[snip](/elasticsearch/src/test/scala/docs/scaladsl/OpensearchV1Spec.scala) {
#run-typed }
+: @@snip
[snip](/elasticsearch/src/test/scala/docs/scaladsl/OpensearchSpec.scala) {
#run-typed }
Java
-: @@snip
[snip](/elasticsearch/src/test/java/docs/javadsl/OpensearchV1Test.java) {
#run-typed }
+: @@snip
[snip](/elasticsearch/src/test/java/docs/javadsl/Opensearch2Test.java) {
#run-typed }
### With JSON source
Use `ElasticsearchSource.create` and `ElasticsearchSink.create` to create
source and sink.
Scala
-: @@snip
[snip](/elasticsearch/src/test/scala/docs/scaladsl/OpensearchV1Spec.scala) {
#run-jsobject }
+: @@snip
[snip](/elasticsearch/src/test/scala/docs/scaladsl/OpensearchSpec.scala) {
#run-jsobject }
Java
-: @@snip
[snip](/elasticsearch/src/test/java/docs/javadsl/OpensearchV1Test.java) {
#run-jsobject }
+: @@snip
[snip](/elasticsearch/src/test/java/docs/javadsl/Opensearch2Test.java) {
#run-jsobject }
### Writing to Opensearch
@@ -103,10 +103,10 @@ In the above examples, `WriteMessage` is used as the
input to `ElasticsearchSink
| WriteMessage.createDeleteMessage | Delete an existing document. If there is
no document with the specified `id`, do nothing. |
Scala
-: @@snip
[snip](/elasticsearch/src/test/scala/docs/scaladsl/OpensearchV1Spec.scala) {
#multiple-operations }
+: @@snip
[snip](/elasticsearch/src/test/scala/docs/scaladsl/OpensearchSpec.scala) {
#multiple-operations }
Java
-: @@snip
[snip](/elasticsearch/src/test/java/docs/javadsl/OpensearchV1Test.java) {
#multiple-operations }
+: @@snip
[snip](/elasticsearch/src/test/java/docs/javadsl/Opensearch2Test.java) {
#multiple-operations }
### Source configuration
@@ -174,7 +174,7 @@ This will be used to:
1. transform the bulk request into a format understood by the corresponding
Opensearch server.
2. determine whether to include the index type mapping in the API calls. See
[removal of
types](https://www.elastic.co/guide/en/elasticsearch/reference/current/removal-of-types.html)
-Currently
[`V1`](https://opensearch.org/docs/1.3/opensearch/rest-api/document-apis/bulk/)
is supported specifically but this parameter does not need to match the server
version exactly (for example, `V1` should also work with Opensearch 2.x).
+Currently
[`V1`](https://opensearch.org/docs/latest/api-reference/document-apis/bulk/) is
supported specifically but this parameter does not need to match the server
version exactly (for example, `V1` also works with Opensearch 2.x and 3.x,
which the connector is tested against).
### Allow explicit index
@@ -186,10 +186,10 @@ You can also build flow stages with
@apidoc[ElasticsearchFlow$].
The API is similar to creating Sinks.
Scala
-: @@snip
[snip](/elasticsearch/src/test/scala/docs/scaladsl/OpensearchV1Spec.scala) {
#run-flow }
+: @@snip
[snip](/elasticsearch/src/test/scala/docs/scaladsl/OpensearchSpec.scala) {
#run-flow }
Java
-: @@snip
[snip](/elasticsearch/src/test/java/docs/javadsl/OpensearchV1Test.java) {
#run-flow }
+: @@snip
[snip](/elasticsearch/src/test/java/docs/javadsl/Opensearch2Test.java) {
#run-flow }
### Storing documents from Strings
@@ -197,10 +197,10 @@ Java
Opensearch requires the documents to be properly formatted JSON. If your data
is available as JSON in Strings, you may use the pre-defined
`StringMessageWriter` to avoid any conversions. For any other JSON
technologies, implement a @scala[`MessageWriter[T]`]@java[`MessageWriter<T>`].
Scala
-: @@snip
[snip](/elasticsearch/src/test/scala/docs/scaladsl/OpensearchV1Spec.scala) {
#string }
+: @@snip
[snip](/elasticsearch/src/test/scala/docs/scaladsl/OpensearchSpec.scala) {
#string }
Java
-: @@snip
[snip](/elasticsearch/src/test/java/docs/javadsl/OpensearchV1Test.java) {
#string }
+: @@snip
[snip](/elasticsearch/src/test/java/docs/javadsl/Opensearch2Test.java) {
#string }
@@ -209,10 +209,10 @@ Java
When streaming documents from Kafka, you might want to commit to Kafka
**AFTER** the document has been written to Opensearch.
Scala
-: @@snip
[snip](/elasticsearch/src/test/scala/docs/scaladsl/OpensearchV1Spec.scala) {
#kafka-example }
+: @@snip
[snip](/elasticsearch/src/test/scala/docs/scaladsl/OpensearchSpec.scala) {
#kafka-example }
Java
-: @@snip
[snip](/elasticsearch/src/test/java/docs/javadsl/OpensearchV1Test.java) {
#kafka-example }
+: @@snip
[snip](/elasticsearch/src/test/java/docs/javadsl/Opensearch2Test.java) {
#kafka-example }
### Specifying custom index-name for every document
@@ -221,7 +221,7 @@ When working with index-patterns using wildcards, you might
need to specify a cu
index-name for each document:
Scala
-: @@snip
[snip](/elasticsearch/src/test/scala/docs/scaladsl/OpensearchV1Spec.scala) {
#custom-index-name-example }
+: @@snip
[snip](/elasticsearch/src/test/scala/docs/scaladsl/OpensearchSpec.scala) {
#custom-index-name-example }
Java
: @@snip
[snip](/elasticsearch/src/test/java/docs/javadsl/OpensearchParameterizedTest.java)
{ #custom-index-name-example }
@@ -245,10 +245,10 @@ The easiest way of using Opensearch-source, is to just
specify the query-param.
like specifying which fields to return and so on. In such cases you can
instead use 'searchParams' instead:
Scala
-: @@snip
[snip](/elasticsearch/src/test/scala/docs/scaladsl/OpensearchV1Spec.scala) {
#custom-search-params }
+: @@snip
[snip](/elasticsearch/src/test/scala/docs/scaladsl/OpensearchSpec.scala) {
#custom-search-params }
Java
-: @@snip
[snip](/elasticsearch/src/test/java/docs/javadsl/OpensearchV1Test.java) {
#custom-search-params }
+: @@snip
[snip](/elasticsearch/src/test/java/docs/javadsl/Opensearch2Test.java) {
#custom-search-params }
#### Routing
diff --git a/elasticsearch/src/test/java/docs/javadsl/OpensearchV1Test.java
b/elasticsearch/src/test/java/docs/javadsl/Opensearch2Test.java
similarity index 99%
rename from elasticsearch/src/test/java/docs/javadsl/OpensearchV1Test.java
rename to elasticsearch/src/test/java/docs/javadsl/Opensearch2Test.java
index c9b94d92b..5d9695cad 100644
--- a/elasticsearch/src/test/java/docs/javadsl/OpensearchV1Test.java
+++ b/elasticsearch/src/test/java/docs/javadsl/Opensearch2Test.java
@@ -32,7 +32,7 @@ import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
-public class OpensearchV1Test extends ElasticsearchTestBase {
+public class Opensearch2Test extends ElasticsearchTestBase {
@BeforeAll
public static void setup() throws IOException {
setupBase();
diff --git
a/elasticsearch/src/test/java/docs/javadsl/OpensearchParameterizedTest.java
b/elasticsearch/src/test/java/docs/javadsl/OpensearchParameterizedTest.java
index 477df6585..fbf46bd8e 100644
--- a/elasticsearch/src/test/java/docs/javadsl/OpensearchParameterizedTest.java
+++ b/elasticsearch/src/test/java/docs/javadsl/OpensearchParameterizedTest.java
@@ -38,7 +38,8 @@ public class OpensearchParameterizedTest extends
ElasticsearchTestBase {
private OpensearchApiVersion apiVersion;
public static Stream<Arguments> data() {
- return Stream.of(Arguments.of(9203, OpensearchApiVersion.V1));
+ return Stream.of(
+ Arguments.of(9203, OpensearchApiVersion.V1), Arguments.of(9204,
OpensearchApiVersion.V1));
}
@AfterEach
diff --git a/elasticsearch/src/test/scala/docs/scaladsl/OpensearchV1Spec.scala
b/elasticsearch/src/test/scala/docs/scaladsl/OpensearchSpec.scala
similarity index 97%
rename from elasticsearch/src/test/scala/docs/scaladsl/OpensearchV1Spec.scala
rename to elasticsearch/src/test/scala/docs/scaladsl/OpensearchSpec.scala
index 5bba48a91..050abed40 100644
--- a/elasticsearch/src/test/scala/docs/scaladsl/OpensearchV1Spec.scala
+++ b/elasticsearch/src/test/scala/docs/scaladsl/OpensearchSpec.scala
@@ -35,10 +35,9 @@ import spray.json.jsonReader
import scala.concurrent.Future
import spray.json._
-class OpensearchV1Spec extends ElasticsearchSpecBase with
ElasticsearchSpecUtils {
+abstract class OpensearchSpec(baseUrl: String) extends ElasticsearchSpecBase
with ElasticsearchSpecUtils {
- private val connectionSettings: ElasticsearchConnectionSettings =
OpensearchConnectionSettings(
- "http://localhost:9203";)
+ private val connectionSettings: ElasticsearchConnectionSettings =
OpensearchConnectionSettings(baseUrl)
private val baseSourceSettings =
OpensearchSourceSettings(connectionSettings).withApiVersion(OpensearchApiVersion.V1)
private val baseWriteSettings =
OpensearchWriteSettings(connectionSettings).withApiVersion(OpensearchApiVersion.V1)
@@ -421,7 +420,8 @@ class OpensearchV1Spec extends ElasticsearchSpecBase with
ElasticsearchSpecUtils
// Assert no errors except a missing document for a update request
val errorMessages = results.flatMap(_.errorReason)
errorMessages should have size 1
- errorMessages.head shouldEqual "[_doc][00004]: document missing"
+ // Opensearch 2.x dropped mapping types, so the error reason no longer
carries a `[_doc]` prefix
+ errorMessages.head shouldEqual "[00004]: document missing"
flushAndRefresh(connectionSettings, indexName)
// Assert docs in sink8/_doc
@@ -527,3 +527,7 @@ class OpensearchV1Spec extends ElasticsearchSpecBase with
ElasticsearchSpecUtils
}
}
}
+
+class Opensearch2Spec extends OpensearchSpec("http://localhost:9203";)
+
+class Opensearch3Spec extends OpensearchSpec("http://localhost:9204";)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
