es 8.9.0 support

.travis.yml

@@ -5,7 +5,7 @@ jdk:
 
 before_install:
   - sudo rm -rf /var/lib/elasticsearch
-  - curl https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-8.8.2-amd64.deb -o elasticsearch.deb && sudo dpkg -i --force-confnew elasticsearch.deb
+  - curl https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-8.9.0-amd64.deb -o elasticsearch.deb && sudo dpkg -i --force-confnew elasticsearch.deb
   - sudo cp ./src/test/resources/elasticsearch.yml /etc/elasticsearch/elasticsearch.yml
   - sudo cat /etc/elasticsearch/elasticsearch.yml
   - sudo java -version

pom.xml

@@ -3,7 +3,7 @@
 	<modelVersion>4.0.0</modelVersion>
 	<groupId>org.nlpcn</groupId>
 	<artifactId>elasticsearch-sql</artifactId>
-	<version>8.8.2.0</version>
+	<version>8.9.0.0</version>
 	<packaging>jar</packaging>
 	<description>Query elasticsearch using SQL</description>
 	<name>elasticsearch-sql</name>
@@ -44,7 +44,7 @@
 		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
 		<runSuite>**/MainTestSuite.class</runSuite>
 		<elasticsearch.plugin.name>sql</elasticsearch.plugin.name>
-		<elasticsearch.version>8.8.2</elasticsearch.version>
+		<elasticsearch.version>8.9.0</elasticsearch.version>
 		<elasticsearch.plugin.classname>org.elasticsearch.plugin.nlpcn.SqlPlug</elasticsearch.plugin.classname>
         <druid.version>1.2.15</druid.version>
         <guava.version>32.0.0-jre</guava.version>

src/main/java/org/elasticsearch/plugin/nlpcn/ActionRequestRestExecuter.java

@@ -3,10 +3,9 @@
 import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.admin.indices.get.GetIndexRequest;
 import org.elasticsearch.action.search.SearchRequest;
-import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.rest.RestChannel;
-import org.elasticsearch.rest.action.RestStatusToXContentListener;
+import org.elasticsearch.rest.action.RestChunkedToXContentListener;
 import org.nlpcn.es4sql.exception.SqlParseException;
 import org.nlpcn.es4sql.query.SqlElasticDeleteByQueryRequestBuilder;
 import org.nlpcn.es4sql.query.SqlElasticRequestBuilder;
@@ -40,7 +39,7 @@ public void execute() throws Exception {
             executeJoinRequestAndSendResponse();
         }
 		else if (request instanceof SearchRequest) {
-			client.search((SearchRequest) request, new RestStatusToXContentListener<SearchResponse>(channel));
+			client.search((SearchRequest) request, new RestChunkedToXContentListener<>(channel));
 		} else if (requestBuilder instanceof SqlElasticDeleteByQueryRequestBuilder) {
             throw new UnsupportedOperationException("currently not support delete on elastic 2.0.0");
         }

src/main/java/org/elasticsearch/plugin/nlpcn/client/ResponseConverter.java

@@ -18,6 +18,7 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.Build;
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.Version;
 import org.elasticsearch.action.admin.cluster.node.info.NodeInfo;
 import org.elasticsearch.action.admin.cluster.node.info.NodesInfoResponse;
@@ -419,7 +420,7 @@ public NodesInfoResponse nodesInfoResponse(co.elastic.clients.elasticsearch.node
                 if (Objects.nonNull(nodeInfo.totalIndexingBuffer())) {
                     totalIndexingBuffer = ByteSizeValue.ofBytes(nodeInfo.totalIndexingBuffer());
                 }
-                nodes.add(new NodeInfo(version, version.transportVersion, build, node, settings, os, process, jvm, threadPool, transport, http, null, plugins, ingest, aggsInfo, totalIndexingBuffer));
+                nodes.add(new NodeInfo(version, TransportVersion.current(), build, node, settings, os, process, jvm, threadPool, transport, http, null, plugins, ingest, aggsInfo, totalIndexingBuffer));
             }
         }
         return new NodesInfoResponse(new ClusterName(nodesInfoResponse.clusterName()), nodes, Collections.emptyList());

