release 9.2.1

Signed-off-by: neo <[email protected]>

Author: neowu

CHANGELOG.md

@@ -1,13 +1,14 @@
 ## Change log
 
-### 9.2.0 (4/24/2025 - 5/19/2025)
+### 9.2.1 (4/24/2025 - 5/19/2025) !!! only support java 24
 
 * java: update to 24
   > refer to https://openjdk.org/jeps/491
-* search: update to ES 9.0.0
-  > migrated to apache httpclient 5   
+* search: update to ES 8.17.1
   > simplify logging
   > remove opentelemetry dependency
+  > !!! es java client 9.x doesn't support with es server 8.x, es java client is forward compatible
+  > refer to https://www.elastic.co/docs/reference/elasticsearch/clients/java
 * action: added ActionLogContext.maxProcessTime(duration)
   > for long and non-critical action (such as test jobs), set max process time to avoid SLOW_PROCESS warning
 

TODO.md

@@ -21,3 +21,5 @@
 * migrate to dragonflydb and support RESP3 (cluster / MOVED handling) ?
 * migrate to opensearch ?
 * log exporter, reimplement in rust?
+
+* update ES to 9.x

build.gradle.kts

@@ -7,7 +7,7 @@ apply(plugin = "project")
 
 subprojects {
     group = "core.framework"
-    version = "9.2.0"
+    version = "9.2.1"
     repositories {
         maven {
             url = uri("https://neowu.github.io/maven-repo/")
@@ -18,8 +18,8 @@ subprojects {
     }
 }
 
-val elasticVersion = "9.0.0"
-val jacksonVersion = "2.18.3"
+val elasticVersion = "8.18.1"
+val jacksonVersion = "2.18.4"
 val junitVersion = "5.12.2"
 val mockitoVersion = "5.17.0"
 val assertjVersion = "3.27.3"
@@ -90,7 +90,6 @@ project("core-ng-search") {
         api("co.elastic.clients:elasticsearch-java:${elasticVersion}") {
             exclude(group = "io.opentelemetry")
         }
-        implementation("commons-logging:commons-logging:1.3.5")
         implementation("com.fasterxml.jackson.core:jackson-databind:${jacksonVersion}")
         testImplementation(project(":core-ng-test"))
     }
@@ -109,7 +108,6 @@ project("core-ng-search-test") {
         implementation("core.framework.elasticsearch.module:lang-painless:${elasticVersion}")
         implementation("core.framework.elasticsearch.module:analysis-common:${elasticVersion}")     // used by elasticsearch stemmer
         implementation("core.framework.elasticsearch.module:reindex:${elasticVersion}")             // used by elasticsearch deleteByQuery
-        implementation("org.apache.httpcomponents:httpasyncclient:4.1.5")                           // used by reindex
         runtimeOnly("org.apache.logging.log4j:log4j-to-slf4j:2.19.0")
     }
 }

core-ng-search-test/src/main/java/core/framework/search/impl/LocalElasticSearch.java

@@ -2,7 +2,7 @@
 
 import core.framework.util.Files;
 import core.framework.util.StopWatch;
-import org.apache.hc.core5.http.HttpHost;
+import org.apache.http.HttpHost;
 import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.common.network.NetworkService;
 import org.elasticsearch.common.settings.Settings;

core-ng-search-test/src/main/java/core/framework/search/module/TestSearchConfig.java

@@ -3,7 +3,7 @@
 import core.framework.internal.module.ModuleContext;
 import core.framework.internal.module.ShutdownHook;
 import core.framework.search.impl.LocalElasticSearch;
-import org.apache.hc.core5.http.HttpHost;
+import org.apache.http.HttpHost;
 import org.elasticsearch.common.logging.LogConfigurator;
 
 import java.util.concurrent.locks.ReentrantLock;

core-ng-search-test/src/test/java/core/framework/search/impl/ElasticSearchAggregationIntegrationTest.java

@@ -62,7 +62,7 @@ void aggregate() {
         assertThat(response.hits).hasSize(1);
         assertThat(response.aggregations).containsKeys("total_value");
 
-        int sum = response.aggregations.get("total_value").sum().value().intValue();
+        int sum = (int) response.aggregations.get("total_value").sum().value();
         assertThat(sum).isEqualTo(21);
     }
 

core-ng-search-test/src/test/java/core/framework/search/impl/ElasticSearchIntegrationTest.java

@@ -124,7 +124,7 @@ void search() {
             .filter(term("enum_field", JSON.toEnumValue(TestDocument.TestEnum.VALUE1))));
 
         request.sorts.add(SortOptions.of(builder -> builder.script(s ->
-            s.script(script -> script.source(source -> source.scriptString("doc['int_field'].value * 3"))).type(ScriptSortType.Number))));
+            s.script(script -> script.source("doc['int_field'].value * 3")).type(ScriptSortType.Number))));
 
         SearchResponse<TestDocument> response = documentType.search(request);
 

