vernedeng commented on code in PR #9057:
URL: https://github.com/apache/inlong/pull/9057#discussion_r1378738335
##########
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sink/es/ElasticsearchApi.java:
##########
@@ -17,98 +17,131 @@
package org.apache.inlong.manager.service.resource.sink.es;
+import org.apache.inlong.manager.common.util.HttpUtils;
+import org.apache.inlong.manager.pojo.sink.es.ElasticsearchCreateIndexResponse;
import org.apache.inlong.manager.pojo.sink.es.ElasticsearchFieldInfo;
-import org.apache.commons.collections.CollectionUtils;
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+import com.google.gson.JsonObject;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
-import org.elasticsearch.action.admin.indices.get.GetIndexRequest;
-import org.elasticsearch.action.search.SearchRequest;
-import org.elasticsearch.action.search.SearchResponse;
-import org.elasticsearch.action.support.master.AcknowledgedResponse;
-import org.elasticsearch.client.RequestOptions;
-import org.elasticsearch.client.RestHighLevelClient;
-import org.elasticsearch.client.indices.CreateIndexRequest;
-import org.elasticsearch.client.indices.CreateIndexResponse;
-import org.elasticsearch.client.indices.GetMappingsRequest;
-import org.elasticsearch.client.indices.PutMappingRequest;
-import org.elasticsearch.cluster.metadata.MappingMetaData;
-import org.elasticsearch.common.xcontent.XContentType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.core.ParameterizedTypeReference;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpMethod;
import org.springframework.stereotype.Component;
+import org.springframework.web.client.HttpClientErrorException.NotFound;
+import sun.misc.BASE64Encoder;
import java.io.IOException;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
/**
* elasticsearch template service
*/
+@Slf4j
@Component
public class ElasticsearchApi {
+ private static final Gson GSON = new GsonBuilder().create();
+
+ private static final String MAPPINGS_KEY = "mappings";
+
private static final String FIELD_KEY = "properties";
+ private static final String CONTENT_TYPE_KEY = "Content-Type";
+
+ private static final String CONTENT_TYPE_VALUE =
"application/json;charset=UTF-8";
+
private static final Logger LOG =
LoggerFactory.getLogger(ElasticsearchApi.class);
@Autowired
private ElasticsearchConfig esConfig;
/**
- * Search
+ * Get http headers by token.
*
- * @param searchRequest The search request of Elasticsearch
- * @return Search reponse of Elasticsearch
- * @throws IOException The io exception may throws
+ * @return http header infos
*/
- public SearchResponse search(SearchRequest searchRequest) throws
IOException {
- return search(searchRequest, RequestOptions.DEFAULT);
+ private HttpHeaders getHttpHeaders() {
+ HttpHeaders headers = new HttpHeaders();
+ headers.add(CONTENT_TYPE_KEY, CONTENT_TYPE_VALUE);
+ if (esConfig.getAuthEnable()) {
+ if (StringUtils.isNotEmpty(esConfig.getUsername()) &&
StringUtils.isNotEmpty(esConfig.getPassword())) {
+ String tokenStr = esConfig.getUsername() + ":" +
esConfig.getPassword();
+ String token = String.valueOf(new
BASE64Encoder().encode(tokenStr.getBytes(StandardCharsets.UTF_8)));
+ headers.add("Authorization", "Basic " + token);
+ }
+ }
+ return headers;
}
/**
- * Search
+ * Search.
*
- * @param searchRequest The search request of Elasticsearch
- * @param options The options of Elasticsearch
- * @return Search reponse of Elasticsearch
- * @throws IOException The io exception may throws
+ * @param indexName The index name
+ * @param request The request json
+ * @return the elasticsearch seqrch result
+ * @throws Exception any exception if occurred
*/
- public SearchResponse search(SearchRequest searchRequest, RequestOptions
options) throws IOException {
- LOG.info("get es search request of {}",
searchRequest.source().toString());
- return getEsClient().search(searchRequest, options);
+ public JsonObject search(String indexName, JsonObject request) throws
Exception {
+ LOG.info("get es search es index:{} request:{}", indexName,
request.toString());
+ final String url = esConfig.getOneHttpUrl() + "/" + indexName +
"/_search";
+ return HttpUtils.request(esConfig.getRestClient(), url,
HttpMethod.POST, request.toString(), getHttpHeaders(),
+ JsonObject.class);
}
/**
- * Check index exists
+ * Check index exists.
*
- * @param indexName The index name of Elasticsearch
- * @return true if exists else false
- * @throws IOException The exception may throws
+ * @param indexName The elasticsearch index name
+ * @return true or false
+ * @throws Exception any exception if occurred
*/
- public boolean indexExists(String indexName) throws IOException {
- GetIndexRequest getIndexRequest = new GetIndexRequest();
- getIndexRequest.indices(indexName);
- return getEsClient().indices().exists(getIndexRequest,
RequestOptions.DEFAULT);
+ public boolean indexExists(String indexName) throws Exception {
+ final String url = esConfig.getOneHttpUrl() + "/" + indexName;
+ try {
+ return HttpUtils.headRequest(esConfig.getRestClient(), url, null,
getHttpHeaders());
+ } catch (NotFound e) {
+ return false;
+ } catch (Exception e) {
+ throw e;
Review Comment:
do something, or not catch this expcetion
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