src/main/java/org/elasticsearch/plugin/nlpcn/executors/ElasticDefaultRestExecutor.java

@@ -10,6 +10,7 @@
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.rest.RestResponse;
+import org.elasticsearch.rest.action.RestChunkedToXContentListener;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.reindex.BulkIndexByScrollResponseContentListener;
 import org.elasticsearch.index.reindex.DeleteByQueryRequest;
@@ -21,7 +22,6 @@
 import org.elasticsearch.plugin.nlpcn.MultiRequestExecutorFactory;
 import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestStatus;
-import org.elasticsearch.rest.action.RestStatusToXContentListener;
 import org.elasticsearch.search.SearchHits;
 import org.nlpcn.es4sql.query.QueryAction;
 import org.nlpcn.es4sql.query.SqlElasticRequestBuilder;
@@ -63,7 +63,7 @@ public void execute(Client client, Map<String, String> params, QueryAction query
         } else if (request instanceof SearchRequest) {
             //zhongshu-comment 对应的QueryAction实现子类：DefaultQueryAction、AggregationQueryAction
             //zhongshu-comment 对应的SqlElasticRequestBuilder实现子类：SqlElasticSearchRequestBuilder
-            client.search((SearchRequest) request, new RestStatusToXContentListener<>(channel));
+            client.search((SearchRequest) request, new RestChunkedToXContentListener<>(channel));
         } else if (request instanceof DeleteByQueryRequest) {
             //zhongshu-comment 对应的QueryAction实现子类：DeleteQueryAction
             //zhongshu-comment 对应的SqlElasticRequestBuilder实现子类：SqlElasticDeleteByQueryRequestBuilder
@@ -73,7 +73,7 @@ public void execute(Client client, Map<String, String> params, QueryAction query
             //zhongshu-comment 对应的SqlElasticRequestBuilder实现子类：是一个匿名内部类，跳进去queryAction.explain()看
             requestBuilder.getBuilder().execute(new GetIndexRequestRestListener(channel, (GetIndexRequest) request));
         } else if (request instanceof SearchScrollRequest) {
-            client.searchScroll((SearchScrollRequest) request, new RestStatusToXContentListener<>(channel));
+            client.searchScroll((SearchScrollRequest) request, new RestChunkedToXContentListener<>(channel));
         } else {
             throw new Exception(String.format("Unsupported ActionRequest provided: %s", request.getClass().getName()));
         }

src/test/java/org/nlpcn/es4sql/MainTestSuite.java

@@ -3,7 +3,6 @@
 import co.elastic.clients.elasticsearch.ElasticsearchClient;
 import co.elastic.clients.json.jackson.JacksonJsonpMapper;
 import co.elastic.clients.transport.ElasticsearchTransport;
-import co.elastic.clients.transport.TransportOptions;
 import co.elastic.clients.transport.Version;
 import co.elastic.clients.transport.rest_client.RestClientOptions;
 import co.elastic.clients.transport.rest_client.RestClientTransport;
@@ -448,7 +447,7 @@ private static TransportAddress getTransportAddress() throws UnknownHostExceptio
     }
 
     private static ElasticsearchTransport getElasticsearchTransport(RestClient restClient) {
-        TransportOptions.Builder transportOptionsBuilder = new RestClientOptions(RequestOptions.DEFAULT).toBuilder();
+        RestClientOptions.Builder transportOptionsBuilder = new RestClientOptions(RequestOptions.DEFAULT).toBuilder();
 
         ContentType jsonContentType = Version.VERSION == null ? ContentType.APPLICATION_JSON
                 : ContentType.create("application/vnd.elasticsearch+json",
@@ -463,7 +462,7 @@ private static ElasticsearchTransport getElasticsearchTransport(RestClient restC
         setHeaderIfNotPresent.accept("Content-Type");
         setHeaderIfNotPresent.accept("Accept");
 
-        TransportOptions transportOptionsWithHeader = transportOptionsBuilder.build();
+        RestClientOptions transportOptionsWithHeader = transportOptionsBuilder.build();
         return new RestClientTransport(restClient, new JacksonJsonpMapper(), transportOptionsWithHeader);
     }
 }