This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new c00ecb1c5c0 [improvement](external) Support Elasticsearch 9.x node
discovery (#69651)
c00ecb1c5c0 is described below
commit c00ecb1c5c016f3823dae48a87213e36cdf8d05d
Author: Gabriel <[email protected]>
AuthorDate: Tue Oct 27 14:07:28 2026 +0800
[improvement](external) Support Elasticsearch 9.x node discovery (#69651)
### What problem does this PR solve?
Related PR: #69650 (reader regression coverage)
With `nodes_discovery=true`, ES metadata loading follows
`EsMetadataFetcher.fetchShards -> EsConnectorRestClient.getHttpNodes ->
EsNodeInfo -> EsMajorVersion.parse`. The parser accepts only majors
through 8, so a 9.x node makes discovery fail with an
unsupported-version exception. The configured-seed path with discovery
disabled does not parse node versions and is not affected in the same
way.
Add the explicit ES 9 major, comparisons and latest-supported bound
while continuing to reject unvalidated future majors with a clear
diagnostic. Add an HTTP unit test containing mixed 8.19/9.0 nodes, IPv6
routing and a node without HTTP, plus an opt-in live compatibility suite
rather than relying on version parsing alone.
The live suite creates isolated temporary indices and checks the
production connector's `_nodes/http`, `_mapping`, `_search_shards`,
index listing and metadata fetching with discovery enabled/disabled. It
also verifies authenticated source/docvalue scroll requests, missing
values, pagination, scroll cleanup and invalid-credential failures.
Enable it with `DORIS_ES9_TEST_URL`, `DORIS_ES9_TEST_USER` and
`DORIS_ES9_TEST_PASSWORD`.
### Release note
Allow Elasticsearch 9.x nodes during ES catalog node discovery.
### Check List (For Author)
- Test
- [x] Version parsing and mixed-node tests reproduce the unsupported-9.x
failure before the change.
- [x] All 124 ES connector tests pass with live authenticated
Elasticsearch 9.0.0 and 9.5.5, including the three opt-in compatibility
tests on each version.
- [x] Default unit-test execution passes with the live suite disabled
when no endpoint is configured.
- [x] FE Checkstyle for the ES module and its reactor dependencies.
- [x] The BE ES reader also reads the authenticated three-row fixture
through source/docvalue scans with batch size 1 on both versions, using
the reader coverage change in the related ES regression PR.
- [ ] Full Doris SQL end-to-end validation against ES 9; the live tests
above validate connector metadata and HTTP reader compatibility.
- Behavior changed:
- [x] Yes: discovery accepts ES 9 nodes; later unverified majors remain
rejected.
- Does this need documentation?
- [x] No new catalog properties or configuration are introduced.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
.../apache/doris/connector/es/EsMajorVersion.java | 8 +-
.../doris/connector/es/Es9CompatibilityTest.java | 163 +++++++++++++++++++++
.../connector/es/EsConnectorRestClientTest.java | 31 ++++
.../doris/connector/es/EsMajorVersionTest.java | 26 +++-
4 files changed, 224 insertions(+), 4 deletions(-)
diff --git
a/fe/fe-connector/fe-connector-es/src/main/java/org/apache/doris/connector/es/EsMajorVersion.java
b/fe/fe-connector/fe-connector-es/src/main/java/org/apache/doris/connector/es/EsMajorVersion.java
index 853ea01c841..be2e776196d 100644
---
a/fe/fe-connector/fe-connector-es/src/main/java/org/apache/doris/connector/es/EsMajorVersion.java
+++
b/fe/fe-connector/fe-connector-es/src/main/java/org/apache/doris/connector/es/EsMajorVersion.java
@@ -32,7 +32,8 @@ public class EsMajorVersion {
public static final EsMajorVersion V_6_X = new EsMajorVersion((byte) 6,
"6.x");
public static final EsMajorVersion V_7_X = new EsMajorVersion((byte) 7,
"7.x");
public static final EsMajorVersion V_8_X = new EsMajorVersion((byte) 8,
"8.x");
- public static final EsMajorVersion LATEST = V_8_X;
+ public static final EsMajorVersion V_9_X = new EsMajorVersion((byte) 9,
"9.x");
+ public static final EsMajorVersion LATEST = V_9_X;
public final byte major;
private final String version;
@@ -88,6 +89,11 @@ public class EsMajorVersion {
if (version.startsWith("8.")) {
return new EsMajorVersion((byte) 8, version);
}
+ // Node discovery parses every node during rolling upgrades; ES 9 uses
the same typeless
+ // mapping and HTTP scan APIs as ES 8. Keep future, unverified majors
explicitly rejected.
+ if (version.startsWith("9.")) {
+ return new EsMajorVersion((byte) 9, version);
+ }
throw new DorisConnectorException(
"Unsupported/Unknown ES Cluster version [" + version + "]. "
+ "Highest supported version is [" + LATEST.version +
"].");
diff --git
a/fe/fe-connector/fe-connector-es/src/test/java/org/apache/doris/connector/es/Es9CompatibilityTest.java
b/fe/fe-connector/fe-connector-es/src/test/java/org/apache/doris/connector/es/Es9CompatibilityTest.java
new file mode 100644
index 00000000000..0b97f55e672
--- /dev/null
+++
b/fe/fe-connector/fe-connector-es/src/test/java/org/apache/doris/connector/es/Es9CompatibilityTest.java
@@ -0,0 +1,163 @@
+// 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.doris.connector.es;
+
+import org.apache.doris.connector.spi.DorisConnectorException;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable;
+
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Base64;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+
+/**
+ * Opt-in live compatibility checks. Set DORIS_ES9_TEST_URL,
DORIS_ES9_TEST_USER and
+ * DORIS_ES9_TEST_PASSWORD to an ES 9 instance with permission to create a
temporary test index.
+ */
+@EnabledIfEnvironmentVariable(named = "DORIS_ES9_TEST_URL", matches = ".+")
+class Es9CompatibilityTest {
+ private static final ObjectMapper JSON = new ObjectMapper();
+
+ private final HttpClient http =
HttpClient.newBuilder().connectTimeout(Duration.ofSeconds(10)).build();
+ private final String endpoint =
System.getenv("DORIS_ES9_TEST_URL").replaceAll("/+$", "");
+ private final String user =
System.getenv().getOrDefault("DORIS_ES9_TEST_USER", "elastic");
+ private final String password = System.getenv("DORIS_ES9_TEST_PASSWORD");
+ private final String index = "doris_es9_compat_" +
UUID.randomUUID().toString().replace("-", "");
+
+ @BeforeEach
+ void createFixture() throws Exception {
+ Assertions.assertNotNull(password, "DORIS_ES9_TEST_PASSWORD is
required");
+ Assertions.assertTrue(jsonRequest("GET", "/",
"").path("version").path("number").asText().startsWith("9."));
+ jsonRequest("PUT", "/" + index,
"{\"settings\":{\"number_of_shards\":2,\"number_of_replicas\":0},"
+ +
"\"mappings\":{\"properties\":{\"id\":{\"type\":\"integer\"},"
+ + "\"name\":{\"type\":\"keyword\"}}}}");
+ StringBuilder bulk = new StringBuilder();
+ for (int id = 1; id <= 3; id++) {
+ bulk.append("{\"index\":{\"_id\":\"").append(id).append("\"}}\n");
+ bulk.append("{\"id\":").append(id).append(",\"name\":")
+ .append(id == 3 ? "null" : id == 1 ? "\"alpha\"" :
"\"beta\"").append("}\n");
+ }
+ Assertions.assertFalse(jsonRequest("POST", "/" + index +
"/_bulk?refresh=true", bulk.toString())
+ .path("errors").asBoolean());
+ }
+
+ @AfterEach
+ void removeFixture() throws Exception {
+ HttpResponse<String> response = request("DELETE", "/" + index, "");
+ Assertions.assertTrue(response.statusCode() == 200 ||
response.statusCode() == 404);
+ }
+
+ @Test
+ void metadataDiscoveryAndSeedRoutingAcceptNine() {
+ EsConnectorRestClient client = client(password);
+ Assertions.assertTrue(client.existIndex(index));
+ Assertions.assertFalse(client.getHttpNodes().isEmpty());
+ Assertions.assertTrue(client.listTable(false).contains(index));
+ Assertions.assertEquals(2,
client.searchShards(index).getShardRoutings().size());
+ Assertions.assertEquals(2, EsTypeMapping.parseMapping(index,
client.getMapping(index), false).size());
+ for (boolean discover : new boolean[]{false, true}) {
+ EsMetadataState state = new EsMetadataState(index, null,
Arrays.asList("id", "name"),
+ discover, new String[]{endpoint});
+ EsMetadataState fetched = new EsMetadataFetcher(client, state,
null).fetch();
+ Assertions.assertEquals(2,
fetched.getShardPartitions().getShardRoutings().size());
+ Assertions.assertEquals(Map.of("id", "id", "name", "name"),
+ fetched.getFieldContext().getDocValueFieldsContext());
+
fetched.getShardPartitions().getShardRoutings().values().forEach(routes ->
routes.forEach(route -> {
+ Assertions.assertNotNull(route.getHttpHost());
+ Assertions.assertTrue(route.getHttpPort() > 0);
+ }));
+ }
+ }
+
+ @Test
+ void sourceAndDocvalueScrollsReturnAllRowsAndCanBeCleared() throws
Exception {
+ for (boolean docvalues : new boolean[]{false, true}) {
+ String projection = docvalues ?
"\"_source\":false,\"docvalue_fields\":[\"id\",\"name\"]"
+ : "\"_source\":[\"id\",\"name\"]";
+ JsonNode response = jsonRequest("POST", "/" + index +
"/_search?scroll=1m",
+
"{\"size\":1,\"sort\":[\"_doc\"],\"query\":{\"match_all\":{}}," + projection +
"}");
+ String scrollId = response.path("_scroll_id").asText();
+ Set<Integer> ids = new HashSet<>();
+ int pages = 0;
+ try {
+ while (!response.path("hits").path("hits").isEmpty()) {
+ for (JsonNode hit : response.path("hits").path("hits")) {
+ JsonNode row = hit.path(docvalues ? "fields" :
"_source");
+ int id = (docvalues ? row.path("id").path(0) :
row.path("id")).asInt();
+ Assertions.assertTrue(ids.add(id));
+ JsonNode name = docvalues ? row.path("name").path(0) :
row.path("name");
+ if (id == 3) {
+ Assertions.assertTrue(name.isNull() ||
name.isMissingNode());
+ } else {
+ Assertions.assertEquals(id == 1 ? "alpha" :
"beta", name.asText());
+ }
+ }
+ pages++;
+ response = jsonRequest("POST", "/_search/scroll",
JSON.writeValueAsString(
+ Map.of("scroll", "1m", "scroll_id", scrollId)));
+ scrollId = response.path("_scroll_id").asText();
+ }
+ Assertions.assertEquals(Set.of(1, 2, 3), ids);
+ Assertions.assertEquals(3, pages);
+ } finally {
+ Assertions.assertTrue(jsonRequest("DELETE", "/_search/scroll",
+ JSON.writeValueAsString(Map.of("scroll_id",
scrollId))).path("succeeded").asBoolean());
+ }
+ }
+ }
+
+ @Test
+ void invalidCredentialsFailInsteadOfHidingIndex() {
+ Assertions.assertThrows(DorisConnectorException.class, () ->
client(password + "-invalid").existIndex(index));
+ }
+
+ private EsConnectorRestClient client(String secret) {
+ return new EsConnectorRestClient(new String[]{endpoint}, user, secret,
false, null);
+ }
+
+ private JsonNode jsonRequest(String method, String path, String body)
throws Exception {
+ HttpResponse<String> response = request(method, path, body);
+ Assertions.assertTrue(response.statusCode() >= 200 &&
response.statusCode() < 300,
+ () -> method + " " + path + " returned HTTP " +
response.statusCode());
+ return JSON.readTree(response.body());
+ }
+
+ private HttpResponse<String> request(String method, String path, String
body) throws Exception {
+ String credentials = Base64.getEncoder().encodeToString((user + ":" +
password)
+ .getBytes(StandardCharsets.UTF_8));
+ HttpRequest request = HttpRequest.newBuilder(URI.create(endpoint +
path)).timeout(Duration.ofSeconds(30))
+ .header("Authorization", "Basic " +
credentials).header("Content-Type", "application/json")
+ .method(method,
HttpRequest.BodyPublishers.ofString(body)).build();
+ return http.send(request, HttpResponse.BodyHandlers.ofString());
+ }
+}
diff --git
a/fe/fe-connector/fe-connector-es/src/test/java/org/apache/doris/connector/es/EsConnectorRestClientTest.java
b/fe/fe-connector/fe-connector-es/src/test/java/org/apache/doris/connector/es/EsConnectorRestClientTest.java
index d1354c93744..874386a7b64 100644
---
a/fe/fe-connector/fe-connector-es/src/test/java/org/apache/doris/connector/es/EsConnectorRestClientTest.java
+++
b/fe/fe-connector/fe-connector-es/src/test/java/org/apache/doris/connector/es/EsConnectorRestClientTest.java
@@ -25,7 +25,9 @@ import org.junit.jupiter.api.Test;
import java.lang.reflect.Field;
import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
import java.util.Collections;
+import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
@@ -38,6 +40,35 @@ import java.util.concurrent.atomic.AtomicInteger;
*/
public class EsConnectorRestClientTest {
+ @Test
+ void mixedEightAndNineNodesAreDiscoveredDuringUpgrade() throws Exception {
+ String nodes = "{\"nodes\":{"
+ +
"\"eight\":{\"version\":\"8.19.0\",\"name\":\"eight\",\"roles\":[\"data\",\"ingest\"],"
+ + "\"http\":{\"publish_address\":\"127.0.0.1:9200\"}},"
+ +
"\"nine\":{\"version\":\"9.0.0\",\"name\":\"nine\",\"roles\":[\"data\"],"
+ + "\"http\":{\"publish_address\":\"node/[::1]:9201\"}},"
+ +
"\"no-http\":{\"version\":\"9.0.0\",\"name\":\"master\",\"roles\":[\"master\"]}}}";
+ byte[] body = nodes.getBytes(StandardCharsets.UTF_8);
+ HttpServer server = HttpServer.create(new
InetSocketAddress("127.0.0.1", 0), 0);
+ server.createContext("/_nodes/http", exchange -> {
+ exchange.sendResponseHeaders(200, body.length);
+ exchange.getResponseBody().write(body);
+ exchange.close();
+ });
+ server.start();
+ try {
+ EsConnectorRestClient client = new EsConnectorRestClient(
+ new String[]{"http://127.0.0.1:"; +
server.getAddress().getPort()}, null, null, false, null);
+ Map<String, EsNodeInfo> discovered = client.getHttpNodes();
+ Assertions.assertEquals(Set.of("eight", "nine"),
discovered.keySet());
+ Assertions.assertEquals("127.0.0.1:9200",
discovered.get("eight").getPublishAddress());
+ Assertions.assertEquals("[::1]:9201",
discovered.get("nine").getPublishAddress());
+ Assertions.assertTrue(discovered.get("nine").isData());
+ } finally {
+ server.stop(0);
+ }
+ }
+
@Test
public void indexAbsenceRequiresNotFoundResponse() throws Exception {
AtomicInteger status = new AtomicInteger(200);
diff --git
a/fe/fe-connector/fe-connector-es/src/test/java/org/apache/doris/connector/es/EsMajorVersionTest.java
b/fe/fe-connector/fe-connector-es/src/test/java/org/apache/doris/connector/es/EsMajorVersionTest.java
index 038c879afc1..667aa56cfe7 100644
---
a/fe/fe-connector/fe-connector-es/src/test/java/org/apache/doris/connector/es/EsMajorVersionTest.java
+++
b/fe/fe-connector/fe-connector-es/src/test/java/org/apache/doris/connector/es/EsMajorVersionTest.java
@@ -39,6 +39,23 @@ class EsMajorVersionTest {
Assertions.assertEquals(8, v.major);
}
+ @Test
+ void testParseVersion9() {
+ for (String version : new String[]{"9.0.0", "9.5.5"}) {
+ EsMajorVersion parsed = EsMajorVersion.parse(version);
+ Assertions.assertEquals(9, parsed.major);
+ Assertions.assertEquals(version, parsed.toString());
+ Assertions.assertTrue(parsed.after(EsMajorVersion.V_8_X));
+ Assertions.assertTrue(parsed.onOrAfter(EsMajorVersion.V_8_X));
+ Assertions.assertTrue(EsMajorVersion.V_8_X.before(parsed));
+ Assertions.assertTrue(EsMajorVersion.V_8_X.onOrBefore(parsed));
+ Assertions.assertFalse(parsed.before(EsMajorVersion.V_8_X));
+ }
+ Assertions.assertEquals(EsMajorVersion.parse("9.0.0"),
EsMajorVersion.parse("9.5.5"));
+ Assertions.assertEquals(EsMajorVersion.parse("9.0.0").hashCode(),
+ EsMajorVersion.parse("9.5.5").hashCode());
+ }
+
@Test
void testParseVersion6() {
EsMajorVersion v = EsMajorVersion.parse("6.8.23");
@@ -61,7 +78,7 @@ class EsMajorVersionTest {
@Test
void testParseUnsupportedVersionThrows() {
Assertions.assertThrows(DorisConnectorException.class,
- () -> EsMajorVersion.parse("9.0.0"));
+ () -> EsMajorVersion.parse("10.0.0"));
Assertions.assertThrows(DorisConnectorException.class,
() -> EsMajorVersion.parse("3.0.0"));
}
@@ -123,7 +140,10 @@ class EsMajorVersionTest {
}
@Test
- void testLatestIsV8() {
- Assertions.assertEquals(EsMajorVersion.V_8_X, EsMajorVersion.LATEST);
+ void testLatestIsV9() {
+ Assertions.assertEquals(EsMajorVersion.parse("9.0.0"),
EsMajorVersion.LATEST);
+ DorisConnectorException failure =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> EsMajorVersion.parse("10.0.0"));
+ Assertions.assertTrue(failure.getMessage().contains("Highest supported
version is [9.x]"));
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
