You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2021/07/08 10:16:12 UTC
[GitHub] [pulsar] eolivelli opened a new pull request #11263: Enhanced ElasticSearch Sink:
eolivelli opened a new pull request #11263:
URL: https://github.com/apache/pulsar/pull/11263
### Motivation
This patch is a set of enhancements to the ElasticSearch Sink:
- support TLS
- support authentication
- support KeyValue schema and other Pulsar Schemas
- support Bulk writes
- support "errors action" (FAIL, IGNORE...)
- support AVRO records (Avro to JSON conversion)
- support for using the "key" of the record as record "_id" in ES (supporting both Message key and KeyValue encoding)
- support for creating the "_id" in ES using a selection of values from the Message (in case of AVRO/JSON message)
- support DELETING documents in case of non empty key and null value
By default the Sink behaves the same way it did with Pulsar 2.8.0, that is basically reading from the Topic and writing to ElasticSearch.
Schema related features are to be enabled explicitly by configuring "schemaEnable: true" in the configuration.
This patch also re-enables ElasticSearch integration tests, that were disabled in 2.8.0.
### Verifying this change
- this change add unit tests that cover the changes and the new features
- this change re-enables ES integration tests (disabled in 2.8.0?)
We (at DataStax/Kesque) have also tested manually the compatibility with the previous version of the sink.
It is hard to setup a backward compatibility test and therefore it will add cost on CI.
### Documentation
Once this patch is merged I will be happy to enhance the documentation, because we have many new features.
--
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-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli closed pull request #11263: Enhanced ElasticSearch Sink (TLS, Auth, KeyValue Schema, INSERT/UPDATE/DELETE, "_id" management, multi index support)
Posted by GitBox <gi...@apache.org>.
eolivelli closed pull request #11263:
URL: https://github.com/apache/pulsar/pull/11263
--
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-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on a change in pull request #11263: Enhanced ElasticSearch Sink (TLS, Auth, KeyValue Schema, INSERT/UPDATE/DELETE, "_id" management, multi index support)
Posted by GitBox <gi...@apache.org>.
eolivelli commented on a change in pull request #11263:
URL: https://github.com/apache/pulsar/pull/11263#discussion_r667031125
##########
File path: pulsar-functions/runtime-all/pom.xml
##########
@@ -86,6 +86,15 @@
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-core</artifactId>
</dependency>
+
+ <!-- avro -->
+ <dependency>
+ <groupId>org.apache.avro</groupId>
Review comment:
see https://github.com/apache/pulsar/issues/11274
for this change.
without adding Avro to the Pulsar IO classpath it is not possible to access the Native Object
--
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-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on a change in pull request #11263: Enhanced ElasticSearch Sink (TLS, Auth, KeyValue Schema, INSERT/UPDATE/DELETE, "_id" management, multi index support)
Posted by GitBox <gi...@apache.org>.
eolivelli commented on a change in pull request #11263:
URL: https://github.com/apache/pulsar/pull/11263#discussion_r667031125
##########
File path: pulsar-functions/runtime-all/pom.xml
##########
@@ -86,6 +86,15 @@
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-core</artifactId>
</dependency>
+
+ <!-- avro -->
+ <dependency>
+ <groupId>org.apache.avro</groupId>
Review comment:
see https://github.com/apache/pulsar/issues/11274
for this change.
without adding Avro to the Pulsar IO classpath it is not possible to access the Native Object
--
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-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] nlu90 commented on a change in pull request #11263: Enhanced ElasticSearch Sink (TLS, Auth, KeyValue Schema, INSERT/UPDATE/DELETE, "_id" management, multi index support)
Posted by GitBox <gi...@apache.org>.
nlu90 commented on a change in pull request #11263:
URL: https://github.com/apache/pulsar/pull/11263#discussion_r670808134
##########
File path: pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/JsonConverter.java
##########
@@ -0,0 +1,204 @@
+/**
Review comment:
This file looks very common. Do we want to put them into some kind of `utils` or `common` library?
--
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-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on a change in pull request #11263: Enhanced ElasticSearch Sink (TLS, Auth, KeyValue Schema, INSERT/UPDATE/DELETE, "_id" management, multi index support)
Posted by GitBox <gi...@apache.org>.
eolivelli commented on a change in pull request #11263:
URL: https://github.com/apache/pulsar/pull/11263#discussion_r667031125
##########
File path: pulsar-functions/runtime-all/pom.xml
##########
@@ -86,6 +86,15 @@
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-core</artifactId>
</dependency>
+
+ <!-- avro -->
+ <dependency>
+ <groupId>org.apache.avro</groupId>
Review comment:
see https://github.com/apache/pulsar/issues/11274
for this change.
without adding Avro to the Pulsar IO classpath it is not possible to access the Native Object
--
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-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on pull request #11263: Enhanced ElasticSearch Sink (TLS, Auth, KeyValue Schema, INSERT/UPDATE/DELETE, "_id" management, multi index support)
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on pull request #11263:
URL: https://github.com/apache/pulsar/pull/11263#issuecomment-883095470
@freeznet Please help review this PR if you have time.
--
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-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on pull request #11263: Enhanced ElasticSearch Sink (TLS, Auth, KeyValue Schema, INSERT/UPDATE/DELETE, "_id" management, multi index support)
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on pull request #11263:
URL: https://github.com/apache/pulsar/pull/11263#issuecomment-883095470
@freeznet Please help review this PR if you have time.
--
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-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli merged pull request #11263: Enhanced ElasticSearch Sink (TLS, Auth, KeyValue Schema, INSERT/UPDATE/DELETE, "_id" management, multi index support)
Posted by GitBox <gi...@apache.org>.
eolivelli merged pull request #11263:
URL: https://github.com/apache/pulsar/pull/11263
--
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-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on pull request #11263: Enhanced ElasticSearch Sink (TLS, Auth, KeyValue Schema, INSERT/UPDATE/DELETE, "_id" management, multi index support)
Posted by GitBox <gi...@apache.org>.
eolivelli commented on pull request #11263:
URL: https://github.com/apache/pulsar/pull/11263#issuecomment-887231878
@Anonymitaet I will follow up with a PR for the docs. Stay tuned
--
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-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on a change in pull request #11263: Enhanced ElasticSearch Sink (TLS, Auth, KeyValue Schema, INSERT/UPDATE/DELETE, "_id" management, multi index support)
Posted by GitBox <gi...@apache.org>.
eolivelli commented on a change in pull request #11263:
URL: https://github.com/apache/pulsar/pull/11263#discussion_r667031125
##########
File path: pulsar-functions/runtime-all/pom.xml
##########
@@ -86,6 +86,15 @@
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-core</artifactId>
</dependency>
+
+ <!-- avro -->
+ <dependency>
+ <groupId>org.apache.avro</groupId>
Review comment:
see https://github.com/apache/pulsar/issues/11274
for this change.
without adding Avro to the Pulsar IO classpath it is not possible to access the Native Object
--
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-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on pull request #11263: Enhanced ElasticSearch Sink:
Posted by GitBox <gi...@apache.org>.
eolivelli commented on pull request #11263:
URL: https://github.com/apache/pulsar/pull/11263#issuecomment-877190481
Integration tests are failing due to this bad problem with the classpath, basically the problem is that the AVRO library is not usable and you cannot manipulate the result of GenericRecord.getNativeObject()
https://github.com/apache/pulsar/issues/11274
--
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-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] nlu90 commented on a change in pull request #11263: Enhanced ElasticSearch Sink (TLS, Auth, KeyValue Schema, INSERT/UPDATE/DELETE, "_id" management, multi index support)
Posted by GitBox <gi...@apache.org>.
nlu90 commented on a change in pull request #11263:
URL: https://github.com/apache/pulsar/pull/11263#discussion_r672699904
##########
File path: pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/JsonConverter.java
##########
@@ -0,0 +1,204 @@
+/**
Review comment:
Makes
##########
File path: pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/JsonConverter.java
##########
@@ -0,0 +1,204 @@
+/**
Review comment:
Make sense to me
--
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-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Anonymitaet edited a comment on pull request #11263: Enhanced ElasticSearch Sink (TLS, Auth, KeyValue Schema, INSERT/UPDATE/DELETE, "_id" management, multi index support)
Posted by GitBox <gi...@apache.org>.
Anonymitaet edited a comment on pull request #11263:
URL: https://github.com/apache/pulsar/pull/11263#issuecomment-899907330
Doc is done here: https://github.com/apache/pulsar/pull/11602
--
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-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on pull request #11263: Enhanced ElasticSearch Sink (TLS, Auth, KeyValue Schema, INSERT/UPDATE/DELETE, "_id" management, multi index support)
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on pull request #11263:
URL: https://github.com/apache/pulsar/pull/11263#issuecomment-883095470
@freeznet Please help review this PR if you have time.
--
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-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] nlu90 commented on a change in pull request #11263: Enhanced ElasticSearch Sink (TLS, Auth, KeyValue Schema, INSERT/UPDATE/DELETE, "_id" management, multi index support)
Posted by GitBox <gi...@apache.org>.
nlu90 commented on a change in pull request #11263:
URL: https://github.com/apache/pulsar/pull/11263#discussion_r672699904
##########
File path: pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/JsonConverter.java
##########
@@ -0,0 +1,204 @@
+/**
Review comment:
Makes
--
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-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Anonymitaet commented on pull request #11263: Enhanced ElasticSearch Sink:
Posted by GitBox <gi...@apache.org>.
Anonymitaet commented on pull request #11263:
URL: https://github.com/apache/pulsar/pull/11263#issuecomment-876880172
@eolivelli thanks for taking the doc into consideration, feel free to ping me if you need a doc review.
--
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-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on a change in pull request #11263: Enhanced ElasticSearch Sink (TLS, Auth, KeyValue Schema, INSERT/UPDATE/DELETE, "_id" management, multi index support)
Posted by GitBox <gi...@apache.org>.
eolivelli commented on a change in pull request #11263:
URL: https://github.com/apache/pulsar/pull/11263#discussion_r671131506
##########
File path: pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/JsonConverter.java
##########
@@ -0,0 +1,204 @@
+/**
Review comment:
@nlu90
Yes, this class may have some value and could be moved to utils/common.
But the ES Sink is a separate Pulsar IO Connector.
if we use a class from Pulsar `utils` or `commons` package it means that that class is part of the Public API of Pulsar IO.
I believe that we could move this class to some shared public package in the future, when we can see other users of it.
Does it make sense to you ?
--
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-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Anonymitaet commented on pull request #11263: Enhanced ElasticSearch Sink (TLS, Auth, KeyValue Schema, INSERT/UPDATE/DELETE, "_id" management, multi index support)
Posted by GitBox <gi...@apache.org>.
Anonymitaet commented on pull request #11263:
URL: https://github.com/apache/pulsar/pull/11263#issuecomment-899907330
Doc: https://github.com/apache/pulsar/pull/11602
--
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-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] nlu90 commented on a change in pull request #11263: Enhanced ElasticSearch Sink (TLS, Auth, KeyValue Schema, INSERT/UPDATE/DELETE, "_id" management, multi index support)
Posted by GitBox <gi...@apache.org>.
nlu90 commented on a change in pull request #11263:
URL: https://github.com/apache/pulsar/pull/11263#discussion_r672699904
##########
File path: pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/JsonConverter.java
##########
@@ -0,0 +1,204 @@
+/**
Review comment:
Makes
##########
File path: pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/JsonConverter.java
##########
@@ -0,0 +1,204 @@
+/**
Review comment:
Make sense to me
--
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-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on a change in pull request #11263: Enhanced ElasticSearch Sink (TLS, Auth, KeyValue Schema, INSERT/UPDATE/DELETE, "_id" management, multi index support)
Posted by GitBox <gi...@apache.org>.
eolivelli commented on a change in pull request #11263:
URL: https://github.com/apache/pulsar/pull/11263#discussion_r673067706
##########
File path: pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClient.java
##########
@@ -0,0 +1,569 @@
+/**
+ * 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.pulsar.io.elasticsearch;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Strings;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.http.HttpHost;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.client.CredentialsProvider;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.config.Registry;
+import org.apache.http.config.RegistryBuilder;
+import org.apache.http.conn.ssl.NoopHostnameVerifier;
+import org.apache.http.conn.ssl.SSLConnectionSocketFactory;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
+import org.apache.http.impl.nio.conn.PoolingNHttpClientConnectionManager;
+import org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor;
+import org.apache.http.impl.nio.reactor.IOReactorConfig;
+import org.apache.http.nio.conn.NHttpClientConnectionManager;
+import org.apache.http.nio.conn.NoopIOSessionStrategy;
+import org.apache.http.nio.conn.SchemeIOSessionStrategy;
+import org.apache.http.nio.conn.ssl.SSLIOSessionStrategy;
+import org.apache.http.nio.reactor.ConnectingIOReactor;
+import org.apache.http.ssl.SSLContextBuilder;
+import org.apache.http.ssl.SSLContexts;
+import org.apache.pulsar.client.api.schema.GenericObject;
+import org.apache.pulsar.functions.api.Record;
+import org.elasticsearch.action.DocWriteRequest;
+import org.elasticsearch.action.DocWriteResponse;
+import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
+import org.elasticsearch.action.bulk.BulkItemResponse;
+import org.elasticsearch.action.bulk.BulkProcessor;
+import org.elasticsearch.action.bulk.BulkRequest;
+import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.delete.DeleteRequest;
+import org.elasticsearch.action.delete.DeleteResponse;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.index.IndexResponse;
+import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.client.*;
Review comment:
sure. fixed
##########
File path: pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java
##########
@@ -18,143 +18,273 @@
*/
package org.apache.pulsar.io.elasticsearch;
-import java.io.IOException;
-import java.net.MalformedURLException;
-import java.net.URL;
-import java.util.Map;
-
-import org.apache.commons.lang3.StringUtils;
-import org.apache.http.HttpHost;
-import org.apache.http.auth.AuthScope;
-import org.apache.http.auth.UsernamePasswordCredentials;
-import org.apache.http.client.CredentialsProvider;
-import org.apache.http.impl.client.BasicCredentialsProvider;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Strings;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.schema.GenericObject;
+import org.apache.pulsar.client.api.schema.GenericRecord;
+import org.apache.pulsar.client.api.schema.KeyValueSchema;
+import org.apache.pulsar.common.schema.KeyValue;
+import org.apache.pulsar.common.schema.SchemaType;
import org.apache.pulsar.functions.api.Record;
-import org.apache.pulsar.io.core.KeyValue;
import org.apache.pulsar.io.core.Sink;
import org.apache.pulsar.io.core.SinkContext;
import org.apache.pulsar.io.core.annotations.Connector;
import org.apache.pulsar.io.core.annotations.IOType;
-import org.elasticsearch.action.DocWriteResponse;
-import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
-import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
-import org.elasticsearch.action.admin.indices.get.GetIndexRequest;
-import org.elasticsearch.action.index.IndexRequest;
-import org.elasticsearch.action.index.IndexResponse;
-import org.elasticsearch.client.RequestOptions;
-import org.elasticsearch.client.Requests;
-import org.elasticsearch.client.RestClient;
-import org.elasticsearch.client.RestClientBuilder;
-import org.elasticsearch.client.RestHighLevelClient;
-import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.xcontent.XContentType;
-/**
- * The base abstract class for ElasticSearch sinks.
- * Users need to implement extractKeyValue function to use this sink.
- * This class assumes that the input will be JSON documents
- */
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
@Connector(
- name = "elastic_search",
- type = IOType.SINK,
- help = "A sink connector that sends pulsar messages to elastic search",
- configClass = ElasticSearchConfig.class
+ name = "elastic_search",
+ type = IOType.SINK,
+ help = "A sink connector that sends pulsar messages to elastic search",
+ configClass = ElasticSearchConfig.class
)
-public class ElasticSearchSink implements Sink<byte[]> {
+@Slf4j
+public class ElasticSearchSink implements Sink<GenericObject> {
- private URL url;
- private RestHighLevelClient client;
- private CredentialsProvider credentialsProvider;
private ElasticSearchConfig elasticSearchConfig;
+ private ElasticSearchClient elasticsearchClient;
+ private ObjectMapper objectMapper = new ObjectMapper();
+ private List<String> primaryFields = null;
@Override
public void open(Map<String, Object> config, SinkContext sinkContext) throws Exception {
elasticSearchConfig = ElasticSearchConfig.load(config);
elasticSearchConfig.validate();
- createIndexIfNeeded();
+ elasticsearchClient = new ElasticSearchClient(elasticSearchConfig);
+ if (!Strings.isNullOrEmpty(elasticSearchConfig.getPrimaryFields())) {
+ primaryFields = Arrays.asList(elasticSearchConfig.getPrimaryFields().split(","));
+ }
}
@Override
- public void close() throws Exception {
- client.close();
+ public void close() {
+ if (elasticsearchClient != null) {
+ elasticsearchClient.close();
+ elasticsearchClient = null;
+ }
}
@Override
- public void write(Record<byte[]> record) {
- KeyValue<String, byte[]> keyValue = extractKeyValue(record);
- IndexRequest indexRequest = Requests.indexRequest(elasticSearchConfig.getIndexName());
- indexRequest.type(elasticSearchConfig.getTypeName());
- indexRequest.source(keyValue.getValue(), XContentType.JSON);
-
- try {
- IndexResponse indexResponse = getClient().index(indexRequest, RequestOptions.DEFAULT);
- if (indexResponse.getResult().equals(DocWriteResponse.Result.CREATED)) {
- record.ack();
- } else {
- record.fail();
+ public void write(Record<GenericObject> record) throws Exception {
+ if (!elasticsearchClient.isFailed()) {
+ Pair<String, String> idAndDoc = extractIdAndDocument(record);
+ try {
+ if (log.isDebugEnabled()) {
+ log.debug("index doc {} {}", idAndDoc.getLeft(), idAndDoc.getRight());
+ }
+ if (idAndDoc.getRight() == null) {
+ switch (elasticSearchConfig.getNullValueAction()) {
+ case DELETE:
+ if (idAndDoc.getLeft() != null) {
+ if (elasticSearchConfig.isBulkEnabled()) {
+ elasticsearchClient.bulkDelete(record, idAndDoc.getLeft());
+ } else {
+ elasticsearchClient.deleteDocument(record, idAndDoc.getLeft());
+ }
+ }
+ break;
+ case IGNORE:
+ break;
+ case FAIL:
+ elasticsearchClient.failed(new PulsarClientException.InvalidMessageException("Unexpected null message value"));
Review comment:
we are throwing `elasticsearchClient.irrecoverableError.get();`
isn't this supposed to be managed as a "record.fail()" ?
##########
File path: pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java
##########
@@ -18,143 +18,273 @@
*/
package org.apache.pulsar.io.elasticsearch;
-import java.io.IOException;
-import java.net.MalformedURLException;
-import java.net.URL;
-import java.util.Map;
-
-import org.apache.commons.lang3.StringUtils;
-import org.apache.http.HttpHost;
-import org.apache.http.auth.AuthScope;
-import org.apache.http.auth.UsernamePasswordCredentials;
-import org.apache.http.client.CredentialsProvider;
-import org.apache.http.impl.client.BasicCredentialsProvider;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Strings;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.schema.GenericObject;
+import org.apache.pulsar.client.api.schema.GenericRecord;
+import org.apache.pulsar.client.api.schema.KeyValueSchema;
+import org.apache.pulsar.common.schema.KeyValue;
+import org.apache.pulsar.common.schema.SchemaType;
import org.apache.pulsar.functions.api.Record;
-import org.apache.pulsar.io.core.KeyValue;
import org.apache.pulsar.io.core.Sink;
import org.apache.pulsar.io.core.SinkContext;
import org.apache.pulsar.io.core.annotations.Connector;
import org.apache.pulsar.io.core.annotations.IOType;
-import org.elasticsearch.action.DocWriteResponse;
-import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
-import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
-import org.elasticsearch.action.admin.indices.get.GetIndexRequest;
-import org.elasticsearch.action.index.IndexRequest;
-import org.elasticsearch.action.index.IndexResponse;
-import org.elasticsearch.client.RequestOptions;
-import org.elasticsearch.client.Requests;
-import org.elasticsearch.client.RestClient;
-import org.elasticsearch.client.RestClientBuilder;
-import org.elasticsearch.client.RestHighLevelClient;
-import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.xcontent.XContentType;
-/**
- * The base abstract class for ElasticSearch sinks.
- * Users need to implement extractKeyValue function to use this sink.
- * This class assumes that the input will be JSON documents
- */
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
@Connector(
- name = "elastic_search",
- type = IOType.SINK,
- help = "A sink connector that sends pulsar messages to elastic search",
- configClass = ElasticSearchConfig.class
+ name = "elastic_search",
+ type = IOType.SINK,
+ help = "A sink connector that sends pulsar messages to elastic search",
+ configClass = ElasticSearchConfig.class
)
-public class ElasticSearchSink implements Sink<byte[]> {
+@Slf4j
+public class ElasticSearchSink implements Sink<GenericObject> {
- private URL url;
- private RestHighLevelClient client;
- private CredentialsProvider credentialsProvider;
private ElasticSearchConfig elasticSearchConfig;
+ private ElasticSearchClient elasticsearchClient;
+ private ObjectMapper objectMapper = new ObjectMapper();
+ private List<String> primaryFields = null;
@Override
public void open(Map<String, Object> config, SinkContext sinkContext) throws Exception {
elasticSearchConfig = ElasticSearchConfig.load(config);
elasticSearchConfig.validate();
- createIndexIfNeeded();
+ elasticsearchClient = new ElasticSearchClient(elasticSearchConfig);
+ if (!Strings.isNullOrEmpty(elasticSearchConfig.getPrimaryFields())) {
+ primaryFields = Arrays.asList(elasticSearchConfig.getPrimaryFields().split(","));
+ }
}
@Override
- public void close() throws Exception {
- client.close();
+ public void close() {
+ if (elasticsearchClient != null) {
+ elasticsearchClient.close();
+ elasticsearchClient = null;
+ }
}
@Override
- public void write(Record<byte[]> record) {
- KeyValue<String, byte[]> keyValue = extractKeyValue(record);
- IndexRequest indexRequest = Requests.indexRequest(elasticSearchConfig.getIndexName());
- indexRequest.type(elasticSearchConfig.getTypeName());
- indexRequest.source(keyValue.getValue(), XContentType.JSON);
-
- try {
- IndexResponse indexResponse = getClient().index(indexRequest, RequestOptions.DEFAULT);
- if (indexResponse.getResult().equals(DocWriteResponse.Result.CREATED)) {
- record.ack();
- } else {
- record.fail();
+ public void write(Record<GenericObject> record) throws Exception {
+ if (!elasticsearchClient.isFailed()) {
+ Pair<String, String> idAndDoc = extractIdAndDocument(record);
+ try {
+ if (log.isDebugEnabled()) {
+ log.debug("index doc {} {}", idAndDoc.getLeft(), idAndDoc.getRight());
+ }
+ if (idAndDoc.getRight() == null) {
+ switch (elasticSearchConfig.getNullValueAction()) {
+ case DELETE:
+ if (idAndDoc.getLeft() != null) {
+ if (elasticSearchConfig.isBulkEnabled()) {
+ elasticsearchClient.bulkDelete(record, idAndDoc.getLeft());
+ } else {
+ elasticsearchClient.deleteDocument(record, idAndDoc.getLeft());
+ }
+ }
+ break;
+ case IGNORE:
+ break;
+ case FAIL:
+ elasticsearchClient.failed(new PulsarClientException.InvalidMessageException("Unexpected null message value"));
+ throw elasticsearchClient.irrecoverableError.get();
+ }
+ } else {
+ if (elasticSearchConfig.isBulkEnabled()) {
+ elasticsearchClient.bulkIndex(record, idAndDoc);
+ } else {
+ elasticsearchClient.indexDocument(record, idAndDoc);
+ }
+ }
+ } catch (JsonProcessingException jsonProcessingException) {
+ switch (elasticSearchConfig.getMalformedDocAction()) {
+ case IGNORE:
+ break;
+ case WARN:
+ log.warn("Ignoring malformed document messageId={}",
+ record.getMessage().map(Message::getMessageId).orElse(null),
+ jsonProcessingException);
+ elasticsearchClient.failed(jsonProcessingException);
+ throw jsonProcessingException;
+ case FAIL:
+ log.error("Malformed document messageId={}",
+ record.getMessage().map(Message::getMessageId).orElse(null),
+ jsonProcessingException);
+ elasticsearchClient.failed(jsonProcessingException);
+ throw jsonProcessingException;
+ }
+ } catch (Exception e) {
+ log.error("write error for {} {}:", idAndDoc.getLeft(), idAndDoc.getRight(), e);
+ throw e;
}
- } catch (final IOException ex) {
- record.fail();
+ } else {
+ throw new IllegalStateException("Elasticsearch client is in FAILED status");
}
}
- public KeyValue<String, byte[]> extractKeyValue(Record<byte[]> record) {
- String key = record.getKey().orElse("");
- return new KeyValue<>(key, record.getValue());
+ @VisibleForTesting
+ ElasticSearchClient getElasticsearchClient() {
+ return this.elasticsearchClient;
}
- private void createIndexIfNeeded() throws IOException {
- GetIndexRequest request = new GetIndexRequest();
- request.indices(elasticSearchConfig.getIndexName());
- boolean exists = getClient().indices().exists(request, RequestOptions.DEFAULT);
+ /**
+ * Extract ES _id and _source using the Schema if available.
+ *
+ * @param record
+ * @return A pair for _id and _source
+ */
+ public Pair<String, String> extractIdAndDocument(Record<GenericObject> record) throws JsonProcessingException {
+ if (elasticSearchConfig.isSchemaEnable()) {
+ Object key = null;
+ GenericObject value = null;
+ Schema<?> keySchema = null;
+ Schema<?> valueSchema = null;
+ if (record.getSchema() != null && record.getSchema() instanceof KeyValueSchema) {
+ KeyValueSchema<GenericObject, GenericObject> keyValueSchema = (KeyValueSchema) record.getSchema();
+ keySchema = keyValueSchema.getKeySchema();
+ valueSchema = keyValueSchema.getValueSchema();
+ KeyValue<GenericObject, GenericObject> keyValue = (KeyValue<GenericObject, GenericObject>) record.getValue().getNativeObject();
+ key = keyValue.getKey();
+ value = keyValue.getValue();
+ } else {
+ key = record.getKey().orElse(null);
+ valueSchema = record.getSchema();
+ value = record.getValue();
+ }
+
+ String id = null;
+ if (elasticSearchConfig.isKeyIgnore() == false && key != null && keySchema != null) {
+ id = stringifyKey(keySchema, key);
+ }
- if (!exists) {
- CreateIndexRequest cireq = new CreateIndexRequest(elasticSearchConfig.getIndexName());
+ String doc = null;
+ if (value != null) {
+ if (valueSchema != null) {
+ doc = stringifyValue(valueSchema, value);
+ } else {
+ if (value.getNativeObject() instanceof byte[]) {
+ // for BWC with the ES-Sink
+ doc = new String((byte[]) value.getNativeObject(), StandardCharsets.UTF_8);
+ } else {
+ doc = value.getNativeObject().toString();
+ }
+ }
+ }
- cireq.settings(Settings.builder()
- .put("index.number_of_shards", elasticSearchConfig.getIndexNumberOfShards())
- .put("index.number_of_replicas", elasticSearchConfig.getIndexNumberOfReplicas()));
+ if (doc != null && primaryFields != null) {
+ try {
+ // extract the PK from the JSON document
+ JsonNode jsonNode = objectMapper.readTree(doc);
+ id = stringifyKey(jsonNode, primaryFields);
+ } catch (JsonProcessingException e) {
+ log.error("Failed to read JSON", e);
+ throw e;
+ }
+ }
- CreateIndexResponse ciresp = getClient().indices().create(cireq, RequestOptions.DEFAULT);
- if (!ciresp.isAcknowledged() || !ciresp.isShardsAcknowledged()) {
- throw new RuntimeException("Unable to create index.");
+ if (log.isDebugEnabled()) {
+ SchemaType schemaType = null;
+ if (record.getSchema() != null && record.getSchema().getSchemaInfo() != null) {
+ schemaType = record.getSchema().getSchemaInfo().getType();
+ }
+ log.debug("recordType={} schemaType={} id={} doc={}",
+ record.getClass().getName(),
+ schemaType,
+ id,
+ doc);
}
+ return Pair.of(id, doc);
+ } else {
+ return Pair.of(null, new String(
Review comment:
we must keep the `_id` null here, because with `schemaEnable = false` the Sink behaves exactly the same way as in Pulsar 2.8.0, that is without handling the primary key of the ES document.
##########
File path: pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClient.java
##########
@@ -0,0 +1,569 @@
+/**
+ * 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.pulsar.io.elasticsearch;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Strings;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.http.HttpHost;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.client.CredentialsProvider;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.config.Registry;
+import org.apache.http.config.RegistryBuilder;
+import org.apache.http.conn.ssl.NoopHostnameVerifier;
+import org.apache.http.conn.ssl.SSLConnectionSocketFactory;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
+import org.apache.http.impl.nio.conn.PoolingNHttpClientConnectionManager;
+import org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor;
+import org.apache.http.impl.nio.reactor.IOReactorConfig;
+import org.apache.http.nio.conn.NHttpClientConnectionManager;
+import org.apache.http.nio.conn.NoopIOSessionStrategy;
+import org.apache.http.nio.conn.SchemeIOSessionStrategy;
+import org.apache.http.nio.conn.ssl.SSLIOSessionStrategy;
+import org.apache.http.nio.reactor.ConnectingIOReactor;
+import org.apache.http.ssl.SSLContextBuilder;
+import org.apache.http.ssl.SSLContexts;
+import org.apache.pulsar.client.api.schema.GenericObject;
+import org.apache.pulsar.functions.api.Record;
+import org.elasticsearch.action.DocWriteRequest;
+import org.elasticsearch.action.DocWriteResponse;
+import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
+import org.elasticsearch.action.bulk.BulkItemResponse;
+import org.elasticsearch.action.bulk.BulkProcessor;
+import org.elasticsearch.action.bulk.BulkRequest;
+import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.delete.DeleteRequest;
+import org.elasticsearch.action.delete.DeleteResponse;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.index.IndexResponse;
+import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.client.*;
+import org.elasticsearch.client.indices.CreateIndexRequest;
+import org.elasticsearch.client.indices.CreateIndexResponse;
+import org.elasticsearch.client.indices.GetIndexRequest;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.ByteSizeUnit;
+import org.elasticsearch.common.unit.ByteSizeValue;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.common.xcontent.XContentType;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.search.SearchHit;
+import org.elasticsearch.search.builder.SearchSourceBuilder;
+
+import javax.net.ssl.HostnameVerifier;
+import javax.net.ssl.SSLContext;
+import java.io.File;
+import java.io.IOException;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.security.KeyManagementException;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
+import java.security.UnrecoverableKeyException;
+import java.security.cert.CertificateException;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicReference;
+
+@Slf4j
+public class ElasticSearchClient {
+
+ static final String[] malformedErrors = {
+ "mapper_parsing_exception",
+ "action_request_validation_exception",
+ "illegal_argument_exception"
+ };
+
+ private ElasticSearchConfig config;
+ private ConfigCallback configCallback;
+ private RestHighLevelClient client;
+
+ final Set<String> indexCache = new HashSet<>();
+ final Map<String, String> topicToIndexCache = new HashMap<>();
+
+ final RandomExponentialRetry backoffRetry;
+ final BulkProcessor bulkProcessor;
+ final ConcurrentMap<DocWriteRequest<?>, Record> records = new ConcurrentHashMap<>();
+ final AtomicReference<Exception> irrecoverableError = new AtomicReference<>();
+ final ScheduledExecutorService executorService;
+
+ ElasticSearchClient(ElasticSearchConfig elasticSearchConfig) throws MalformedURLException {
+ this.config = elasticSearchConfig;
+ this.configCallback = new ConfigCallback();
+ this.backoffRetry = new RandomExponentialRetry(elasticSearchConfig.getMaxRetryTimeInSec());
+ if (config.isBulkEnabled() == false) {
+ bulkProcessor = null;
+ } else {
+ BulkProcessor.Builder builder = BulkProcessor.builder(
+ (bulkRequest, bulkResponseActionListener) -> client.bulkAsync(bulkRequest, RequestOptions.DEFAULT, bulkResponseActionListener),
+ new BulkProcessor.Listener() {
+ @Override
+ public void beforeBulk(long l, BulkRequest bulkRequest) {
+ }
+
+ @Override
+ public void afterBulk(long l, BulkRequest bulkRequest, BulkResponse bulkResponse) {
+ log.trace("Bulk request id={} size={}:", l, bulkRequest.requests().size());
+ for (int i = 0; i < bulkResponse.getItems().length; i++) {
+ DocWriteRequest<?> request = bulkRequest.requests().get(i);
+ Record record = records.get(request);
+ BulkItemResponse bulkItemResponse = bulkResponse.getItems()[i];
+ if (bulkItemResponse.isFailed()) {
+ record.fail();
+ try {
+ hasIrrecoverableError(bulkItemResponse);
+ } catch(Exception e) {
+ log.warn("Unrecoverable error:", e);
+ }
+ } else {
+ record.ack();
+ }
+ records.remove(request);
+ }
+ }
+
+ @Override
+ public void afterBulk(long l, BulkRequest bulkRequest, Throwable throwable) {
+ log.warn("Bulk request id={} failed:", l, throwable);
+ for (DocWriteRequest<?> request : bulkRequest.requests()) {
+ Record record = records.remove(request);
+ record.fail();
+ }
+ }
+ }
+ )
+ .setBulkActions(config.getBulkActions())
+ .setBulkSize(new ByteSizeValue(config.getBulkSizeInMb(), ByteSizeUnit.MB))
+ .setConcurrentRequests(config.getBulkConcurrentRequests())
+ .setBackoffPolicy(new RandomExponentialBackoffPolicy(backoffRetry,
+ config.getRetryBackoffInMs(),
+ config.getMaxRetries()
+ ));
+ if (config.getBulkFlushIntervalInMs() > 0) {
+ builder.setFlushInterval(new TimeValue(config.getBulkFlushIntervalInMs(), TimeUnit.MILLISECONDS));
+ }
+ this.bulkProcessor = builder.build();
+ }
+
+ // idle+expired connection evictor thread
+ this.executorService = Executors.newSingleThreadScheduledExecutor();
+ this.executorService.scheduleAtFixedRate(new Runnable() {
+ @Override
+ public void run() {
+ configCallback.connectionManager.closeExpiredConnections();
+ configCallback.connectionManager.closeIdleConnections(
+ config.getConnectionIdleTimeoutInMs(), TimeUnit.MILLISECONDS);
+ }
+ },
+ config.getConnectionIdleTimeoutInMs(),
+ config.getConnectionIdleTimeoutInMs(),
+ TimeUnit.MILLISECONDS
+ );
+
+ URL url = new URL(config.getElasticSearchUrl());
+ log.info("ElasticSearch URL {}", url);
+ RestClientBuilder builder = RestClient.builder(new HttpHost(url.getHost(), url.getPort(), url.getProtocol()))
+ .setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() {
+ @Override
+ public RequestConfig.Builder customizeRequestConfig(RequestConfig.Builder builder) {
+ return builder
+ .setContentCompressionEnabled(config.isCompressionEnabled())
+ .setConnectionRequestTimeout(config.getConnectionRequestTimeoutInMs())
+ .setConnectTimeout(config.getConnectTimeoutInMs())
+ .setSocketTimeout(config.getSocketTimeoutInMs());
+ }
+ })
+ .setHttpClientConfigCallback(this.configCallback)
+ .setFailureListener(new RestClient.FailureListener() {
+ public void onFailure(Node node) {
+ log.warn("Node host={} failed", node.getHost());
+ }
+ });
+ this.client = new RestHighLevelClient(builder);
+ }
+
+ void failed(Exception e) throws Exception {
+ if (irrecoverableError.compareAndSet(null, e)) {
+ log.error("Irrecoverable error:", e);
+ }
+ }
+
+ boolean isFailed() {
+ return irrecoverableError.get() != null;
+ }
+
+ void hasIrrecoverableError(BulkItemResponse bulkItemResponse) throws Exception {
+ for (String error : malformedErrors) {
+ if (bulkItemResponse.getFailureMessage().contains(error)) {
+ switch (config.getMalformedDocAction()) {
+ case IGNORE:
+ break;
+ case WARN:
+ log.warn("Ignoring malformed document index={} id={}",
+ bulkItemResponse.getIndex(),
+ bulkItemResponse.getId(),
+ bulkItemResponse.getFailure().getCause());
+ break;
+ case FAIL:
+ log.error("Failure due to the malformed document index={} id={}",
+ bulkItemResponse.getIndex(),
+ bulkItemResponse.getId(),
+ bulkItemResponse.getFailure().getCause());
+ failed(bulkItemResponse.getFailure().getCause());
+ break;
+ }
+ }
+ }
+ }
+
+ public void bulkIndex(Record record, Pair<String, String> idAndDoc) throws Exception {
+ try {
+ checkNotFailed();
+ checkIndexExists(record.getTopicName());
+ IndexRequest indexRequest = Requests.indexRequest(config.getIndexName());
+ if (!Strings.isNullOrEmpty(idAndDoc.getLeft()))
+ indexRequest.id(idAndDoc.getLeft());
+ indexRequest.type(config.getTypeName());
+ indexRequest.source(idAndDoc.getRight(), XContentType.JSON);
+
+ records.put(indexRequest, record);
+ bulkProcessor.add(indexRequest);
+ } catch(Exception e) {
+ log.debug("index failed id=" + idAndDoc.getLeft(), e);
+ record.fail();
+ throw e;
+ }
+ }
+
+ /**
+ * Index an elasticsearch document and ack the record.
+ * @param record
+ * @param idAndDoc
+ * @return
+ * @throws Exception
+ */
+ public boolean indexDocument(Record<GenericObject> record, Pair<String, String> idAndDoc) throws Exception {
+ try {
+ checkNotFailed();
+ checkIndexExists(record.getTopicName());
+ IndexRequest indexRequest = Requests.indexRequest(config.getIndexName());
+ if (!Strings.isNullOrEmpty(idAndDoc.getLeft()))
+ indexRequest.id(idAndDoc.getLeft());
+ indexRequest.type(config.getTypeName());
+ indexRequest.source(idAndDoc.getRight(), XContentType.JSON);
+ IndexResponse indexResponse = client.index(indexRequest, RequestOptions.DEFAULT);
+ if (indexResponse.getResult().equals(DocWriteResponse.Result.CREATED) ||
+ indexResponse.getResult().equals(DocWriteResponse.Result.UPDATED)) {
+ record.ack();
+ return true;
+ } else {
+ record.fail();
+ return false;
+ }
+ } catch (final Exception ex) {
+ log.error("index failed id=" + idAndDoc.getLeft(), ex);
+ record.fail();
+ throw ex;
+ }
+ }
+
+ public void bulkDelete(Record<GenericObject> record, String id) throws Exception {
+ try {
+ checkNotFailed();
+ checkIndexExists(record.getTopicName());
+ DeleteRequest deleteRequest = Requests.deleteRequest(config.getIndexName());
+ deleteRequest.id(id);
+ deleteRequest.type(config.getTypeName());
+
+ records.put(deleteRequest, record);
+ bulkProcessor.add(deleteRequest);
+ } catch(Exception e) {
+ log.debug("delete failed id=" + id, e);
+ record.fail();
+ throw e;
+ }
+ }
+
+ /**
+ * Delete an elasticsearch document and ack the record.
+ * @param record
+ * @param id
+ * @return
+ * @throws IOException
+ */
+ public boolean deleteDocument(Record<GenericObject> record, String id) throws Exception {
+ try {
+ checkNotFailed();
+ checkIndexExists(record.getTopicName());
+ DeleteRequest deleteRequest = Requests.deleteRequest(config.getIndexName());
+ deleteRequest.id(id);
+ deleteRequest.type(config.getTypeName());
+ DeleteResponse deleteResponse = client.delete(deleteRequest, RequestOptions.DEFAULT);
+ log.debug("delete result=" + deleteResponse.getResult());
+ if (deleteResponse.getResult().equals(DocWriteResponse.Result.DELETED) ||
+ deleteResponse.getResult().equals(DocWriteResponse.Result.NOT_FOUND)) {
+ record.ack();
+ return true;
+ }
+ record.fail();
+ return false;
+ } catch (final Exception ex) {
+ log.debug("index failed id=" + id, ex);
+ record.fail();
+ throw ex;
+ }
+ }
+
+ /**
+ * Flushes the bulk processor.
+ */
+ public void flush() {
+ bulkProcessor.flush();
+ }
+
+ public void close() {
+ try {
+ if (bulkProcessor != null) {
+ bulkProcessor.awaitClose(5000L, TimeUnit.MILLISECONDS);
+ }
+ } catch (InterruptedException e) {
+ log.warn("Elasticsearch bulk processor close error:", e);
+ }
+ try {
+ this.executorService.shutdown();
+ if (this.client != null) {
+ this.client.close();
+ }
+ } catch (IOException e) {
+ log.warn("Elasticsearch client close error:", e);
+ }
+ }
+
+ private void checkNotFailed() throws Exception {
+ if (irrecoverableError.get() != null) {
+ throw irrecoverableError.get();
+ }
+ }
+
+ private void checkIndexExists(Optional<String> topicName) throws IOException {
+ if (!config.isCreateIndexIfNeeded()) {
+ return;
+ }
+ String indexName = indexName(topicName);
+ if (!indexCache.contains(indexName)) {
+ synchronized (this) {
+ if (!indexCache.contains(indexName)) {
+ createIndexIfNeeded(indexName);
+ indexCache.add(indexName);
+ }
+ }
+ }
+ }
+
+ private String indexName(Optional<String> topicName) throws IOException {
+ if (config.getIndexName() != null) {
+ // Use the configured indexName if provided.
+ return config.getIndexName();
+ }
+ if (!topicName.isPresent()) {
+ throw new IOException("Elasticsearch index name configuration and topic name are empty");
+ }
+ return topicToIndexName(topicName.get());
+ }
+
+ @VisibleForTesting
+ public String topicToIndexName(String topicName) {
+ return topicToIndexCache.computeIfAbsent(topicName, k -> {
+ // see elasticsearch limitations https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-create-index.html#indices-create-api-path-params
+ String indexName = topicName.toLowerCase(Locale.ROOT);
+
+ // remove the pulsar topic info persistent://tenant/namespace/topic
+ String[] parts = indexName.split("/");
+ if (parts.length > 1) {
+ indexName = parts[parts.length-1];
+ }
+
+ // truncate to the max bytes length
+ while (indexName.getBytes(StandardCharsets.UTF_8).length > 255) {
+ indexName = indexName.substring(0, indexName.length() - 1);
+ }
+ if (indexName.length() <= 0 || !indexName.matches("[a-zA-Z\\.0-9][a-zA-Z_\\.\\-\\+0-9]*")) {
+ throw new RuntimeException(new IOException("Cannot convert the topic name='" + topicName + "' to a valid elasticsearch index name"));
+ }
+ if (log.isDebugEnabled()) {
+ log.debug("Translate topic={} to index={}", k, indexName);
+ }
+ return indexName;
+ });
+ }
+
+ @VisibleForTesting
+ public boolean createIndexIfNeeded(String indexName) throws IOException {
+ if (indexExists(indexName)) {
+ return false;
+ }
+ final CreateIndexRequest cireq = new CreateIndexRequest(indexName);
+ cireq.settings(Settings.builder()
+ .put("index.number_of_shards", config.getIndexNumberOfShards())
+ .put("index.number_of_replicas", config.getIndexNumberOfReplicas()));
+ return retry(() -> {
+ CreateIndexResponse resp = client.indices().create(cireq, RequestOptions.DEFAULT);
+ if (!resp.isAcknowledged() || !resp.isShardsAcknowledged()) {
+ throw new IOException("Unable to create index.");
+ }
+ return true;
+ }, "create index");
+ }
+
+ public boolean indexExists(final String indexName) throws IOException {
+ final GetIndexRequest request = new GetIndexRequest(indexName);
+ return retry(() -> client.indices().exists(request, RequestOptions.DEFAULT), "index exists");
+ }
+
+ @VisibleForTesting
+ protected long totalHits(String indexName) throws IOException {
+ client.indices().refresh(new RefreshRequest(indexName), RequestOptions.DEFAULT);
+ SearchResponse response = client.search(
+ new SearchRequest()
+ .indices(indexName)
+ .source(new SearchSourceBuilder().query(QueryBuilders.matchAllQuery())),
+ RequestOptions.DEFAULT);
+ for(SearchHit searchHit : response.getHits()) {
+ System.out.println(searchHit.getId()+": "+searchHit.getFields());
Review comment:
correct (this is only for the tests, but it is better to use the logger consistently)
##########
File path: pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClient.java
##########
@@ -0,0 +1,569 @@
+/**
+ * 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.pulsar.io.elasticsearch;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Strings;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.http.HttpHost;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.client.CredentialsProvider;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.config.Registry;
+import org.apache.http.config.RegistryBuilder;
+import org.apache.http.conn.ssl.NoopHostnameVerifier;
+import org.apache.http.conn.ssl.SSLConnectionSocketFactory;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
+import org.apache.http.impl.nio.conn.PoolingNHttpClientConnectionManager;
+import org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor;
+import org.apache.http.impl.nio.reactor.IOReactorConfig;
+import org.apache.http.nio.conn.NHttpClientConnectionManager;
+import org.apache.http.nio.conn.NoopIOSessionStrategy;
+import org.apache.http.nio.conn.SchemeIOSessionStrategy;
+import org.apache.http.nio.conn.ssl.SSLIOSessionStrategy;
+import org.apache.http.nio.reactor.ConnectingIOReactor;
+import org.apache.http.ssl.SSLContextBuilder;
+import org.apache.http.ssl.SSLContexts;
+import org.apache.pulsar.client.api.schema.GenericObject;
+import org.apache.pulsar.functions.api.Record;
+import org.elasticsearch.action.DocWriteRequest;
+import org.elasticsearch.action.DocWriteResponse;
+import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
+import org.elasticsearch.action.bulk.BulkItemResponse;
+import org.elasticsearch.action.bulk.BulkProcessor;
+import org.elasticsearch.action.bulk.BulkRequest;
+import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.delete.DeleteRequest;
+import org.elasticsearch.action.delete.DeleteResponse;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.index.IndexResponse;
+import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.client.*;
+import org.elasticsearch.client.indices.CreateIndexRequest;
+import org.elasticsearch.client.indices.CreateIndexResponse;
+import org.elasticsearch.client.indices.GetIndexRequest;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.ByteSizeUnit;
+import org.elasticsearch.common.unit.ByteSizeValue;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.common.xcontent.XContentType;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.search.SearchHit;
+import org.elasticsearch.search.builder.SearchSourceBuilder;
+
+import javax.net.ssl.HostnameVerifier;
+import javax.net.ssl.SSLContext;
+import java.io.File;
+import java.io.IOException;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.security.KeyManagementException;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
+import java.security.UnrecoverableKeyException;
+import java.security.cert.CertificateException;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicReference;
+
+@Slf4j
+public class ElasticSearchClient {
+
+ static final String[] malformedErrors = {
+ "mapper_parsing_exception",
+ "action_request_validation_exception",
+ "illegal_argument_exception"
+ };
+
+ private ElasticSearchConfig config;
+ private ConfigCallback configCallback;
+ private RestHighLevelClient client;
+
+ final Set<String> indexCache = new HashSet<>();
+ final Map<String, String> topicToIndexCache = new HashMap<>();
+
+ final RandomExponentialRetry backoffRetry;
+ final BulkProcessor bulkProcessor;
+ final ConcurrentMap<DocWriteRequest<?>, Record> records = new ConcurrentHashMap<>();
+ final AtomicReference<Exception> irrecoverableError = new AtomicReference<>();
+ final ScheduledExecutorService executorService;
+
+ ElasticSearchClient(ElasticSearchConfig elasticSearchConfig) throws MalformedURLException {
+ this.config = elasticSearchConfig;
+ this.configCallback = new ConfigCallback();
+ this.backoffRetry = new RandomExponentialRetry(elasticSearchConfig.getMaxRetryTimeInSec());
+ if (config.isBulkEnabled() == false) {
+ bulkProcessor = null;
+ } else {
+ BulkProcessor.Builder builder = BulkProcessor.builder(
+ (bulkRequest, bulkResponseActionListener) -> client.bulkAsync(bulkRequest, RequestOptions.DEFAULT, bulkResponseActionListener),
+ new BulkProcessor.Listener() {
+ @Override
+ public void beforeBulk(long l, BulkRequest bulkRequest) {
+ }
+
+ @Override
+ public void afterBulk(long l, BulkRequest bulkRequest, BulkResponse bulkResponse) {
+ log.trace("Bulk request id={} size={}:", l, bulkRequest.requests().size());
+ for (int i = 0; i < bulkResponse.getItems().length; i++) {
+ DocWriteRequest<?> request = bulkRequest.requests().get(i);
+ Record record = records.get(request);
+ BulkItemResponse bulkItemResponse = bulkResponse.getItems()[i];
+ if (bulkItemResponse.isFailed()) {
+ record.fail();
+ try {
+ hasIrrecoverableError(bulkItemResponse);
+ } catch(Exception e) {
+ log.warn("Unrecoverable error:", e);
+ }
+ } else {
+ record.ack();
+ }
+ records.remove(request);
+ }
+ }
+
+ @Override
+ public void afterBulk(long l, BulkRequest bulkRequest, Throwable throwable) {
+ log.warn("Bulk request id={} failed:", l, throwable);
+ for (DocWriteRequest<?> request : bulkRequest.requests()) {
+ Record record = records.remove(request);
+ record.fail();
+ }
+ }
+ }
+ )
+ .setBulkActions(config.getBulkActions())
+ .setBulkSize(new ByteSizeValue(config.getBulkSizeInMb(), ByteSizeUnit.MB))
+ .setConcurrentRequests(config.getBulkConcurrentRequests())
+ .setBackoffPolicy(new RandomExponentialBackoffPolicy(backoffRetry,
+ config.getRetryBackoffInMs(),
+ config.getMaxRetries()
+ ));
+ if (config.getBulkFlushIntervalInMs() > 0) {
+ builder.setFlushInterval(new TimeValue(config.getBulkFlushIntervalInMs(), TimeUnit.MILLISECONDS));
+ }
+ this.bulkProcessor = builder.build();
+ }
+
+ // idle+expired connection evictor thread
+ this.executorService = Executors.newSingleThreadScheduledExecutor();
+ this.executorService.scheduleAtFixedRate(new Runnable() {
+ @Override
+ public void run() {
+ configCallback.connectionManager.closeExpiredConnections();
+ configCallback.connectionManager.closeIdleConnections(
+ config.getConnectionIdleTimeoutInMs(), TimeUnit.MILLISECONDS);
+ }
+ },
+ config.getConnectionIdleTimeoutInMs(),
+ config.getConnectionIdleTimeoutInMs(),
+ TimeUnit.MILLISECONDS
+ );
+
+ URL url = new URL(config.getElasticSearchUrl());
+ log.info("ElasticSearch URL {}", url);
+ RestClientBuilder builder = RestClient.builder(new HttpHost(url.getHost(), url.getPort(), url.getProtocol()))
+ .setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() {
+ @Override
+ public RequestConfig.Builder customizeRequestConfig(RequestConfig.Builder builder) {
+ return builder
+ .setContentCompressionEnabled(config.isCompressionEnabled())
+ .setConnectionRequestTimeout(config.getConnectionRequestTimeoutInMs())
+ .setConnectTimeout(config.getConnectTimeoutInMs())
+ .setSocketTimeout(config.getSocketTimeoutInMs());
+ }
+ })
+ .setHttpClientConfigCallback(this.configCallback)
+ .setFailureListener(new RestClient.FailureListener() {
+ public void onFailure(Node node) {
+ log.warn("Node host={} failed", node.getHost());
+ }
+ });
+ this.client = new RestHighLevelClient(builder);
+ }
+
+ void failed(Exception e) throws Exception {
+ if (irrecoverableError.compareAndSet(null, e)) {
+ log.error("Irrecoverable error:", e);
+ }
+ }
+
+ boolean isFailed() {
+ return irrecoverableError.get() != null;
+ }
+
+ void hasIrrecoverableError(BulkItemResponse bulkItemResponse) throws Exception {
+ for (String error : malformedErrors) {
+ if (bulkItemResponse.getFailureMessage().contains(error)) {
+ switch (config.getMalformedDocAction()) {
+ case IGNORE:
+ break;
+ case WARN:
+ log.warn("Ignoring malformed document index={} id={}",
+ bulkItemResponse.getIndex(),
+ bulkItemResponse.getId(),
+ bulkItemResponse.getFailure().getCause());
+ break;
+ case FAIL:
+ log.error("Failure due to the malformed document index={} id={}",
+ bulkItemResponse.getIndex(),
+ bulkItemResponse.getId(),
+ bulkItemResponse.getFailure().getCause());
+ failed(bulkItemResponse.getFailure().getCause());
+ break;
+ }
+ }
+ }
+ }
+
+ public void bulkIndex(Record record, Pair<String, String> idAndDoc) throws Exception {
+ try {
+ checkNotFailed();
+ checkIndexExists(record.getTopicName());
+ IndexRequest indexRequest = Requests.indexRequest(config.getIndexName());
+ if (!Strings.isNullOrEmpty(idAndDoc.getLeft()))
+ indexRequest.id(idAndDoc.getLeft());
+ indexRequest.type(config.getTypeName());
+ indexRequest.source(idAndDoc.getRight(), XContentType.JSON);
+
+ records.put(indexRequest, record);
+ bulkProcessor.add(indexRequest);
+ } catch(Exception e) {
+ log.debug("index failed id=" + idAndDoc.getLeft(), e);
+ record.fail();
+ throw e;
+ }
+ }
+
+ /**
+ * Index an elasticsearch document and ack the record.
+ * @param record
+ * @param idAndDoc
+ * @return
+ * @throws Exception
+ */
+ public boolean indexDocument(Record<GenericObject> record, Pair<String, String> idAndDoc) throws Exception {
+ try {
+ checkNotFailed();
+ checkIndexExists(record.getTopicName());
+ IndexRequest indexRequest = Requests.indexRequest(config.getIndexName());
+ if (!Strings.isNullOrEmpty(idAndDoc.getLeft()))
+ indexRequest.id(idAndDoc.getLeft());
+ indexRequest.type(config.getTypeName());
+ indexRequest.source(idAndDoc.getRight(), XContentType.JSON);
+ IndexResponse indexResponse = client.index(indexRequest, RequestOptions.DEFAULT);
+ if (indexResponse.getResult().equals(DocWriteResponse.Result.CREATED) ||
+ indexResponse.getResult().equals(DocWriteResponse.Result.UPDATED)) {
+ record.ack();
+ return true;
+ } else {
+ record.fail();
+ return false;
+ }
+ } catch (final Exception ex) {
+ log.error("index failed id=" + idAndDoc.getLeft(), ex);
+ record.fail();
+ throw ex;
+ }
+ }
+
+ public void bulkDelete(Record<GenericObject> record, String id) throws Exception {
+ try {
+ checkNotFailed();
+ checkIndexExists(record.getTopicName());
+ DeleteRequest deleteRequest = Requests.deleteRequest(config.getIndexName());
+ deleteRequest.id(id);
+ deleteRequest.type(config.getTypeName());
+
+ records.put(deleteRequest, record);
+ bulkProcessor.add(deleteRequest);
+ } catch(Exception e) {
+ log.debug("delete failed id=" + id, e);
+ record.fail();
+ throw e;
+ }
+ }
+
+ /**
+ * Delete an elasticsearch document and ack the record.
+ * @param record
+ * @param id
+ * @return
+ * @throws IOException
+ */
+ public boolean deleteDocument(Record<GenericObject> record, String id) throws Exception {
+ try {
+ checkNotFailed();
+ checkIndexExists(record.getTopicName());
+ DeleteRequest deleteRequest = Requests.deleteRequest(config.getIndexName());
+ deleteRequest.id(id);
+ deleteRequest.type(config.getTypeName());
+ DeleteResponse deleteResponse = client.delete(deleteRequest, RequestOptions.DEFAULT);
+ log.debug("delete result=" + deleteResponse.getResult());
+ if (deleteResponse.getResult().equals(DocWriteResponse.Result.DELETED) ||
+ deleteResponse.getResult().equals(DocWriteResponse.Result.NOT_FOUND)) {
+ record.ack();
+ return true;
+ }
+ record.fail();
+ return false;
+ } catch (final Exception ex) {
+ log.debug("index failed id=" + id, ex);
+ record.fail();
+ throw ex;
+ }
+ }
+
+ /**
+ * Flushes the bulk processor.
+ */
+ public void flush() {
+ bulkProcessor.flush();
+ }
+
+ public void close() {
+ try {
+ if (bulkProcessor != null) {
+ bulkProcessor.awaitClose(5000L, TimeUnit.MILLISECONDS);
+ }
+ } catch (InterruptedException e) {
+ log.warn("Elasticsearch bulk processor close error:", e);
+ }
+ try {
+ this.executorService.shutdown();
+ if (this.client != null) {
+ this.client.close();
+ }
+ } catch (IOException e) {
+ log.warn("Elasticsearch client close error:", e);
+ }
+ }
+
+ private void checkNotFailed() throws Exception {
+ if (irrecoverableError.get() != null) {
+ throw irrecoverableError.get();
+ }
+ }
+
+ private void checkIndexExists(Optional<String> topicName) throws IOException {
+ if (!config.isCreateIndexIfNeeded()) {
+ return;
+ }
+ String indexName = indexName(topicName);
+ if (!indexCache.contains(indexName)) {
+ synchronized (this) {
+ if (!indexCache.contains(indexName)) {
+ createIndexIfNeeded(indexName);
+ indexCache.add(indexName);
+ }
+ }
+ }
+ }
+
+ private String indexName(Optional<String> topicName) throws IOException {
+ if (config.getIndexName() != null) {
+ // Use the configured indexName if provided.
+ return config.getIndexName();
+ }
+ if (!topicName.isPresent()) {
+ throw new IOException("Elasticsearch index name configuration and topic name are empty");
+ }
+ return topicToIndexName(topicName.get());
+ }
+
+ @VisibleForTesting
+ public String topicToIndexName(String topicName) {
+ return topicToIndexCache.computeIfAbsent(topicName, k -> {
+ // see elasticsearch limitations https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-create-index.html#indices-create-api-path-params
+ String indexName = topicName.toLowerCase(Locale.ROOT);
+
+ // remove the pulsar topic info persistent://tenant/namespace/topic
+ String[] parts = indexName.split("/");
+ if (parts.length > 1) {
+ indexName = parts[parts.length-1];
+ }
+
+ // truncate to the max bytes length
+ while (indexName.getBytes(StandardCharsets.UTF_8).length > 255) {
+ indexName = indexName.substring(0, indexName.length() - 1);
+ }
+ if (indexName.length() <= 0 || !indexName.matches("[a-zA-Z\\.0-9][a-zA-Z_\\.\\-\\+0-9]*")) {
+ throw new RuntimeException(new IOException("Cannot convert the topic name='" + topicName + "' to a valid elasticsearch index name"));
+ }
+ if (log.isDebugEnabled()) {
+ log.debug("Translate topic={} to index={}", k, indexName);
+ }
+ return indexName;
+ });
+ }
+
+ @VisibleForTesting
+ public boolean createIndexIfNeeded(String indexName) throws IOException {
+ if (indexExists(indexName)) {
+ return false;
+ }
+ final CreateIndexRequest cireq = new CreateIndexRequest(indexName);
+ cireq.settings(Settings.builder()
+ .put("index.number_of_shards", config.getIndexNumberOfShards())
+ .put("index.number_of_replicas", config.getIndexNumberOfReplicas()));
+ return retry(() -> {
+ CreateIndexResponse resp = client.indices().create(cireq, RequestOptions.DEFAULT);
+ if (!resp.isAcknowledged() || !resp.isShardsAcknowledged()) {
+ throw new IOException("Unable to create index.");
+ }
+ return true;
+ }, "create index");
+ }
+
+ public boolean indexExists(final String indexName) throws IOException {
+ final GetIndexRequest request = new GetIndexRequest(indexName);
+ return retry(() -> client.indices().exists(request, RequestOptions.DEFAULT), "index exists");
+ }
+
+ @VisibleForTesting
+ protected long totalHits(String indexName) throws IOException {
Review comment:
good idea. btw this method is only for the tests
--
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-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] freeznet commented on a change in pull request #11263: Enhanced ElasticSearch Sink (TLS, Auth, KeyValue Schema, INSERT/UPDATE/DELETE, "_id" management, multi index support)
Posted by GitBox <gi...@apache.org>.
freeznet commented on a change in pull request #11263:
URL: https://github.com/apache/pulsar/pull/11263#discussion_r673032960
##########
File path: pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClient.java
##########
@@ -0,0 +1,569 @@
+/**
+ * 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.pulsar.io.elasticsearch;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Strings;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.http.HttpHost;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.client.CredentialsProvider;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.config.Registry;
+import org.apache.http.config.RegistryBuilder;
+import org.apache.http.conn.ssl.NoopHostnameVerifier;
+import org.apache.http.conn.ssl.SSLConnectionSocketFactory;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
+import org.apache.http.impl.nio.conn.PoolingNHttpClientConnectionManager;
+import org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor;
+import org.apache.http.impl.nio.reactor.IOReactorConfig;
+import org.apache.http.nio.conn.NHttpClientConnectionManager;
+import org.apache.http.nio.conn.NoopIOSessionStrategy;
+import org.apache.http.nio.conn.SchemeIOSessionStrategy;
+import org.apache.http.nio.conn.ssl.SSLIOSessionStrategy;
+import org.apache.http.nio.reactor.ConnectingIOReactor;
+import org.apache.http.ssl.SSLContextBuilder;
+import org.apache.http.ssl.SSLContexts;
+import org.apache.pulsar.client.api.schema.GenericObject;
+import org.apache.pulsar.functions.api.Record;
+import org.elasticsearch.action.DocWriteRequest;
+import org.elasticsearch.action.DocWriteResponse;
+import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
+import org.elasticsearch.action.bulk.BulkItemResponse;
+import org.elasticsearch.action.bulk.BulkProcessor;
+import org.elasticsearch.action.bulk.BulkRequest;
+import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.delete.DeleteRequest;
+import org.elasticsearch.action.delete.DeleteResponse;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.index.IndexResponse;
+import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.client.*;
Review comment:
please avoid importing *
##########
File path: pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClient.java
##########
@@ -0,0 +1,569 @@
+/**
+ * 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.pulsar.io.elasticsearch;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Strings;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.http.HttpHost;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.client.CredentialsProvider;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.config.Registry;
+import org.apache.http.config.RegistryBuilder;
+import org.apache.http.conn.ssl.NoopHostnameVerifier;
+import org.apache.http.conn.ssl.SSLConnectionSocketFactory;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
+import org.apache.http.impl.nio.conn.PoolingNHttpClientConnectionManager;
+import org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor;
+import org.apache.http.impl.nio.reactor.IOReactorConfig;
+import org.apache.http.nio.conn.NHttpClientConnectionManager;
+import org.apache.http.nio.conn.NoopIOSessionStrategy;
+import org.apache.http.nio.conn.SchemeIOSessionStrategy;
+import org.apache.http.nio.conn.ssl.SSLIOSessionStrategy;
+import org.apache.http.nio.reactor.ConnectingIOReactor;
+import org.apache.http.ssl.SSLContextBuilder;
+import org.apache.http.ssl.SSLContexts;
+import org.apache.pulsar.client.api.schema.GenericObject;
+import org.apache.pulsar.functions.api.Record;
+import org.elasticsearch.action.DocWriteRequest;
+import org.elasticsearch.action.DocWriteResponse;
+import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
+import org.elasticsearch.action.bulk.BulkItemResponse;
+import org.elasticsearch.action.bulk.BulkProcessor;
+import org.elasticsearch.action.bulk.BulkRequest;
+import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.delete.DeleteRequest;
+import org.elasticsearch.action.delete.DeleteResponse;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.index.IndexResponse;
+import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.client.*;
+import org.elasticsearch.client.indices.CreateIndexRequest;
+import org.elasticsearch.client.indices.CreateIndexResponse;
+import org.elasticsearch.client.indices.GetIndexRequest;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.ByteSizeUnit;
+import org.elasticsearch.common.unit.ByteSizeValue;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.common.xcontent.XContentType;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.search.SearchHit;
+import org.elasticsearch.search.builder.SearchSourceBuilder;
+
+import javax.net.ssl.HostnameVerifier;
+import javax.net.ssl.SSLContext;
+import java.io.File;
+import java.io.IOException;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.security.KeyManagementException;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
+import java.security.UnrecoverableKeyException;
+import java.security.cert.CertificateException;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicReference;
+
+@Slf4j
+public class ElasticSearchClient {
+
+ static final String[] malformedErrors = {
+ "mapper_parsing_exception",
+ "action_request_validation_exception",
+ "illegal_argument_exception"
+ };
+
+ private ElasticSearchConfig config;
+ private ConfigCallback configCallback;
+ private RestHighLevelClient client;
+
+ final Set<String> indexCache = new HashSet<>();
+ final Map<String, String> topicToIndexCache = new HashMap<>();
+
+ final RandomExponentialRetry backoffRetry;
+ final BulkProcessor bulkProcessor;
+ final ConcurrentMap<DocWriteRequest<?>, Record> records = new ConcurrentHashMap<>();
+ final AtomicReference<Exception> irrecoverableError = new AtomicReference<>();
+ final ScheduledExecutorService executorService;
+
+ ElasticSearchClient(ElasticSearchConfig elasticSearchConfig) throws MalformedURLException {
+ this.config = elasticSearchConfig;
+ this.configCallback = new ConfigCallback();
+ this.backoffRetry = new RandomExponentialRetry(elasticSearchConfig.getMaxRetryTimeInSec());
+ if (config.isBulkEnabled() == false) {
+ bulkProcessor = null;
+ } else {
+ BulkProcessor.Builder builder = BulkProcessor.builder(
+ (bulkRequest, bulkResponseActionListener) -> client.bulkAsync(bulkRequest, RequestOptions.DEFAULT, bulkResponseActionListener),
+ new BulkProcessor.Listener() {
+ @Override
+ public void beforeBulk(long l, BulkRequest bulkRequest) {
+ }
+
+ @Override
+ public void afterBulk(long l, BulkRequest bulkRequest, BulkResponse bulkResponse) {
+ log.trace("Bulk request id={} size={}:", l, bulkRequest.requests().size());
+ for (int i = 0; i < bulkResponse.getItems().length; i++) {
+ DocWriteRequest<?> request = bulkRequest.requests().get(i);
+ Record record = records.get(request);
+ BulkItemResponse bulkItemResponse = bulkResponse.getItems()[i];
+ if (bulkItemResponse.isFailed()) {
+ record.fail();
+ try {
+ hasIrrecoverableError(bulkItemResponse);
+ } catch(Exception e) {
+ log.warn("Unrecoverable error:", e);
+ }
+ } else {
+ record.ack();
+ }
+ records.remove(request);
+ }
+ }
+
+ @Override
+ public void afterBulk(long l, BulkRequest bulkRequest, Throwable throwable) {
+ log.warn("Bulk request id={} failed:", l, throwable);
+ for (DocWriteRequest<?> request : bulkRequest.requests()) {
+ Record record = records.remove(request);
+ record.fail();
+ }
+ }
+ }
+ )
+ .setBulkActions(config.getBulkActions())
+ .setBulkSize(new ByteSizeValue(config.getBulkSizeInMb(), ByteSizeUnit.MB))
+ .setConcurrentRequests(config.getBulkConcurrentRequests())
+ .setBackoffPolicy(new RandomExponentialBackoffPolicy(backoffRetry,
+ config.getRetryBackoffInMs(),
+ config.getMaxRetries()
+ ));
+ if (config.getBulkFlushIntervalInMs() > 0) {
+ builder.setFlushInterval(new TimeValue(config.getBulkFlushIntervalInMs(), TimeUnit.MILLISECONDS));
+ }
+ this.bulkProcessor = builder.build();
+ }
+
+ // idle+expired connection evictor thread
+ this.executorService = Executors.newSingleThreadScheduledExecutor();
+ this.executorService.scheduleAtFixedRate(new Runnable() {
+ @Override
+ public void run() {
+ configCallback.connectionManager.closeExpiredConnections();
+ configCallback.connectionManager.closeIdleConnections(
+ config.getConnectionIdleTimeoutInMs(), TimeUnit.MILLISECONDS);
+ }
+ },
+ config.getConnectionIdleTimeoutInMs(),
+ config.getConnectionIdleTimeoutInMs(),
+ TimeUnit.MILLISECONDS
+ );
+
+ URL url = new URL(config.getElasticSearchUrl());
+ log.info("ElasticSearch URL {}", url);
+ RestClientBuilder builder = RestClient.builder(new HttpHost(url.getHost(), url.getPort(), url.getProtocol()))
+ .setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() {
+ @Override
+ public RequestConfig.Builder customizeRequestConfig(RequestConfig.Builder builder) {
+ return builder
+ .setContentCompressionEnabled(config.isCompressionEnabled())
+ .setConnectionRequestTimeout(config.getConnectionRequestTimeoutInMs())
+ .setConnectTimeout(config.getConnectTimeoutInMs())
+ .setSocketTimeout(config.getSocketTimeoutInMs());
+ }
+ })
+ .setHttpClientConfigCallback(this.configCallback)
+ .setFailureListener(new RestClient.FailureListener() {
+ public void onFailure(Node node) {
+ log.warn("Node host={} failed", node.getHost());
+ }
+ });
+ this.client = new RestHighLevelClient(builder);
+ }
+
+ void failed(Exception e) throws Exception {
+ if (irrecoverableError.compareAndSet(null, e)) {
+ log.error("Irrecoverable error:", e);
+ }
+ }
+
+ boolean isFailed() {
+ return irrecoverableError.get() != null;
+ }
+
+ void hasIrrecoverableError(BulkItemResponse bulkItemResponse) throws Exception {
+ for (String error : malformedErrors) {
+ if (bulkItemResponse.getFailureMessage().contains(error)) {
+ switch (config.getMalformedDocAction()) {
+ case IGNORE:
+ break;
+ case WARN:
+ log.warn("Ignoring malformed document index={} id={}",
+ bulkItemResponse.getIndex(),
+ bulkItemResponse.getId(),
+ bulkItemResponse.getFailure().getCause());
+ break;
+ case FAIL:
+ log.error("Failure due to the malformed document index={} id={}",
+ bulkItemResponse.getIndex(),
+ bulkItemResponse.getId(),
+ bulkItemResponse.getFailure().getCause());
+ failed(bulkItemResponse.getFailure().getCause());
+ break;
+ }
+ }
+ }
+ }
+
+ public void bulkIndex(Record record, Pair<String, String> idAndDoc) throws Exception {
+ try {
+ checkNotFailed();
+ checkIndexExists(record.getTopicName());
+ IndexRequest indexRequest = Requests.indexRequest(config.getIndexName());
+ if (!Strings.isNullOrEmpty(idAndDoc.getLeft()))
+ indexRequest.id(idAndDoc.getLeft());
+ indexRequest.type(config.getTypeName());
+ indexRequest.source(idAndDoc.getRight(), XContentType.JSON);
+
+ records.put(indexRequest, record);
+ bulkProcessor.add(indexRequest);
+ } catch(Exception e) {
+ log.debug("index failed id=" + idAndDoc.getLeft(), e);
+ record.fail();
+ throw e;
+ }
+ }
+
+ /**
+ * Index an elasticsearch document and ack the record.
+ * @param record
+ * @param idAndDoc
+ * @return
+ * @throws Exception
+ */
+ public boolean indexDocument(Record<GenericObject> record, Pair<String, String> idAndDoc) throws Exception {
+ try {
+ checkNotFailed();
+ checkIndexExists(record.getTopicName());
+ IndexRequest indexRequest = Requests.indexRequest(config.getIndexName());
+ if (!Strings.isNullOrEmpty(idAndDoc.getLeft()))
+ indexRequest.id(idAndDoc.getLeft());
+ indexRequest.type(config.getTypeName());
+ indexRequest.source(idAndDoc.getRight(), XContentType.JSON);
+ IndexResponse indexResponse = client.index(indexRequest, RequestOptions.DEFAULT);
+ if (indexResponse.getResult().equals(DocWriteResponse.Result.CREATED) ||
+ indexResponse.getResult().equals(DocWriteResponse.Result.UPDATED)) {
+ record.ack();
+ return true;
+ } else {
+ record.fail();
+ return false;
+ }
+ } catch (final Exception ex) {
+ log.error("index failed id=" + idAndDoc.getLeft(), ex);
+ record.fail();
+ throw ex;
+ }
+ }
+
+ public void bulkDelete(Record<GenericObject> record, String id) throws Exception {
+ try {
+ checkNotFailed();
+ checkIndexExists(record.getTopicName());
+ DeleteRequest deleteRequest = Requests.deleteRequest(config.getIndexName());
+ deleteRequest.id(id);
+ deleteRequest.type(config.getTypeName());
+
+ records.put(deleteRequest, record);
+ bulkProcessor.add(deleteRequest);
+ } catch(Exception e) {
+ log.debug("delete failed id=" + id, e);
+ record.fail();
+ throw e;
+ }
+ }
+
+ /**
+ * Delete an elasticsearch document and ack the record.
+ * @param record
+ * @param id
+ * @return
+ * @throws IOException
+ */
+ public boolean deleteDocument(Record<GenericObject> record, String id) throws Exception {
+ try {
+ checkNotFailed();
+ checkIndexExists(record.getTopicName());
+ DeleteRequest deleteRequest = Requests.deleteRequest(config.getIndexName());
+ deleteRequest.id(id);
+ deleteRequest.type(config.getTypeName());
+ DeleteResponse deleteResponse = client.delete(deleteRequest, RequestOptions.DEFAULT);
+ log.debug("delete result=" + deleteResponse.getResult());
+ if (deleteResponse.getResult().equals(DocWriteResponse.Result.DELETED) ||
+ deleteResponse.getResult().equals(DocWriteResponse.Result.NOT_FOUND)) {
+ record.ack();
+ return true;
+ }
+ record.fail();
+ return false;
+ } catch (final Exception ex) {
+ log.debug("index failed id=" + id, ex);
+ record.fail();
+ throw ex;
+ }
+ }
+
+ /**
+ * Flushes the bulk processor.
+ */
+ public void flush() {
+ bulkProcessor.flush();
+ }
+
+ public void close() {
+ try {
+ if (bulkProcessor != null) {
+ bulkProcessor.awaitClose(5000L, TimeUnit.MILLISECONDS);
+ }
+ } catch (InterruptedException e) {
+ log.warn("Elasticsearch bulk processor close error:", e);
+ }
+ try {
+ this.executorService.shutdown();
+ if (this.client != null) {
+ this.client.close();
+ }
+ } catch (IOException e) {
+ log.warn("Elasticsearch client close error:", e);
+ }
+ }
+
+ private void checkNotFailed() throws Exception {
+ if (irrecoverableError.get() != null) {
+ throw irrecoverableError.get();
+ }
+ }
+
+ private void checkIndexExists(Optional<String> topicName) throws IOException {
+ if (!config.isCreateIndexIfNeeded()) {
+ return;
+ }
+ String indexName = indexName(topicName);
+ if (!indexCache.contains(indexName)) {
+ synchronized (this) {
+ if (!indexCache.contains(indexName)) {
+ createIndexIfNeeded(indexName);
+ indexCache.add(indexName);
+ }
+ }
+ }
+ }
+
+ private String indexName(Optional<String> topicName) throws IOException {
+ if (config.getIndexName() != null) {
+ // Use the configured indexName if provided.
+ return config.getIndexName();
+ }
+ if (!topicName.isPresent()) {
+ throw new IOException("Elasticsearch index name configuration and topic name are empty");
+ }
+ return topicToIndexName(topicName.get());
+ }
+
+ @VisibleForTesting
+ public String topicToIndexName(String topicName) {
+ return topicToIndexCache.computeIfAbsent(topicName, k -> {
+ // see elasticsearch limitations https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-create-index.html#indices-create-api-path-params
+ String indexName = topicName.toLowerCase(Locale.ROOT);
+
+ // remove the pulsar topic info persistent://tenant/namespace/topic
+ String[] parts = indexName.split("/");
+ if (parts.length > 1) {
+ indexName = parts[parts.length-1];
+ }
+
+ // truncate to the max bytes length
+ while (indexName.getBytes(StandardCharsets.UTF_8).length > 255) {
+ indexName = indexName.substring(0, indexName.length() - 1);
+ }
+ if (indexName.length() <= 0 || !indexName.matches("[a-zA-Z\\.0-9][a-zA-Z_\\.\\-\\+0-9]*")) {
+ throw new RuntimeException(new IOException("Cannot convert the topic name='" + topicName + "' to a valid elasticsearch index name"));
+ }
+ if (log.isDebugEnabled()) {
+ log.debug("Translate topic={} to index={}", k, indexName);
+ }
+ return indexName;
+ });
+ }
+
+ @VisibleForTesting
+ public boolean createIndexIfNeeded(String indexName) throws IOException {
+ if (indexExists(indexName)) {
+ return false;
+ }
+ final CreateIndexRequest cireq = new CreateIndexRequest(indexName);
+ cireq.settings(Settings.builder()
+ .put("index.number_of_shards", config.getIndexNumberOfShards())
+ .put("index.number_of_replicas", config.getIndexNumberOfReplicas()));
+ return retry(() -> {
+ CreateIndexResponse resp = client.indices().create(cireq, RequestOptions.DEFAULT);
+ if (!resp.isAcknowledged() || !resp.isShardsAcknowledged()) {
+ throw new IOException("Unable to create index.");
+ }
+ return true;
+ }, "create index");
+ }
+
+ public boolean indexExists(final String indexName) throws IOException {
+ final GetIndexRequest request = new GetIndexRequest(indexName);
+ return retry(() -> client.indices().exists(request, RequestOptions.DEFAULT), "index exists");
+ }
+
+ @VisibleForTesting
+ protected long totalHits(String indexName) throws IOException {
+ client.indices().refresh(new RefreshRequest(indexName), RequestOptions.DEFAULT);
+ SearchResponse response = client.search(
+ new SearchRequest()
+ .indices(indexName)
+ .source(new SearchSourceBuilder().query(QueryBuilders.matchAllQuery())),
+ RequestOptions.DEFAULT);
+ for(SearchHit searchHit : response.getHits()) {
+ System.out.println(searchHit.getId()+": "+searchHit.getFields());
Review comment:
why not print with logger?
##########
File path: pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java
##########
@@ -18,143 +18,273 @@
*/
package org.apache.pulsar.io.elasticsearch;
-import java.io.IOException;
-import java.net.MalformedURLException;
-import java.net.URL;
-import java.util.Map;
-
-import org.apache.commons.lang3.StringUtils;
-import org.apache.http.HttpHost;
-import org.apache.http.auth.AuthScope;
-import org.apache.http.auth.UsernamePasswordCredentials;
-import org.apache.http.client.CredentialsProvider;
-import org.apache.http.impl.client.BasicCredentialsProvider;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Strings;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.schema.GenericObject;
+import org.apache.pulsar.client.api.schema.GenericRecord;
+import org.apache.pulsar.client.api.schema.KeyValueSchema;
+import org.apache.pulsar.common.schema.KeyValue;
+import org.apache.pulsar.common.schema.SchemaType;
import org.apache.pulsar.functions.api.Record;
-import org.apache.pulsar.io.core.KeyValue;
import org.apache.pulsar.io.core.Sink;
import org.apache.pulsar.io.core.SinkContext;
import org.apache.pulsar.io.core.annotations.Connector;
import org.apache.pulsar.io.core.annotations.IOType;
-import org.elasticsearch.action.DocWriteResponse;
-import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
-import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
-import org.elasticsearch.action.admin.indices.get.GetIndexRequest;
-import org.elasticsearch.action.index.IndexRequest;
-import org.elasticsearch.action.index.IndexResponse;
-import org.elasticsearch.client.RequestOptions;
-import org.elasticsearch.client.Requests;
-import org.elasticsearch.client.RestClient;
-import org.elasticsearch.client.RestClientBuilder;
-import org.elasticsearch.client.RestHighLevelClient;
-import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.xcontent.XContentType;
-/**
- * The base abstract class for ElasticSearch sinks.
- * Users need to implement extractKeyValue function to use this sink.
- * This class assumes that the input will be JSON documents
- */
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
@Connector(
- name = "elastic_search",
- type = IOType.SINK,
- help = "A sink connector that sends pulsar messages to elastic search",
- configClass = ElasticSearchConfig.class
+ name = "elastic_search",
+ type = IOType.SINK,
+ help = "A sink connector that sends pulsar messages to elastic search",
+ configClass = ElasticSearchConfig.class
)
-public class ElasticSearchSink implements Sink<byte[]> {
+@Slf4j
+public class ElasticSearchSink implements Sink<GenericObject> {
- private URL url;
- private RestHighLevelClient client;
- private CredentialsProvider credentialsProvider;
private ElasticSearchConfig elasticSearchConfig;
+ private ElasticSearchClient elasticsearchClient;
+ private ObjectMapper objectMapper = new ObjectMapper();
+ private List<String> primaryFields = null;
@Override
public void open(Map<String, Object> config, SinkContext sinkContext) throws Exception {
elasticSearchConfig = ElasticSearchConfig.load(config);
elasticSearchConfig.validate();
- createIndexIfNeeded();
+ elasticsearchClient = new ElasticSearchClient(elasticSearchConfig);
+ if (!Strings.isNullOrEmpty(elasticSearchConfig.getPrimaryFields())) {
+ primaryFields = Arrays.asList(elasticSearchConfig.getPrimaryFields().split(","));
+ }
}
@Override
- public void close() throws Exception {
- client.close();
+ public void close() {
+ if (elasticsearchClient != null) {
+ elasticsearchClient.close();
+ elasticsearchClient = null;
+ }
}
@Override
- public void write(Record<byte[]> record) {
- KeyValue<String, byte[]> keyValue = extractKeyValue(record);
- IndexRequest indexRequest = Requests.indexRequest(elasticSearchConfig.getIndexName());
- indexRequest.type(elasticSearchConfig.getTypeName());
- indexRequest.source(keyValue.getValue(), XContentType.JSON);
-
- try {
- IndexResponse indexResponse = getClient().index(indexRequest, RequestOptions.DEFAULT);
- if (indexResponse.getResult().equals(DocWriteResponse.Result.CREATED)) {
- record.ack();
- } else {
- record.fail();
+ public void write(Record<GenericObject> record) throws Exception {
+ if (!elasticsearchClient.isFailed()) {
+ Pair<String, String> idAndDoc = extractIdAndDocument(record);
+ try {
+ if (log.isDebugEnabled()) {
+ log.debug("index doc {} {}", idAndDoc.getLeft(), idAndDoc.getRight());
+ }
+ if (idAndDoc.getRight() == null) {
+ switch (elasticSearchConfig.getNullValueAction()) {
+ case DELETE:
+ if (idAndDoc.getLeft() != null) {
+ if (elasticSearchConfig.isBulkEnabled()) {
+ elasticsearchClient.bulkDelete(record, idAndDoc.getLeft());
+ } else {
+ elasticsearchClient.deleteDocument(record, idAndDoc.getLeft());
+ }
+ }
+ break;
+ case IGNORE:
+ break;
+ case FAIL:
+ elasticsearchClient.failed(new PulsarClientException.InvalidMessageException("Unexpected null message value"));
+ throw elasticsearchClient.irrecoverableError.get();
+ }
+ } else {
+ if (elasticSearchConfig.isBulkEnabled()) {
+ elasticsearchClient.bulkIndex(record, idAndDoc);
+ } else {
+ elasticsearchClient.indexDocument(record, idAndDoc);
+ }
+ }
+ } catch (JsonProcessingException jsonProcessingException) {
+ switch (elasticSearchConfig.getMalformedDocAction()) {
+ case IGNORE:
+ break;
+ case WARN:
+ log.warn("Ignoring malformed document messageId={}",
+ record.getMessage().map(Message::getMessageId).orElse(null),
+ jsonProcessingException);
+ elasticsearchClient.failed(jsonProcessingException);
+ throw jsonProcessingException;
+ case FAIL:
+ log.error("Malformed document messageId={}",
+ record.getMessage().map(Message::getMessageId).orElse(null),
+ jsonProcessingException);
+ elasticsearchClient.failed(jsonProcessingException);
+ throw jsonProcessingException;
+ }
+ } catch (Exception e) {
+ log.error("write error for {} {}:", idAndDoc.getLeft(), idAndDoc.getRight(), e);
+ throw e;
}
- } catch (final IOException ex) {
- record.fail();
+ } else {
+ throw new IllegalStateException("Elasticsearch client is in FAILED status");
}
}
- public KeyValue<String, byte[]> extractKeyValue(Record<byte[]> record) {
- String key = record.getKey().orElse("");
- return new KeyValue<>(key, record.getValue());
+ @VisibleForTesting
+ ElasticSearchClient getElasticsearchClient() {
+ return this.elasticsearchClient;
}
- private void createIndexIfNeeded() throws IOException {
- GetIndexRequest request = new GetIndexRequest();
- request.indices(elasticSearchConfig.getIndexName());
- boolean exists = getClient().indices().exists(request, RequestOptions.DEFAULT);
+ /**
+ * Extract ES _id and _source using the Schema if available.
+ *
+ * @param record
+ * @return A pair for _id and _source
+ */
+ public Pair<String, String> extractIdAndDocument(Record<GenericObject> record) throws JsonProcessingException {
+ if (elasticSearchConfig.isSchemaEnable()) {
+ Object key = null;
+ GenericObject value = null;
+ Schema<?> keySchema = null;
+ Schema<?> valueSchema = null;
+ if (record.getSchema() != null && record.getSchema() instanceof KeyValueSchema) {
+ KeyValueSchema<GenericObject, GenericObject> keyValueSchema = (KeyValueSchema) record.getSchema();
+ keySchema = keyValueSchema.getKeySchema();
+ valueSchema = keyValueSchema.getValueSchema();
+ KeyValue<GenericObject, GenericObject> keyValue = (KeyValue<GenericObject, GenericObject>) record.getValue().getNativeObject();
+ key = keyValue.getKey();
+ value = keyValue.getValue();
+ } else {
+ key = record.getKey().orElse(null);
+ valueSchema = record.getSchema();
+ value = record.getValue();
+ }
+
+ String id = null;
+ if (elasticSearchConfig.isKeyIgnore() == false && key != null && keySchema != null) {
+ id = stringifyKey(keySchema, key);
+ }
- if (!exists) {
- CreateIndexRequest cireq = new CreateIndexRequest(elasticSearchConfig.getIndexName());
+ String doc = null;
+ if (value != null) {
+ if (valueSchema != null) {
+ doc = stringifyValue(valueSchema, value);
+ } else {
+ if (value.getNativeObject() instanceof byte[]) {
+ // for BWC with the ES-Sink
+ doc = new String((byte[]) value.getNativeObject(), StandardCharsets.UTF_8);
+ } else {
+ doc = value.getNativeObject().toString();
+ }
+ }
+ }
- cireq.settings(Settings.builder()
- .put("index.number_of_shards", elasticSearchConfig.getIndexNumberOfShards())
- .put("index.number_of_replicas", elasticSearchConfig.getIndexNumberOfReplicas()));
+ if (doc != null && primaryFields != null) {
+ try {
+ // extract the PK from the JSON document
+ JsonNode jsonNode = objectMapper.readTree(doc);
+ id = stringifyKey(jsonNode, primaryFields);
+ } catch (JsonProcessingException e) {
+ log.error("Failed to read JSON", e);
+ throw e;
+ }
+ }
- CreateIndexResponse ciresp = getClient().indices().create(cireq, RequestOptions.DEFAULT);
- if (!ciresp.isAcknowledged() || !ciresp.isShardsAcknowledged()) {
- throw new RuntimeException("Unable to create index.");
+ if (log.isDebugEnabled()) {
+ SchemaType schemaType = null;
+ if (record.getSchema() != null && record.getSchema().getSchemaInfo() != null) {
+ schemaType = record.getSchema().getSchemaInfo().getType();
+ }
+ log.debug("recordType={} schemaType={} id={} doc={}",
+ record.getClass().getName(),
+ schemaType,
+ id,
+ doc);
}
+ return Pair.of(id, doc);
+ } else {
+ return Pair.of(null, new String(
Review comment:
should the record key be the `_id` here?
##########
File path: pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClient.java
##########
@@ -0,0 +1,569 @@
+/**
+ * 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.pulsar.io.elasticsearch;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Strings;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.http.HttpHost;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.client.CredentialsProvider;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.config.Registry;
+import org.apache.http.config.RegistryBuilder;
+import org.apache.http.conn.ssl.NoopHostnameVerifier;
+import org.apache.http.conn.ssl.SSLConnectionSocketFactory;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
+import org.apache.http.impl.nio.conn.PoolingNHttpClientConnectionManager;
+import org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor;
+import org.apache.http.impl.nio.reactor.IOReactorConfig;
+import org.apache.http.nio.conn.NHttpClientConnectionManager;
+import org.apache.http.nio.conn.NoopIOSessionStrategy;
+import org.apache.http.nio.conn.SchemeIOSessionStrategy;
+import org.apache.http.nio.conn.ssl.SSLIOSessionStrategy;
+import org.apache.http.nio.reactor.ConnectingIOReactor;
+import org.apache.http.ssl.SSLContextBuilder;
+import org.apache.http.ssl.SSLContexts;
+import org.apache.pulsar.client.api.schema.GenericObject;
+import org.apache.pulsar.functions.api.Record;
+import org.elasticsearch.action.DocWriteRequest;
+import org.elasticsearch.action.DocWriteResponse;
+import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
+import org.elasticsearch.action.bulk.BulkItemResponse;
+import org.elasticsearch.action.bulk.BulkProcessor;
+import org.elasticsearch.action.bulk.BulkRequest;
+import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.delete.DeleteRequest;
+import org.elasticsearch.action.delete.DeleteResponse;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.index.IndexResponse;
+import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.client.*;
+import org.elasticsearch.client.indices.CreateIndexRequest;
+import org.elasticsearch.client.indices.CreateIndexResponse;
+import org.elasticsearch.client.indices.GetIndexRequest;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.ByteSizeUnit;
+import org.elasticsearch.common.unit.ByteSizeValue;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.common.xcontent.XContentType;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.search.SearchHit;
+import org.elasticsearch.search.builder.SearchSourceBuilder;
+
+import javax.net.ssl.HostnameVerifier;
+import javax.net.ssl.SSLContext;
+import java.io.File;
+import java.io.IOException;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.security.KeyManagementException;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
+import java.security.UnrecoverableKeyException;
+import java.security.cert.CertificateException;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicReference;
+
+@Slf4j
+public class ElasticSearchClient {
+
+ static final String[] malformedErrors = {
+ "mapper_parsing_exception",
+ "action_request_validation_exception",
+ "illegal_argument_exception"
+ };
+
+ private ElasticSearchConfig config;
+ private ConfigCallback configCallback;
+ private RestHighLevelClient client;
+
+ final Set<String> indexCache = new HashSet<>();
+ final Map<String, String> topicToIndexCache = new HashMap<>();
+
+ final RandomExponentialRetry backoffRetry;
+ final BulkProcessor bulkProcessor;
+ final ConcurrentMap<DocWriteRequest<?>, Record> records = new ConcurrentHashMap<>();
+ final AtomicReference<Exception> irrecoverableError = new AtomicReference<>();
+ final ScheduledExecutorService executorService;
+
+ ElasticSearchClient(ElasticSearchConfig elasticSearchConfig) throws MalformedURLException {
+ this.config = elasticSearchConfig;
+ this.configCallback = new ConfigCallback();
+ this.backoffRetry = new RandomExponentialRetry(elasticSearchConfig.getMaxRetryTimeInSec());
+ if (config.isBulkEnabled() == false) {
+ bulkProcessor = null;
+ } else {
+ BulkProcessor.Builder builder = BulkProcessor.builder(
+ (bulkRequest, bulkResponseActionListener) -> client.bulkAsync(bulkRequest, RequestOptions.DEFAULT, bulkResponseActionListener),
+ new BulkProcessor.Listener() {
+ @Override
+ public void beforeBulk(long l, BulkRequest bulkRequest) {
+ }
+
+ @Override
+ public void afterBulk(long l, BulkRequest bulkRequest, BulkResponse bulkResponse) {
+ log.trace("Bulk request id={} size={}:", l, bulkRequest.requests().size());
+ for (int i = 0; i < bulkResponse.getItems().length; i++) {
+ DocWriteRequest<?> request = bulkRequest.requests().get(i);
+ Record record = records.get(request);
+ BulkItemResponse bulkItemResponse = bulkResponse.getItems()[i];
+ if (bulkItemResponse.isFailed()) {
+ record.fail();
+ try {
+ hasIrrecoverableError(bulkItemResponse);
+ } catch(Exception e) {
+ log.warn("Unrecoverable error:", e);
+ }
+ } else {
+ record.ack();
+ }
+ records.remove(request);
+ }
+ }
+
+ @Override
+ public void afterBulk(long l, BulkRequest bulkRequest, Throwable throwable) {
+ log.warn("Bulk request id={} failed:", l, throwable);
+ for (DocWriteRequest<?> request : bulkRequest.requests()) {
+ Record record = records.remove(request);
+ record.fail();
+ }
+ }
+ }
+ )
+ .setBulkActions(config.getBulkActions())
+ .setBulkSize(new ByteSizeValue(config.getBulkSizeInMb(), ByteSizeUnit.MB))
+ .setConcurrentRequests(config.getBulkConcurrentRequests())
+ .setBackoffPolicy(new RandomExponentialBackoffPolicy(backoffRetry,
+ config.getRetryBackoffInMs(),
+ config.getMaxRetries()
+ ));
+ if (config.getBulkFlushIntervalInMs() > 0) {
+ builder.setFlushInterval(new TimeValue(config.getBulkFlushIntervalInMs(), TimeUnit.MILLISECONDS));
+ }
+ this.bulkProcessor = builder.build();
+ }
+
+ // idle+expired connection evictor thread
+ this.executorService = Executors.newSingleThreadScheduledExecutor();
+ this.executorService.scheduleAtFixedRate(new Runnable() {
+ @Override
+ public void run() {
+ configCallback.connectionManager.closeExpiredConnections();
+ configCallback.connectionManager.closeIdleConnections(
+ config.getConnectionIdleTimeoutInMs(), TimeUnit.MILLISECONDS);
+ }
+ },
+ config.getConnectionIdleTimeoutInMs(),
+ config.getConnectionIdleTimeoutInMs(),
+ TimeUnit.MILLISECONDS
+ );
+
+ URL url = new URL(config.getElasticSearchUrl());
+ log.info("ElasticSearch URL {}", url);
+ RestClientBuilder builder = RestClient.builder(new HttpHost(url.getHost(), url.getPort(), url.getProtocol()))
+ .setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() {
+ @Override
+ public RequestConfig.Builder customizeRequestConfig(RequestConfig.Builder builder) {
+ return builder
+ .setContentCompressionEnabled(config.isCompressionEnabled())
+ .setConnectionRequestTimeout(config.getConnectionRequestTimeoutInMs())
+ .setConnectTimeout(config.getConnectTimeoutInMs())
+ .setSocketTimeout(config.getSocketTimeoutInMs());
+ }
+ })
+ .setHttpClientConfigCallback(this.configCallback)
+ .setFailureListener(new RestClient.FailureListener() {
+ public void onFailure(Node node) {
+ log.warn("Node host={} failed", node.getHost());
+ }
+ });
+ this.client = new RestHighLevelClient(builder);
+ }
+
+ void failed(Exception e) throws Exception {
+ if (irrecoverableError.compareAndSet(null, e)) {
+ log.error("Irrecoverable error:", e);
+ }
+ }
+
+ boolean isFailed() {
+ return irrecoverableError.get() != null;
+ }
+
+ void hasIrrecoverableError(BulkItemResponse bulkItemResponse) throws Exception {
+ for (String error : malformedErrors) {
+ if (bulkItemResponse.getFailureMessage().contains(error)) {
+ switch (config.getMalformedDocAction()) {
+ case IGNORE:
+ break;
+ case WARN:
+ log.warn("Ignoring malformed document index={} id={}",
+ bulkItemResponse.getIndex(),
+ bulkItemResponse.getId(),
+ bulkItemResponse.getFailure().getCause());
+ break;
+ case FAIL:
+ log.error("Failure due to the malformed document index={} id={}",
+ bulkItemResponse.getIndex(),
+ bulkItemResponse.getId(),
+ bulkItemResponse.getFailure().getCause());
+ failed(bulkItemResponse.getFailure().getCause());
+ break;
+ }
+ }
+ }
+ }
+
+ public void bulkIndex(Record record, Pair<String, String> idAndDoc) throws Exception {
+ try {
+ checkNotFailed();
+ checkIndexExists(record.getTopicName());
+ IndexRequest indexRequest = Requests.indexRequest(config.getIndexName());
+ if (!Strings.isNullOrEmpty(idAndDoc.getLeft()))
+ indexRequest.id(idAndDoc.getLeft());
+ indexRequest.type(config.getTypeName());
+ indexRequest.source(idAndDoc.getRight(), XContentType.JSON);
+
+ records.put(indexRequest, record);
+ bulkProcessor.add(indexRequest);
+ } catch(Exception e) {
+ log.debug("index failed id=" + idAndDoc.getLeft(), e);
+ record.fail();
+ throw e;
+ }
+ }
+
+ /**
+ * Index an elasticsearch document and ack the record.
+ * @param record
+ * @param idAndDoc
+ * @return
+ * @throws Exception
+ */
+ public boolean indexDocument(Record<GenericObject> record, Pair<String, String> idAndDoc) throws Exception {
+ try {
+ checkNotFailed();
+ checkIndexExists(record.getTopicName());
+ IndexRequest indexRequest = Requests.indexRequest(config.getIndexName());
+ if (!Strings.isNullOrEmpty(idAndDoc.getLeft()))
+ indexRequest.id(idAndDoc.getLeft());
+ indexRequest.type(config.getTypeName());
+ indexRequest.source(idAndDoc.getRight(), XContentType.JSON);
+ IndexResponse indexResponse = client.index(indexRequest, RequestOptions.DEFAULT);
+ if (indexResponse.getResult().equals(DocWriteResponse.Result.CREATED) ||
+ indexResponse.getResult().equals(DocWriteResponse.Result.UPDATED)) {
+ record.ack();
+ return true;
+ } else {
+ record.fail();
+ return false;
+ }
+ } catch (final Exception ex) {
+ log.error("index failed id=" + idAndDoc.getLeft(), ex);
+ record.fail();
+ throw ex;
+ }
+ }
+
+ public void bulkDelete(Record<GenericObject> record, String id) throws Exception {
+ try {
+ checkNotFailed();
+ checkIndexExists(record.getTopicName());
+ DeleteRequest deleteRequest = Requests.deleteRequest(config.getIndexName());
+ deleteRequest.id(id);
+ deleteRequest.type(config.getTypeName());
+
+ records.put(deleteRequest, record);
+ bulkProcessor.add(deleteRequest);
+ } catch(Exception e) {
+ log.debug("delete failed id=" + id, e);
+ record.fail();
+ throw e;
+ }
+ }
+
+ /**
+ * Delete an elasticsearch document and ack the record.
+ * @param record
+ * @param id
+ * @return
+ * @throws IOException
+ */
+ public boolean deleteDocument(Record<GenericObject> record, String id) throws Exception {
+ try {
+ checkNotFailed();
+ checkIndexExists(record.getTopicName());
+ DeleteRequest deleteRequest = Requests.deleteRequest(config.getIndexName());
+ deleteRequest.id(id);
+ deleteRequest.type(config.getTypeName());
+ DeleteResponse deleteResponse = client.delete(deleteRequest, RequestOptions.DEFAULT);
+ log.debug("delete result=" + deleteResponse.getResult());
+ if (deleteResponse.getResult().equals(DocWriteResponse.Result.DELETED) ||
+ deleteResponse.getResult().equals(DocWriteResponse.Result.NOT_FOUND)) {
+ record.ack();
+ return true;
+ }
+ record.fail();
+ return false;
+ } catch (final Exception ex) {
+ log.debug("index failed id=" + id, ex);
+ record.fail();
+ throw ex;
+ }
+ }
+
+ /**
+ * Flushes the bulk processor.
+ */
+ public void flush() {
+ bulkProcessor.flush();
+ }
+
+ public void close() {
+ try {
+ if (bulkProcessor != null) {
+ bulkProcessor.awaitClose(5000L, TimeUnit.MILLISECONDS);
+ }
+ } catch (InterruptedException e) {
+ log.warn("Elasticsearch bulk processor close error:", e);
+ }
+ try {
+ this.executorService.shutdown();
+ if (this.client != null) {
+ this.client.close();
+ }
+ } catch (IOException e) {
+ log.warn("Elasticsearch client close error:", e);
+ }
+ }
+
+ private void checkNotFailed() throws Exception {
+ if (irrecoverableError.get() != null) {
+ throw irrecoverableError.get();
+ }
+ }
+
+ private void checkIndexExists(Optional<String> topicName) throws IOException {
+ if (!config.isCreateIndexIfNeeded()) {
+ return;
+ }
+ String indexName = indexName(topicName);
+ if (!indexCache.contains(indexName)) {
+ synchronized (this) {
+ if (!indexCache.contains(indexName)) {
+ createIndexIfNeeded(indexName);
+ indexCache.add(indexName);
+ }
+ }
+ }
+ }
+
+ private String indexName(Optional<String> topicName) throws IOException {
+ if (config.getIndexName() != null) {
+ // Use the configured indexName if provided.
+ return config.getIndexName();
+ }
+ if (!topicName.isPresent()) {
+ throw new IOException("Elasticsearch index name configuration and topic name are empty");
+ }
+ return topicToIndexName(topicName.get());
+ }
+
+ @VisibleForTesting
+ public String topicToIndexName(String topicName) {
+ return topicToIndexCache.computeIfAbsent(topicName, k -> {
+ // see elasticsearch limitations https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-create-index.html#indices-create-api-path-params
+ String indexName = topicName.toLowerCase(Locale.ROOT);
+
+ // remove the pulsar topic info persistent://tenant/namespace/topic
+ String[] parts = indexName.split("/");
+ if (parts.length > 1) {
+ indexName = parts[parts.length-1];
+ }
+
+ // truncate to the max bytes length
+ while (indexName.getBytes(StandardCharsets.UTF_8).length > 255) {
+ indexName = indexName.substring(0, indexName.length() - 1);
+ }
+ if (indexName.length() <= 0 || !indexName.matches("[a-zA-Z\\.0-9][a-zA-Z_\\.\\-\\+0-9]*")) {
+ throw new RuntimeException(new IOException("Cannot convert the topic name='" + topicName + "' to a valid elasticsearch index name"));
+ }
+ if (log.isDebugEnabled()) {
+ log.debug("Translate topic={} to index={}", k, indexName);
+ }
+ return indexName;
+ });
+ }
+
+ @VisibleForTesting
+ public boolean createIndexIfNeeded(String indexName) throws IOException {
+ if (indexExists(indexName)) {
+ return false;
+ }
+ final CreateIndexRequest cireq = new CreateIndexRequest(indexName);
+ cireq.settings(Settings.builder()
+ .put("index.number_of_shards", config.getIndexNumberOfShards())
+ .put("index.number_of_replicas", config.getIndexNumberOfReplicas()));
+ return retry(() -> {
+ CreateIndexResponse resp = client.indices().create(cireq, RequestOptions.DEFAULT);
+ if (!resp.isAcknowledged() || !resp.isShardsAcknowledged()) {
+ throw new IOException("Unable to create index.");
+ }
+ return true;
+ }, "create index");
+ }
+
+ public boolean indexExists(final String indexName) throws IOException {
+ final GetIndexRequest request = new GetIndexRequest(indexName);
+ return retry(() -> client.indices().exists(request, RequestOptions.DEFAULT), "index exists");
+ }
+
+ @VisibleForTesting
+ protected long totalHits(String indexName) throws IOException {
Review comment:
noticed that there are some code duplicates in `totalHits(String indexName)` and `search(String indexName)`, can `totalHits` reuse `search` and return the `getHits().getTotalHits().value`?
##########
File path: pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java
##########
@@ -18,143 +18,273 @@
*/
package org.apache.pulsar.io.elasticsearch;
-import java.io.IOException;
-import java.net.MalformedURLException;
-import java.net.URL;
-import java.util.Map;
-
-import org.apache.commons.lang3.StringUtils;
-import org.apache.http.HttpHost;
-import org.apache.http.auth.AuthScope;
-import org.apache.http.auth.UsernamePasswordCredentials;
-import org.apache.http.client.CredentialsProvider;
-import org.apache.http.impl.client.BasicCredentialsProvider;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Strings;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.schema.GenericObject;
+import org.apache.pulsar.client.api.schema.GenericRecord;
+import org.apache.pulsar.client.api.schema.KeyValueSchema;
+import org.apache.pulsar.common.schema.KeyValue;
+import org.apache.pulsar.common.schema.SchemaType;
import org.apache.pulsar.functions.api.Record;
-import org.apache.pulsar.io.core.KeyValue;
import org.apache.pulsar.io.core.Sink;
import org.apache.pulsar.io.core.SinkContext;
import org.apache.pulsar.io.core.annotations.Connector;
import org.apache.pulsar.io.core.annotations.IOType;
-import org.elasticsearch.action.DocWriteResponse;
-import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
-import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
-import org.elasticsearch.action.admin.indices.get.GetIndexRequest;
-import org.elasticsearch.action.index.IndexRequest;
-import org.elasticsearch.action.index.IndexResponse;
-import org.elasticsearch.client.RequestOptions;
-import org.elasticsearch.client.Requests;
-import org.elasticsearch.client.RestClient;
-import org.elasticsearch.client.RestClientBuilder;
-import org.elasticsearch.client.RestHighLevelClient;
-import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.xcontent.XContentType;
-/**
- * The base abstract class for ElasticSearch sinks.
- * Users need to implement extractKeyValue function to use this sink.
- * This class assumes that the input will be JSON documents
- */
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
@Connector(
- name = "elastic_search",
- type = IOType.SINK,
- help = "A sink connector that sends pulsar messages to elastic search",
- configClass = ElasticSearchConfig.class
+ name = "elastic_search",
+ type = IOType.SINK,
+ help = "A sink connector that sends pulsar messages to elastic search",
+ configClass = ElasticSearchConfig.class
)
-public class ElasticSearchSink implements Sink<byte[]> {
+@Slf4j
+public class ElasticSearchSink implements Sink<GenericObject> {
- private URL url;
- private RestHighLevelClient client;
- private CredentialsProvider credentialsProvider;
private ElasticSearchConfig elasticSearchConfig;
+ private ElasticSearchClient elasticsearchClient;
+ private ObjectMapper objectMapper = new ObjectMapper();
+ private List<String> primaryFields = null;
@Override
public void open(Map<String, Object> config, SinkContext sinkContext) throws Exception {
elasticSearchConfig = ElasticSearchConfig.load(config);
elasticSearchConfig.validate();
- createIndexIfNeeded();
+ elasticsearchClient = new ElasticSearchClient(elasticSearchConfig);
+ if (!Strings.isNullOrEmpty(elasticSearchConfig.getPrimaryFields())) {
+ primaryFields = Arrays.asList(elasticSearchConfig.getPrimaryFields().split(","));
+ }
}
@Override
- public void close() throws Exception {
- client.close();
+ public void close() {
+ if (elasticsearchClient != null) {
+ elasticsearchClient.close();
+ elasticsearchClient = null;
+ }
}
@Override
- public void write(Record<byte[]> record) {
- KeyValue<String, byte[]> keyValue = extractKeyValue(record);
- IndexRequest indexRequest = Requests.indexRequest(elasticSearchConfig.getIndexName());
- indexRequest.type(elasticSearchConfig.getTypeName());
- indexRequest.source(keyValue.getValue(), XContentType.JSON);
-
- try {
- IndexResponse indexResponse = getClient().index(indexRequest, RequestOptions.DEFAULT);
- if (indexResponse.getResult().equals(DocWriteResponse.Result.CREATED)) {
- record.ack();
- } else {
- record.fail();
+ public void write(Record<GenericObject> record) throws Exception {
+ if (!elasticsearchClient.isFailed()) {
+ Pair<String, String> idAndDoc = extractIdAndDocument(record);
+ try {
+ if (log.isDebugEnabled()) {
+ log.debug("index doc {} {}", idAndDoc.getLeft(), idAndDoc.getRight());
+ }
+ if (idAndDoc.getRight() == null) {
+ switch (elasticSearchConfig.getNullValueAction()) {
+ case DELETE:
+ if (idAndDoc.getLeft() != null) {
+ if (elasticSearchConfig.isBulkEnabled()) {
+ elasticsearchClient.bulkDelete(record, idAndDoc.getLeft());
+ } else {
+ elasticsearchClient.deleteDocument(record, idAndDoc.getLeft());
+ }
+ }
+ break;
+ case IGNORE:
+ break;
+ case FAIL:
+ elasticsearchClient.failed(new PulsarClientException.InvalidMessageException("Unexpected null message value"));
Review comment:
should `record.fail();` been called here?
--
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-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] nlu90 commented on a change in pull request #11263: Enhanced ElasticSearch Sink (TLS, Auth, KeyValue Schema, INSERT/UPDATE/DELETE, "_id" management, multi index support)
Posted by GitBox <gi...@apache.org>.
nlu90 commented on a change in pull request #11263:
URL: https://github.com/apache/pulsar/pull/11263#discussion_r672699904
##########
File path: pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/JsonConverter.java
##########
@@ -0,0 +1,204 @@
+/**
Review comment:
Make sense to me
--
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-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] sijie commented on pull request #11263: Enhanced ElasticSearch Sink:
Posted by GitBox <gi...@apache.org>.
sijie commented on pull request #11263:
URL: https://github.com/apache/pulsar/pull/11263#issuecomment-876850314
@nlu90 @freeznet Can you review this pull request?
--
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-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org