core-ng-search/src/main/java/core/framework/search/ElasticSearchMigration.java

@@ -3,7 +3,7 @@
 import core.framework.internal.module.PropertyManager;
 import core.framework.search.impl.ElasticSearchHost;
 import core.framework.search.impl.ElasticSearchImpl;
-import org.apache.hc.core5.http.HttpHost;
+import org.apache.http.HttpHost;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

core-ng-search/src/main/java/core/framework/search/impl/ElasticSearchHost.java

@@ -1,7 +1,7 @@
 package core.framework.search.impl;
 
 import core.framework.util.Strings;
-import org.apache.hc.core5.http.HttpHost;
+import org.apache.http.HttpHost;
 
 /**
  * @author neo
@@ -34,6 +34,6 @@ private static HttpHost host(String value) {
             port = Integer.parseInt(value.substring(portIndex + 1));
             hostEnd = portIndex;
         }
-        return new HttpHost(schema, value.substring(hostStart, hostEnd), port);
+        return new HttpHost(value.substring(hostStart, hostEnd), port, schema);
     }
 }

core-ng-search/src/main/java/core/framework/search/impl/ElasticSearchImpl.java

@@ -6,11 +6,8 @@
 import co.elastic.clients.elasticsearch.indices.ElasticsearchIndicesClient;
 import co.elastic.clients.json.JsonData;
 import co.elastic.clients.json.jackson.JacksonJsonpMapper;
-import co.elastic.clients.transport.rest5_client.Rest5ClientTransport;
-import co.elastic.clients.transport.rest5_client.low_level.Request;
-import co.elastic.clients.transport.rest5_client.low_level.Response;
-import co.elastic.clients.transport.rest5_client.low_level.Rest5Client;
-import co.elastic.clients.transport.rest5_client.low_level.Rest5ClientBuilder;
+import co.elastic.clients.transport.instrumentation.NoopInstrumentation;
+import co.elastic.clients.transport.rest_client.RestClientTransport;
 import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import core.framework.internal.json.JSONMapper;
@@ -21,26 +18,22 @@
 import core.framework.search.SearchException;
 import core.framework.util.Encodings;
 import core.framework.util.StopWatch;
-import org.apache.hc.client5.http.config.ConnectionConfig;
-import org.apache.hc.client5.http.config.RequestConfig;
-import org.apache.hc.client5.http.impl.async.HttpAsyncClientBuilder;
-import org.apache.hc.client5.http.impl.nio.PoolingAsyncClientConnectionManager;
-import org.apache.hc.client5.http.impl.nio.PoolingAsyncClientConnectionManagerBuilder;
-import org.apache.hc.core5.http.Header;
-import org.apache.hc.core5.http.HttpEntity;
-import org.apache.hc.core5.http.HttpHost;
-import org.apache.hc.core5.http.io.entity.EntityUtils;
-import org.apache.hc.core5.http.message.BasicHeader;
-import org.apache.hc.core5.util.TimeValue;
-import org.apache.hc.core5.util.Timeout;
+import org.apache.http.Header;
+import org.apache.http.HttpEntity;
+import org.apache.http.HttpHost;
+import org.apache.http.message.BasicHeader;
+import org.apache.http.util.EntityUtils;
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.Response;
+import org.elasticsearch.client.RestClient;
+import org.elasticsearch.client.RestClientBuilder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.time.Duration;
 import java.util.Map;
-import java.util.concurrent.TimeUnit;
 
 /**
  * @author neo
@@ -53,47 +46,26 @@ public class ElasticSearchImpl implements ElasticSearch {
     public int maxResultWindow = 10000;
     ElasticsearchClient client;
     Header authHeader;
-    private Rest5Client restClient;
+    private RestClient restClient;
     private ObjectMapper mapper;
 
     // initialize will be called in startup hook, no need to synchronize
     public void initialize() {
-        if (client == null) {
-            // initialize can be called by initSearch explicitly during test,
-            Rest5ClientBuilder builder = Rest5Client.builder(hosts);
+        if (client == null) {   // initialize can be called by initSearch explicitly during test,
+            RestClientBuilder builder = RestClient.builder(hosts);
             if (authHeader != null) {
                 builder.setDefaultHeaders(new Header[]{authHeader});
             }
-
-            RequestConfig requestConfig = RequestConfig.custom()
-                .setConnectionRequestTimeout(Timeout.ofSeconds(5))
-                .setResponseTimeout((int) timeout.toMillis(), TimeUnit.MILLISECONDS)
-                .build();
-
-            ConnectionConfig connectionConfig = ConnectionConfig.custom()
-                .setConnectTimeout(Timeout.ofSeconds(5))
-                .setSocketTimeout((int) timeout.toMillis(), TimeUnit.MILLISECONDS)
-                .setTimeToLive(TimeValue.ofSeconds(30))
-                .build();
-
-            PoolingAsyncClientConnectionManager connectionManager =
-                PoolingAsyncClientConnectionManagerBuilder.create()
-                    .setDefaultConnectionConfig(connectionConfig)
-                    .setMaxConnPerRoute(100)
-                    .setMaxConnTotal(100)
-                    .build();
-
-            HttpAsyncClientBuilder httpClientBuilder = HttpAsyncClientBuilder.create()
-                .setDefaultRequestConfig(requestConfig)
-                .setConnectionManager(connectionManager)
-                .setUserAgent("elasticsearch-java/core-ng")
-                .setThreadFactory(Thread.ofPlatform().name("elasticsearch-rest-client-", 0).factory());
-
-            builder.setHttpClient(httpClientBuilder.build());
-
+            builder.setRequestConfigCallback(config -> config.setConnectionRequestTimeout(5_000)    // timeout of requesting connection from connection pool
+                .setConnectTimeout(5_000)   // 5s, usually es is within same network, use shorter timeout to fail fast
+                .setSocketTimeout((int) timeout.toMillis()));
+            builder.setHttpClientConfigCallback(config -> config.setMaxConnTotal(100)
+                .setMaxConnPerRoute(100)
+                .setKeepAliveStrategy((response, context) -> Duration.ofSeconds(30).toMillis())
+                .addInterceptorFirst(new ElasticSearchLogInterceptor()));
             restClient = builder.build();
             mapper = JSONMapper.builder().serializationInclusion(JsonInclude.Include.NON_NULL).build();
-            client = new ElasticsearchClient(new Rest5ClientTransport(restClient, new JacksonJsonpMapper(mapper), null, new ElasticSearchLogInstrumentation()));
+            client = new ElasticsearchClient(new RestClientTransport(restClient, new JacksonJsonpMapper(mapper), null, NoopInstrumentation.INSTANCE));
         }
     }
 
@@ -212,7 +184,7 @@ public void deleteIndex(String index) {
     public ClusterStateResponse state() {
         var watch = new StopWatch();
         try {
-            return client.cluster().state(builder -> builder.metric("metadata")).state().to(ClusterStateResponse.class);
+            return client.cluster().state(builder -> builder.metric("metadata")).valueBody().to(ClusterStateResponse.class);
         } catch (IOException e) {
             throw new UncheckedIOException(e);
         } catch (ElasticsearchException e) {

core-ng-search/src/main/java/core/framework/search/impl/ElasticSearchLogInstrumentation.java

@@ -0,0 +1,63 @@
+package core.framework.search.impl;
+
+import core.framework.internal.log.filter.LogParam;
+import org.apache.http.HttpEntity;
+import org.apache.http.HttpEntityEnclosingRequest;
+import org.apache.http.HttpRequest;
+import org.apache.http.HttpRequestInterceptor;
+import org.apache.http.RequestLine;
+import org.apache.http.protocol.HttpContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.Reader;
+import java.nio.charset.StandardCharsets;
+import java.util.Set;
+
+/**
+ * @author neo
+ */
+public class ElasticSearchLogInterceptor implements HttpRequestInterceptor {
+    private final Logger logger = LoggerFactory.getLogger(ElasticSearchLogInterceptor.class);
+
+    // only request can be logged, apache http client execute in different thread (NIO), and response entity can only be consumed once
+    @Override
+    public void process(HttpRequest request, HttpContext context) {
+        RequestLine requestLine = request.getRequestLine();
+        logger.debug("[request] method={}, uri={}", requestLine.getMethod(), requestLine.getUri());
+        if (request instanceof final HttpEntityEnclosingRequest entityRequest) {
+            HttpEntity entity = entityRequest.getEntity();
+            if (entity != null) {
+                logger.debug("[request] body={}", new BodyParam(entity));
+            }
+        }
+    }
+
+    record BodyParam(HttpEntity entity) implements LogParam {
+        @Override
+        public void append(StringBuilder builder, Set<String> maskedFields, int maxParamLength) {
+            // refer to co.elastic.clients.transport.rest_client.RestClientTransport.prepareLowLevelRequest
+            // it always uses ByteArrayEntity, thus always has content length
+            try (Reader reader = new InputStreamReader(entity.getContent(), StandardCharsets.UTF_8)) {
+                char[] buffer;
+
+                if (entity.isChunked()) {
+                    // with bulkIndex, es uses co.elastic.clients.transport.rest_client.MultiBufferEntity, which is chunked and content length = -1
+                    buffer = new char[maxParamLength + 1];
+                } else {
+                    buffer = new char[Math.min((int) entity.getContentLength(), maxParamLength + 1)];
+                }
+                int read = reader.read(buffer);
+
+                if (read > 0) builder.append(buffer, 0, Math.min(read, maxParamLength));    // read = -1 if nothing read into buffer
+                if (read > maxParamLength) {
+                    builder.append("...(truncated)");
+                }
+            } catch (IOException e) {
+                throw new Error(e); // not expected io exception, as it's from ByteArrayEntity
+            }
+        }
+    }
+}