diff --git a/pom.xml b/pom.xml index 4579486f94803..9d28e848fcc2b 100644 --- a/pom.xml +++ b/pom.xml @@ -156,6 +156,7 @@ flexible messaging model and an intuitive client API. 3.3.1 2.4.7 1.2.4 + 8.1.0 334 2.13 2.13.6 @@ -1121,6 +1122,12 @@ flexible messaging model and an intuitive client API. ${opensearch.version} + + co.elastic.clients + elasticsearch-java + ${elasticsearch-java.version} + + joda-time joda-time diff --git a/pulsar-io/elastic-search/pom.xml b/pulsar-io/elastic-search/pom.xml index 378936d3ba67a..d4fc7fcf6ed5d 100644 --- a/pulsar-io/elastic-search/pom.xml +++ b/pulsar-io/elastic-search/pom.xml @@ -87,6 +87,11 @@ opensearch-rest-high-level-client + + co.elastic.clients + elasticsearch-java + + org.testcontainers elasticsearch diff --git a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClient.java b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClient.java index 9a797212fa128..8ac43a59afb23 100644 --- a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClient.java +++ b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClient.java @@ -19,80 +19,30 @@ 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.opensearch.action.DocWriteRequest; -import org.opensearch.action.DocWriteResponse; -import org.opensearch.action.admin.indices.create.CreateIndexRequest; -import org.opensearch.action.admin.indices.create.CreateIndexResponse; -import org.opensearch.action.admin.indices.delete.DeleteIndexRequest; -import org.opensearch.action.admin.indices.refresh.RefreshRequest; -import org.opensearch.action.bulk.BulkItemResponse; -import org.opensearch.action.bulk.BulkProcessor; -import org.opensearch.action.bulk.BulkRequest; -import org.opensearch.action.bulk.BulkResponse; -import org.opensearch.action.delete.DeleteRequest; -import org.opensearch.action.delete.DeleteResponse; -import org.opensearch.action.index.IndexRequest; -import org.opensearch.action.index.IndexResponse; -import org.opensearch.action.search.SearchRequest; -import org.opensearch.action.search.SearchResponse; -import org.opensearch.action.support.master.AcknowledgedResponse; -import org.opensearch.client.Node; -import org.opensearch.client.RequestOptions; -import org.opensearch.client.Requests; -import org.opensearch.client.RestClient; -import org.opensearch.client.RestClientBuilder; -import org.opensearch.client.RestHighLevelClient; -import org.opensearch.client.indices.GetIndexRequest; -import org.opensearch.common.settings.Settings; -import org.opensearch.common.unit.ByteSizeUnit; -import org.opensearch.common.unit.ByteSizeValue; -import org.opensearch.common.unit.TimeValue; -import org.opensearch.common.xcontent.XContentType; -import org.opensearch.index.query.QueryBuilders; -import org.opensearch.search.SearchHit; -import org.opensearch.search.builder.SearchSourceBuilder; - -import javax.net.ssl.HostnameVerifier; -import javax.net.ssl.SSLContext; -import java.io.File; +import org.apache.pulsar.io.elasticsearch.client.BulkProcessor; +import org.apache.pulsar.io.elasticsearch.client.RestClient; +import org.apache.pulsar.io.elasticsearch.client.RestClientFactory; +import org.apache.pulsar.io.elasticsearch.client.elastic.ElasticSearchJavaRestClient; +import org.apache.pulsar.io.elasticsearch.client.opensearch.OpenSearchHighLevelRestClient; + 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.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.Callable; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; @Slf4j @@ -105,114 +55,56 @@ public class ElasticSearchClient implements AutoCloseable { }; private ElasticSearchConfig config; - private ConfigCallback configCallback; - private RestHighLevelClient client; + private RestClient client; + private final RandomExponentialRetry backoffRetry; final Set indexCache = new HashSet<>(); final Map topicToIndexCache = new HashMap<>(); - final RandomExponentialRetry backoffRetry; - final BulkProcessor bulkProcessor; - final ConcurrentMap, Record> records = new ConcurrentHashMap<>(); + final ConcurrentMap records = new ConcurrentHashMap<>(); final AtomicReference irrecoverableError = new AtomicReference<>(); - final ScheduledExecutorService executorService; + final AtomicLong bulkOperationIdGenerator = new AtomicLong(); - ElasticSearchClient(ElasticSearchConfig elasticSearchConfig) throws MalformedURLException { + public 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(); - } - } + final BulkProcessor.Listener bulkListener = new BulkProcessor.Listener() { + + private Record removeAndGetRecordForOperation(BulkProcessor.BulkOperationRequest operation) { + return records.remove(operation.getOperationId()); + + } + @Override + public void afterBulk(long executionId, List bulkOperationList, + List results) { + if (log.isTraceEnabled()) { + log.trace("Bulk request id={} size={}:", executionId, bulkOperationList.size()); + } + int index = 0; + for (BulkProcessor.BulkOperationResult result: results) { + final Record record = removeAndGetRecordForOperation(bulkOperationList.get(index++)); + if (result.isError()) { + record.fail(); + checkForIrrecoverableError(result); + } else { + record.ack(); } - ) - .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); + @Override + public void afterBulk(long executionId, List bulkOperationList, Throwable throwable) { + log.warn("Bulk request id={} failed:", executionId, throwable); + for (BulkProcessor.BulkOperationRequest operation: bulkOperationList) { + final Record record = removeAndGetRecordForOperation(operation); + record.fail(); + } + } + }; + this.backoffRetry = new RandomExponentialRetry(elasticSearchConfig.getMaxRetryTimeInSec()); + this.client = retry(() -> RestClientFactory.createClient(config, bulkListener), -1, "client creation"); } - void failed(Exception e) throws Exception { + void failed(Exception e) { if (irrecoverableError.compareAndSet(null, e)) { log.error("Irrecoverable error:", e); } @@ -222,24 +114,28 @@ boolean isFailed() { return irrecoverableError.get() != null; } - void hasIrrecoverableError(BulkItemResponse bulkItemResponse) throws Exception { + void checkForIrrecoverableError(BulkProcessor.BulkOperationResult result) { + if (!result.isError()) { + return; + } + final String errorCause = result.getError(); for (String error : malformedErrors) { - if (bulkItemResponse.getFailureMessage().contains(error)) { + if (errorCause.contains(error)) { switch (config.getMalformedDocAction()) { case IGNORE: break; case WARN: log.warn("Ignoring malformed document index={} id={}", - bulkItemResponse.getIndex(), - bulkItemResponse.getId(), - bulkItemResponse.getFailure().getCause()); + result.getIndex(), + result.getDocumentId(), + error); break; case FAIL: log.error("Failure due to the malformed document index={} id={}", - bulkItemResponse.getIndex(), - bulkItemResponse.getId(), - bulkItemResponse.getFailure().getCause()); - failed(bulkItemResponse.getFailure().getCause()); + result.getIndex(), + result.getDocumentId(), + error); + failed(new Exception(error)); break; } } @@ -250,14 +146,19 @@ public void bulkIndex(Record record, Pair idAndDoc) throws Excep 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); + final String documentId = idAndDoc.getLeft(); + final String documentSource = idAndDoc.getRight(); + + final long operationId = bulkOperationIdGenerator.incrementAndGet(); + final BulkProcessor.BulkIndexRequest bulkIndexRequest = BulkProcessor.BulkIndexRequest.builder() + .index(config.getIndexName()) + .documentId(documentId) + .documentSource(documentSource) + .requestId(operationId) + .build(); + + records.put(operationId, record); + client.getBulkProcessor().appendIndexRequest(bulkIndexRequest); } catch(Exception e) { log.debug("index failed id=" + idAndDoc.getLeft(), e); record.fail(); @@ -276,20 +177,18 @@ public boolean indexDocument(Record record, Pair 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)) { + + final String indexName = config.getIndexName(); + final String documentId = idAndDoc.getLeft(); + final String documentSource = idAndDoc.getRight(); + + final boolean createdOrUpdated = client.indexDocument(indexName, documentId, documentSource); + if (createdOrUpdated) { record.ack(); - return true; } else { record.fail(); - return false; } + return createdOrUpdated; } catch (final Exception ex) { log.error("index failed id=" + idAndDoc.getLeft(), ex); record.fail(); @@ -301,14 +200,18 @@ public void bulkDelete(Record 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); + final long operationId = bulkOperationIdGenerator.incrementAndGet(); + final BulkProcessor.BulkDeleteRequest bulkDeleteRequest = BulkProcessor.BulkDeleteRequest.builder() + .index(config.getIndexName()) + .documentId(id) + .requestId(operationId) + .build(); + + records.put(operationId, record); + client.getBulkProcessor().appendDeleteRequest(bulkDeleteRequest); } catch(Exception e) { - log.debug("delete failed id=" + id, e); + log.debug("delete failed id: {}", id, e); record.fail(); throw e; } @@ -325,20 +228,17 @@ public boolean deleteDocument(Record record, String id) throws Ex 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)) { + + + final boolean deleted = client.deleteDocument(config.getIndexName(), id); + if (deleted) { record.ack(); - return true; + } else { + record.fail(); } - record.fail(); - return false; + return deleted; } catch (final Exception ex) { - log.debug("index failed id=" + id, ex); + log.debug("index failed id: {}", id, ex); record.fail(); throw ex; } @@ -348,25 +248,14 @@ public boolean deleteDocument(Record record, String id) throws Ex * Flushes the bulk processor. */ public void flush() { - bulkProcessor.flush(); + client.getBulkProcessor().flush(); } @Override 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); + if (client != null) { + client.close(); + client = null; } } @@ -429,154 +318,33 @@ public String topicToIndexName(String topicName) { } @VisibleForTesting - public boolean createIndexIfNeeded(String indexName) throws IOException { + public boolean createIndexIfNeeded(String indexName) { 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"); + return retry(() -> client.createIndex(indexName), "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"); + public boolean indexExists(final String indexName) { + return retry(() -> client.indexExists(indexName), "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()); - } - return response.getHits().getTotalHits().value; - } - - @VisibleForTesting - protected SearchResponse search(String indexName) throws IOException { - client.indices().refresh(new RefreshRequest(indexName), RequestOptions.DEFAULT); - return client.search( - new SearchRequest() - .indices(indexName) - .source(new SearchSourceBuilder().query(QueryBuilders.matchAllQuery())), - RequestOptions.DEFAULT); - } - - @VisibleForTesting - protected AcknowledgedResponse delete(String indexName) throws IOException { - return client.indices().delete(new DeleteIndexRequest(indexName), RequestOptions.DEFAULT); + private T retry(Callable callable, String source) { + return retry(callable, config.getMaxRetries(), source); } - private T retry(Callable callable, String source) { + private T retry(Callable callable, int maxRetries, String source) { try { - return backoffRetry.retry(callable, config.getMaxRetries(), config.getRetryBackoffInMs(), source); + return backoffRetry.retry(callable, maxRetries, config.getRetryBackoffInMs(), source); } catch (Exception e) { log.error("error in command {} wth retry", source, e); throw new ElasticSearchConnectionException(source + " failed", e); } } - public class ConfigCallback implements RestClientBuilder.HttpClientConfigCallback { - final NHttpClientConnectionManager connectionManager; - final CredentialsProvider credentialsProvider; - - public ConfigCallback() { - this.connectionManager = buildConnectionManager(ElasticSearchClient.this.config); - this.credentialsProvider = buildCredentialsProvider(ElasticSearchClient.this.config); - } - - @Override - public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder builder) { - builder.setMaxConnPerRoute(config.getBulkConcurrentRequests()); - builder.setMaxConnTotal(config.getBulkConcurrentRequests()); - builder.setConnectionManager(connectionManager); - - if (this.credentialsProvider != null) { - builder.setDefaultCredentialsProvider(credentialsProvider); - } - return builder; - } - - public NHttpClientConnectionManager buildConnectionManager(ElasticSearchConfig config) { - try { - IOReactorConfig ioReactorConfig = IOReactorConfig.custom() - .setConnectTimeout(config.getConnectTimeoutInMs()) - .setSoTimeout(config.getSocketTimeoutInMs()) - .build(); - ConnectingIOReactor ioReactor = new DefaultConnectingIOReactor(ioReactorConfig); - PoolingNHttpClientConnectionManager connManager; - if (config.getSsl().isEnabled()) { - ElasticSearchSslConfig sslConfig = config.getSsl(); - HostnameVerifier hostnameVerifier = config.getSsl().isHostnameVerification() - ? SSLConnectionSocketFactory.getDefaultHostnameVerifier() - : new NoopHostnameVerifier(); - String[] cipherSuites = null; - if (!Strings.isNullOrEmpty(sslConfig.getCipherSuites())) { - cipherSuites = sslConfig.getCipherSuites().split(","); - } - String[] protocols = null; - if (!Strings.isNullOrEmpty(sslConfig.getProtocols())) { - protocols = sslConfig.getProtocols().split(","); - } - Registry registry = RegistryBuilder.create() - .register("http", NoopIOSessionStrategy.INSTANCE) - .register("https", new SSLIOSessionStrategy( - buildSslContext(config), - protocols, - cipherSuites, - hostnameVerifier)) - .build(); - connManager = new PoolingNHttpClientConnectionManager(ioReactor, registry); - } else { - connManager = new PoolingNHttpClientConnectionManager(ioReactor); - } - return connManager; - } catch(Exception e) { - throw new ElasticSearchConnectionException(e); - } - } + RestClient getRestClient() { + return client; + } - private SSLContext buildSslContext(ElasticSearchConfig config) throws NoSuchAlgorithmException, KeyManagementException, CertificateException, KeyStoreException, IOException, UnrecoverableKeyException { - ElasticSearchSslConfig sslConfig = config.getSsl(); - SSLContextBuilder sslContextBuilder = SSLContexts.custom(); - if (!Strings.isNullOrEmpty(sslConfig.getProvider())) { - sslContextBuilder.setProvider(sslConfig.getProvider()); - } - if (!Strings.isNullOrEmpty(sslConfig.getProtocols())) { - sslContextBuilder.setProtocol(sslConfig.getProtocols()); - } - if (!Strings.isNullOrEmpty(sslConfig.getTruststorePath()) && !Strings.isNullOrEmpty(sslConfig.getTruststorePassword())) { - sslContextBuilder.loadTrustMaterial(new File(sslConfig.getTruststorePath()), sslConfig.getTruststorePassword().toCharArray()); - } - if (!Strings.isNullOrEmpty(sslConfig.getKeystorePath()) && !Strings.isNullOrEmpty(sslConfig.getKeystorePassword())) { - sslContextBuilder.loadKeyMaterial(new File(sslConfig.getKeystorePath()), - sslConfig.getKeystorePassword().toCharArray(), - sslConfig.getKeystorePassword().toCharArray()); - } - return sslContextBuilder.build(); - } - private CredentialsProvider buildCredentialsProvider(ElasticSearchConfig config) { - if (StringUtils.isEmpty(config.getUsername()) || StringUtils.isEmpty(config.getPassword())) { - return null; - } - CredentialsProvider credentialsProvider = new BasicCredentialsProvider(); - credentialsProvider.setCredentials(AuthScope.ANY, - new UsernamePasswordCredentials(config.getUsername(), config.getPassword())); - return credentialsProvider; - } - } } diff --git a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchConfig.java b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchConfig.java index 2a2710f6bfb0a..7d7e9f921f03a 100644 --- a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchConfig.java +++ b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchConfig.java @@ -258,6 +258,18 @@ public class ElasticSearchConfig implements Serializable { ) private boolean ignoreUnsupportedFields = false; + + @FieldDoc( + required = false, + defaultValue = "AUTO", + help = "Specify compatibility mode with the ElasticSearch cluster. " + + "'AUTO' value will try to auto detect the correct compatibility mode to use. " + + "Use 'ELASTICSEARCH_7' if the target cluster is running ElasticSearch 7 or prior. " + + "Use 'ELASTICSEARCH' if the target cluster is running ElasticSearch 8 or higher. " + + "Use 'OPENSEARCH' if the target cluster is running OpenSearch." + ) + private CompatibilityMode compatibilityMode = CompatibilityMode.AUTO; + public enum MalformedDocAction { IGNORE, WARN, @@ -270,6 +282,13 @@ public enum NullValueAction { FAIL } + public enum CompatibilityMode { + AUTO, + ELASTICSEARCH_7, + ELASTICSEARCH, + OPENSEARCH + } + public static ElasticSearchConfig load(String yamlFile) throws IOException { ObjectMapper mapper = new ObjectMapper(new YAMLFactory()); return mapper.readValue(new File(yamlFile), ElasticSearchConfig.class); diff --git a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/RandomExponentialRetry.java b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/RandomExponentialRetry.java index f8c43798493ab..d51f6442bb4be 100644 --- a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/RandomExponentialRetry.java +++ b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/RandomExponentialRetry.java @@ -58,11 +58,11 @@ public long randomWaitInMs(int attempt, long backoffInMs) { return ThreadLocalRandom.current().nextLong(0, waitInMs(attempt, backoffInMs)); } - protected T retry(Callable function, int maxAttempts, long initialBackoff, String source) throws Exception { + public T retry(Callable function, int maxAttempts, long initialBackoff, String source) throws Exception { return retry(function, maxAttempts, initialBackoff, source, new Time()); } - protected T retry(Callable function, int maxAttempts, long initialBackoff, String source, Time clock) throws Exception { + public T retry(Callable function, int maxAttempts, long initialBackoff, String source, Time clock) throws Exception { Exception lastException = null; for(int i = 0; i < maxAttempts || maxAttempts == -1; i++) { try { @@ -70,7 +70,8 @@ protected T retry(Callable function, int maxAttempts, long initialBackoff } catch (Exception e) { lastException = e; long backoff = randomWaitInMs(i, initialBackoff); - log.info("Trying source={} attempt {}/{} failed, waiting {}ms", source, i, maxAttempts, backoff); + log.info("Executing '{}', attempt {}/{}, next retry in {} ms, caused by: {}", source, i, + maxAttempts, backoff, e.getMessage()); clock.sleep(backoff); } } diff --git a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/client/BulkProcessor.java b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/client/BulkProcessor.java new file mode 100644 index 0000000000000..c3adcb7e1abd5 --- /dev/null +++ b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/client/BulkProcessor.java @@ -0,0 +1,84 @@ +/** + * 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.client; + +import lombok.Builder; +import lombok.Getter; + +import java.io.Closeable; +import java.io.IOException; +import java.util.List; +import java.util.concurrent.TimeUnit; + +/** + * Processor for "bulk" call to the Elastic REST Endpoint. + */ +public interface BulkProcessor extends Closeable { + + @Builder + @Getter + class BulkOperationRequest { + private long operationId; + } + + @Builder + @Getter + class BulkOperationResult { + private String error; + private String index; + private String documentId; + public boolean isError() { + return error != null; + } + } + + interface Listener { + + void afterBulk(long executionId, List bulkOperationList, List results); + + void afterBulk(long executionId, List bulkOperationList, Throwable throwable); + } + + @Builder + @Getter + class BulkIndexRequest { + private long requestId; + private String index; + private String documentId; + private String documentSource; + } + + @Builder + @Getter + class BulkDeleteRequest { + private long requestId; + private String index; + private String documentId; + } + + + void appendIndexRequest(BulkIndexRequest request) throws IOException; + + void appendDeleteRequest(BulkDeleteRequest request) throws IOException; + + void flush(); + + void awaitClose(long timeout, TimeUnit unit) throws InterruptedException; + +} diff --git a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/client/RestClient.java b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/client/RestClient.java new file mode 100644 index 0000000000000..5711e72420774 --- /dev/null +++ b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/client/RestClient.java @@ -0,0 +1,193 @@ +/** + * 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.client; + +import com.google.common.base.Strings; +import org.apache.commons.lang3.StringUtils; +import org.apache.http.auth.AuthScope; +import org.apache.http.auth.UsernamePasswordCredentials; +import org.apache.http.client.CredentialsProvider; +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.io.elasticsearch.ElasticSearchConfig; +import org.apache.pulsar.io.elasticsearch.ElasticSearchConnectionException; +import org.apache.pulsar.io.elasticsearch.ElasticSearchSslConfig; +import org.apache.pulsar.io.elasticsearch.RandomExponentialRetry; +import org.elasticsearch.client.RestClientBuilder; + +import javax.net.ssl.HostnameVerifier; +import javax.net.ssl.SSLContext; +import java.io.Closeable; +import java.io.File; +import java.io.IOException; +import java.net.MalformedURLException; +import java.security.KeyManagementException; +import java.security.KeyStoreException; +import java.security.NoSuchAlgorithmException; +import java.security.UnrecoverableKeyException; +import java.security.cert.CertificateException; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +public abstract class RestClient implements Closeable { + + protected final ElasticSearchConfig config; + protected final ConfigCallback configCallback; + private final ScheduledExecutorService executorService; + + public RestClient(ElasticSearchConfig elasticSearchConfig, BulkProcessor.Listener bulkProcessorListener) throws MalformedURLException { + this.config = elasticSearchConfig; + this.configCallback = new ConfigCallback(); + + // idle+expired connection evictor thread + this.executorService = Executors.newSingleThreadScheduledExecutor(); + this.executorService.scheduleAtFixedRate(() -> { + configCallback.connectionManager.closeExpiredConnections(); + configCallback.connectionManager.closeIdleConnections( + config.getConnectionIdleTimeoutInMs(), TimeUnit.MILLISECONDS); + }, + config.getConnectionIdleTimeoutInMs(), + config.getConnectionIdleTimeoutInMs(), + TimeUnit.MILLISECONDS + ); + } + + public abstract boolean indexExists(String index) throws IOException; + public abstract boolean createIndex(String index) throws IOException; + public abstract boolean deleteIndex(String index) throws IOException; + + public abstract boolean indexDocument(String index, String documentId, String documentSource) throws IOException; + public abstract boolean deleteDocument(String index, String documentId) throws IOException; + + public abstract long totalHits(String index) throws IOException; + + public abstract BulkProcessor getBulkProcessor(); + + public class ConfigCallback implements RestClientBuilder.HttpClientConfigCallback, + org.opensearch.client.RestClientBuilder.HttpClientConfigCallback { + final NHttpClientConnectionManager connectionManager; + final CredentialsProvider credentialsProvider; + + public ConfigCallback() { + this.connectionManager = buildConnectionManager(RestClient.this.config); + this.credentialsProvider = buildCredentialsProvider(RestClient.this.config); + } + + @Override + public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder builder) { + builder.setMaxConnPerRoute(config.getBulkConcurrentRequests()); + builder.setMaxConnTotal(config.getBulkConcurrentRequests()); + builder.setConnectionManager(connectionManager); + + if (this.credentialsProvider != null) { + builder.setDefaultCredentialsProvider(credentialsProvider); + } + return builder; + } + + public NHttpClientConnectionManager buildConnectionManager(ElasticSearchConfig config) { + try { + IOReactorConfig ioReactorConfig = IOReactorConfig.custom() + .setConnectTimeout(config.getConnectTimeoutInMs()) + .setSoTimeout(config.getSocketTimeoutInMs()) + .build(); + ConnectingIOReactor ioReactor = new DefaultConnectingIOReactor(ioReactorConfig); + PoolingNHttpClientConnectionManager connManager; + if (config.getSsl().isEnabled()) { + ElasticSearchSslConfig sslConfig = config.getSsl(); + HostnameVerifier hostnameVerifier = config.getSsl().isHostnameVerification() + ? SSLConnectionSocketFactory.getDefaultHostnameVerifier() + : new NoopHostnameVerifier(); + String[] cipherSuites = null; + if (!Strings.isNullOrEmpty(sslConfig.getCipherSuites())) { + cipherSuites = sslConfig.getCipherSuites().split(","); + } + String[] protocols = null; + if (!Strings.isNullOrEmpty(sslConfig.getProtocols())) { + protocols = sslConfig.getProtocols().split(","); + } + Registry registry = RegistryBuilder.create() + .register("http", NoopIOSessionStrategy.INSTANCE) + .register("https", new SSLIOSessionStrategy( + buildSslContext(config), + protocols, + cipherSuites, + hostnameVerifier)) + .build(); + connManager = new PoolingNHttpClientConnectionManager(ioReactor, registry); + } else { + connManager = new PoolingNHttpClientConnectionManager(ioReactor); + } + return connManager; + } catch(Exception e) { + throw new ElasticSearchConnectionException(e); + } + } + + private SSLContext buildSslContext(ElasticSearchConfig config) throws NoSuchAlgorithmException, KeyManagementException, CertificateException, KeyStoreException, IOException, UnrecoverableKeyException { + ElasticSearchSslConfig sslConfig = config.getSsl(); + SSLContextBuilder sslContextBuilder = SSLContexts.custom(); + if (!Strings.isNullOrEmpty(sslConfig.getProvider())) { + sslContextBuilder.setProvider(sslConfig.getProvider()); + } + if (!Strings.isNullOrEmpty(sslConfig.getProtocols())) { + sslContextBuilder.setProtocol(sslConfig.getProtocols()); + } + if (!Strings.isNullOrEmpty(sslConfig.getTruststorePath()) && !Strings.isNullOrEmpty(sslConfig.getTruststorePassword())) { + sslContextBuilder.loadTrustMaterial(new File(sslConfig.getTruststorePath()), sslConfig.getTruststorePassword().toCharArray()); + } + if (!Strings.isNullOrEmpty(sslConfig.getKeystorePath()) && !Strings.isNullOrEmpty(sslConfig.getKeystorePassword())) { + sslContextBuilder.loadKeyMaterial(new File(sslConfig.getKeystorePath()), + sslConfig.getKeystorePassword().toCharArray(), + sslConfig.getKeystorePassword().toCharArray()); + } + return sslContextBuilder.build(); + } + + private CredentialsProvider buildCredentialsProvider(ElasticSearchConfig config) { + if (StringUtils.isEmpty(config.getUsername()) || StringUtils.isEmpty(config.getPassword())) { + return null; + } + CredentialsProvider credentialsProvider = new BasicCredentialsProvider(); + credentialsProvider.setCredentials(AuthScope.ANY, + new UsernamePasswordCredentials(config.getUsername(), config.getPassword())); + return credentialsProvider; + } + } + + @Override + public void close() { + executorService.shutdown(); + } +} diff --git a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/client/RestClientFactory.java b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/client/RestClientFactory.java new file mode 100644 index 0000000000000..4edef71564eb2 --- /dev/null +++ b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/client/RestClientFactory.java @@ -0,0 +1,104 @@ +/** + * 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.client; + + +import com.fasterxml.jackson.databind.ObjectMapper; +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.StringUtils; +import org.apache.http.client.methods.HttpGet; +import org.apache.pulsar.io.elasticsearch.ElasticSearchConfig; +import org.apache.pulsar.io.elasticsearch.client.elastic.ElasticSearchJavaRestClient; +import org.apache.pulsar.io.elasticsearch.client.opensearch.OpenSearchHighLevelRestClient; +import org.opensearch.client.Request; +import org.opensearch.client.Response; + +import java.io.IOException; +import java.net.MalformedURLException; +import java.util.Map; + +@Slf4j +public class RestClientFactory { + + private static final ObjectMapper MAPPER = new ObjectMapper(); + + public static RestClient createClient(ElasticSearchConfig config, BulkProcessor.Listener bulkListener) throws IOException { + if (config.getCompatibilityMode() == ElasticSearchConfig.CompatibilityMode.ELASTICSEARCH) { + log.info("Found compatibilityMode set to '{}', using the ElasticSearch Java client.", config.getCompatibilityMode()); + return new ElasticSearchJavaRestClient(config, bulkListener); + } else if (config.getCompatibilityMode() == ElasticSearchConfig.CompatibilityMode.ELASTICSEARCH_7 || + config.getCompatibilityMode() == ElasticSearchConfig.CompatibilityMode.OPENSEARCH) { + log.info("Found compatibilityMode set to '{}', using the OpenSearch High Level Rest API Client.", config.getCompatibilityMode()); + return new OpenSearchHighLevelRestClient(config, bulkListener); + } + log.info("Found compatibilityMode set to '{}', will try to auto detect the best client to use.", config.getCompatibilityMode()); + try { + final Map jsonResponse = requestInfo(config); + final boolean useOpenSearchHighLevelClient = useOpenSearchHighLevelClient(jsonResponse); + log.info("useOpenSearchHighLevelClient={}, got info response: {}", useOpenSearchHighLevelClient, jsonResponse); + if (useOpenSearchHighLevelClient) { + return new OpenSearchHighLevelRestClient(config, bulkListener); + } + return new ElasticSearchJavaRestClient(config, bulkListener); + } catch (IOException ioException) { + log.warn("Got error while performing info request to detect Elastic version: {}", + ioException.getMessage()); + throw ioException; + } + } + + private static Map requestInfo(ElasticSearchConfig config) throws IOException { + try (final OpenSearchHighLevelRestClient openSearchHighLevelRestClient = + new OpenSearchHighLevelRestClient(config, null)) { + final Response response = openSearchHighLevelRestClient.getClient().getLowLevelClient() + .performRequest(new Request(HttpGet.METHOD_NAME, "/")); + + return (Map) MAPPER.readValue(response.getEntity().getContent(), Map.class); + } + } + + private static boolean useOpenSearchHighLevelClient(Map jsonResponse) { + final Map versionMap = (Map) jsonResponse.get("version"); + final String distribution = (String) versionMap.get("distribution"); + if (!StringUtils.isBlank(distribution)) { + if (distribution.equals("opensearch")) { + return true; + } + } + final String version = (String) versionMap.get("number"); + if (StringUtils.isBlank(version)) { + return true; + } + final String mainVersion = version.substring(0, version.indexOf(".")); + try { + final int numVersion = Integer.parseInt(mainVersion); + if (numVersion <= 7) { + return true; + } + // For Elastic 8+ use Elastic Java client + return false; + } catch (NumberFormatException nfe) { + log.warn("Not able to parse version: {}", mainVersion, nfe); + return true; + } + } + + + +} diff --git a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/client/elastic/ElasticBulkProcessor.java b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/client/elastic/ElasticBulkProcessor.java new file mode 100644 index 0000000000000..cf72aca4be1f8 --- /dev/null +++ b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/client/elastic/ElasticBulkProcessor.java @@ -0,0 +1,357 @@ +/** + * 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.client.elastic; + +import co.elastic.clients.elasticsearch.ElasticsearchClient; +import co.elastic.clients.elasticsearch.core.BulkRequest; +import co.elastic.clients.elasticsearch.core.BulkResponse; +import co.elastic.clients.elasticsearch.core.IndexRequest; +import co.elastic.clients.elasticsearch.core.bulk.BulkOperation; +import co.elastic.clients.elasticsearch.core.bulk.BulkOperationVariant; +import co.elastic.clients.elasticsearch.core.bulk.DeleteOperation; +import co.elastic.clients.elasticsearch.core.bulk.IndexOperation; +import com.fasterxml.jackson.databind.ObjectMapper; +import io.netty.util.concurrent.DefaultThreadFactory; +import lombok.extern.slf4j.Slf4j; +import org.apache.pulsar.io.elasticsearch.ElasticSearchConfig; +import org.apache.pulsar.io.elasticsearch.RandomExponentialRetry; +import org.apache.pulsar.io.elasticsearch.client.BulkProcessor; + +import java.io.Closeable; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.Callable; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.locks.ReentrantLock; +import java.util.stream.Collectors; + +@Slf4j +public class ElasticBulkProcessor implements BulkProcessor { + private final ElasticSearchConfig config; + private final ElasticsearchClient client; + + private final AtomicLong executionIdGen = new AtomicLong(); + private final int bulkActions; + private final long bulkSize; + private final List pendingOperations = new ArrayList<>(); + private final BulkRequestHandler bulkRequestHandler; + private volatile boolean closed = false; + private final ReentrantLock lock; + private final ExecutorService internalExecutorService; + private ScheduledFuture futureFlushTask; + private final ObjectMapper mapper = new ObjectMapper(); + + public ElasticBulkProcessor(ElasticSearchConfig config, ElasticsearchClient client, Listener listener) { + this.config = config; + this.client = client; + this.lock = new ReentrantLock(); + this.bulkActions = config.getBulkActions(); + this.bulkSize = config.getBulkSizeInMb() * 1024 * 1024; + this.internalExecutorService = Executors.newFixedThreadPool(Math.max(1, config.getBulkConcurrentRequests()), + new DefaultThreadFactory("elastic-bulk-executor")); + this.bulkRequestHandler = new BulkRequestHandler(new RandomExponentialRetry(config.getMaxRetryTimeInSec()), + config.getBulkConcurrentRequests(), listener); + + ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor( + new DefaultThreadFactory("elastic-flush-task")); + if (config.getBulkFlushIntervalInMs() > 0) { + futureFlushTask = executor.scheduleWithFixedDelay(new Flush(), + config.getBulkFlushIntervalInMs(), + config.getBulkFlushIntervalInMs(), + TimeUnit.MILLISECONDS); + } + } + + @Override + public void appendIndexRequest(BulkIndexRequest request) throws IOException { + final Map mapped = mapper.readValue(request.getDocumentSource(), Map.class); + + final IndexOperation indexOperation = new IndexOperation.Builder() + .index(config.getIndexName()) + .id(request.getDocumentId()) + .document(mapped) + .build(); + + long sourceLength = 0; + if (config.getBulkSizeInMb() > 0) { + sourceLength = request.getDocumentSource().getBytes(StandardCharsets.UTF_8).length; + } + add(BulkOperationWithId.indexOperation(indexOperation, request.getRequestId(), sourceLength)); + } + + @Override + public void appendDeleteRequest(BulkDeleteRequest request) { + final DeleteOperation deleteOperation = new DeleteOperation.Builder() + .index(request.getIndex()) + .id(request.getDocumentId()) + .build(); + add(BulkOperationWithId.deleteOperation(deleteOperation, request.getRequestId())); + } + + protected void ensureOpen() { + if (this.closed) { + throw new IllegalStateException("bulk process already closed"); + } + } + + private BulkRequest createBulkRequestAndResetPendingOps() { + final BulkRequest bulkRequest = new BulkRequest.Builder() + .operations(new ArrayList<>(pendingOperations)) + .build(); + this.pendingOperations.clear(); + return bulkRequest; + } + + private void execute(boolean force) { + long executionId; + BulkRequest bulkRequest; + lock.lock(); + try { + ensureOpen(); + if (pendingOperations.isEmpty()) { + return; + } + if (!force && !isOverTheLimit()) { + return; + } + bulkRequest = createBulkRequestAndResetPendingOps(); + executionId = executionIdGen.incrementAndGet(); + } finally { + lock.unlock(); + } + this.execute(bulkRequest, executionId); + } + + private boolean isOverTheLimit() { + if (pendingOperations.isEmpty()) { + return false; + } + if (this.bulkActions > 0 && pendingOperations.size() >= this.bulkActions) { + return true; + } else { + return this.bulkSize > 0L && + pendingOperations.stream().mapToLong(op -> op.getEstimatedSizeInBytes()).sum() >= this.bulkSize; + } + } + + public void flush() { + execute(true); + } + + private void execute(BulkRequest bulkRequest, long executionId) { + this.bulkRequestHandler.execute(bulkRequest, executionId); + } + + private void executeIfNeeded() { + execute(false); + } + + public void add(BulkOperationWithId bulkOperation) { + lock.lock(); + try { + ensureOpen(); + this.pendingOperations.add(bulkOperation); + } finally { + lock.unlock(); + } + executeIfNeeded(); + } + + @Override + public void close() { + try { + awaitClose(0L, TimeUnit.NANOSECONDS); + } catch (InterruptedException var2) { + Thread.currentThread().interrupt(); + } + } + + @Override + public void awaitClose(long timeout, TimeUnit unit) throws InterruptedException { + lock.lock(); + try { + if (this.closed) { + return; + } + if (futureFlushTask != null) { + futureFlushTask.cancel(false); + } + flush(); + bulkRequestHandler.awaitClose(timeout, unit); + closed = true; + } finally { + lock.unlock(); + } + } + + public static class BulkOperationWithId extends BulkOperation { + + /** + * REQUEST_OVERHEAD: https://github.com/elastic/elasticsearch/blob/4b2b3fa7e738009a0a52ed2bf89b4c0c018f7a0c/server/src/main/java/org/elasticsearch/action/bulk/BulkRequest.java#L61 + */ + private static final int REQUEST_OVERHEAD = 50; + + public static BulkOperationWithId indexOperation(IndexOperation indexOperation, + long operationId, + long sourceLength) { + long estimatedSizeInBytes = REQUEST_OVERHEAD + sourceLength; + return new BulkOperationWithId(indexOperation, operationId, estimatedSizeInBytes); + } + + public static BulkOperationWithId deleteOperation(DeleteOperation indexOperation, + long operationId) { + return new BulkOperationWithId(indexOperation, operationId, REQUEST_OVERHEAD); + } + + private final long operationId; + private final long estimatedSizeInBytes; + + public BulkOperationWithId(BulkOperationVariant value, long operationId, long estimatedSizeInBytes) { + super(value); + this.operationId = operationId; + this.estimatedSizeInBytes = estimatedSizeInBytes; + } + + public long getOperationId() { + return operationId; + } + + public long getEstimatedSizeInBytes() { + return estimatedSizeInBytes; + } + } + + class Flush implements Runnable { + Flush() { + } + + public void run() { + if (!closed) { + ElasticBulkProcessor.this.flush(); + } + } + } + + public final class BulkRequestHandler { + private final Listener listener; + private final Semaphore semaphore; + private final RandomExponentialRetry retry; + private final int concurrentRequests; + + BulkRequestHandler(RandomExponentialRetry retry, int concurrentRequests, Listener listener) { + assert concurrentRequests >= 0; + this.concurrentRequests = concurrentRequests; + this.retry = retry; + this.semaphore = new Semaphore(concurrentRequests > 0 ? concurrentRequests : 1); + this.listener = listener; + } + + public void execute(final BulkRequest bulkRequest, final long executionId) { + try { + this.semaphore.acquire(); + } catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + listener.afterBulk(executionId, convertBulkRequest(bulkRequest), ex); + return; + } + + CompletableFuture promise = new CompletableFuture<>(); + + Runnable responseCallable = () -> { + Callable callable = () -> client.bulk(bulkRequest); + + try { + if (log.isDebugEnabled()) { + log.debug("Sending bulk {}", executionId); + } + final BulkResponse bulkResponse = retry.retry(callable, config.getMaxRetries(), config.getRetryBackoffInMs(), + "bulk"); + if (log.isDebugEnabled()) { + log.debug("Sending bulk {} completed", executionId); + } + promise.complete(bulkResponse); + } catch (Throwable ex) { + log.warn("Failed to execute bulk request {}", executionId, ex); + promise.completeExceptionally(ex); + } + }; + internalExecutorService.submit(responseCallable); + + CompletableFuture listenerCalledPromise = new CompletableFuture(); + + promise.thenApply((bulkResponse) -> { + this.semaphore.release(); + listener.afterBulk(executionId, convertBulkRequest(bulkRequest), convertBulkResponse(bulkResponse)); + listenerCalledPromise.complete(null); + return null; + }).exceptionally(ex -> { + this.semaphore.release(); + listener.afterBulk(executionId, convertBulkRequest(bulkRequest), ex); + log.warn("Failed to execute bulk request " + executionId, ex); + listenerCalledPromise.complete(null); + return null; + }); + if (config.getBulkConcurrentRequests() == 0) { + // keep the execution sync in case of non-concurrent bulk requests configuration + listenerCalledPromise.join(); + } + } + + boolean awaitClose(long timeout, TimeUnit unit) throws InterruptedException { + if (this.semaphore.tryAcquire(this.concurrentRequests, timeout, unit)) { + this.semaphore.release(this.concurrentRequests); + return true; + } else { + return false; + } + } + + private List convertBulkRequest(BulkRequest bulkRequest) { + return bulkRequest.operations().stream().map(op -> { + BulkOperationWithId opWithId = (BulkOperationWithId) op; + return BulkOperationRequest.builder() + .operationId(opWithId.getOperationId()) + .build(); + }).collect(Collectors.toList()); + } + + private List convertBulkResponse(BulkResponse bulkResponse) { + return bulkResponse.items().stream().map(responseItem -> { + final String error = responseItem.error() != null ? responseItem.error().type() : null; + return BulkOperationResult.builder() + .error(error) + .index(responseItem.index()) + .documentId(responseItem.id()) + .build(); + }).collect(Collectors.toList()); + } + } + + +} diff --git a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/client/elastic/ElasticSearchJavaRestClient.java b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/client/elastic/ElasticSearchJavaRestClient.java new file mode 100644 index 0000000000000..b156d0dab1c30 --- /dev/null +++ b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/client/elastic/ElasticSearchJavaRestClient.java @@ -0,0 +1,199 @@ +/** + * 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.client.elastic; + +import co.elastic.clients.elasticsearch.ElasticsearchClient; +import co.elastic.clients.elasticsearch._types.ElasticsearchException; +import co.elastic.clients.elasticsearch._types.Result; +import co.elastic.clients.elasticsearch.core.DeleteRequest; +import co.elastic.clients.elasticsearch.core.DeleteResponse; +import co.elastic.clients.elasticsearch.core.IndexRequest; +import co.elastic.clients.elasticsearch.core.IndexResponse; +import co.elastic.clients.elasticsearch.core.SearchRequest; +import co.elastic.clients.elasticsearch.core.SearchResponse; +import co.elastic.clients.elasticsearch.indices.CreateIndexRequest; +import co.elastic.clients.elasticsearch.indices.CreateIndexResponse; +import co.elastic.clients.elasticsearch.indices.DeleteIndexRequest; +import co.elastic.clients.elasticsearch.indices.ExistsRequest; +import co.elastic.clients.elasticsearch.indices.IndexSettings; +import co.elastic.clients.elasticsearch.indices.RefreshRequest; +import co.elastic.clients.json.jackson.JacksonJsonpMapper; +import co.elastic.clients.transport.ElasticsearchTransport; +import co.elastic.clients.transport.rest_client.RestClientTransport; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.annotations.VisibleForTesting; +import lombok.extern.slf4j.Slf4j; +import org.apache.http.HttpHost; +import org.apache.pulsar.io.elasticsearch.ElasticSearchConfig; +import org.apache.pulsar.io.elasticsearch.client.BulkProcessor; +import org.apache.pulsar.io.elasticsearch.client.RestClient; +import org.elasticsearch.client.Node; +import org.elasticsearch.client.RestClientBuilder; + +import java.io.IOException; +import java.net.MalformedURLException; +import java.net.URL; +import java.util.Map; +import java.util.concurrent.TimeUnit; + +@Slf4j +public class ElasticSearchJavaRestClient extends RestClient { + + private final ElasticsearchClient client; + private final ObjectMapper objectMapper = new ObjectMapper(); + private final BulkProcessor bulkProcessor; + + public ElasticSearchJavaRestClient(ElasticSearchConfig elasticSearchConfig, + BulkProcessor.Listener bulkProcessorListener) throws MalformedURLException { + super(elasticSearchConfig, bulkProcessorListener); + + URL url = new URL(config.getElasticSearchUrl()); + log.info("ElasticSearch URL {}", url); + RestClientBuilder builder = org.elasticsearch.client.RestClient.builder(new HttpHost(url.getHost(), url.getPort(), url.getProtocol())) + .setRequestConfigCallback(builder1 -> builder1 + .setContentCompressionEnabled(config.isCompressionEnabled()) + .setConnectionRequestTimeout(config.getConnectionRequestTimeoutInMs()) + .setConnectTimeout(config.getConnectTimeoutInMs()) + .setSocketTimeout(config.getSocketTimeoutInMs())) + .setHttpClientConfigCallback(this.configCallback) + .setFailureListener(new org.elasticsearch.client.RestClient.FailureListener() { + public void onFailure(Node node) { + log.warn("Node host={} failed", node.getHost()); + } + }); + ElasticsearchTransport transport = new RestClientTransport(builder.build(), + new JacksonJsonpMapper()); + this.client = new ElasticsearchClient(transport); + if (elasticSearchConfig.isBulkEnabled()) { + bulkProcessor = new ElasticBulkProcessor(elasticSearchConfig, client, bulkProcessorListener); + } else { + bulkProcessor = null; + } + } + + @Override + public boolean indexExists(String index) throws IOException { + final ExistsRequest request = new ExistsRequest.Builder() + .index(index) + .build(); + return client.indices().exists(request).value(); + } + + @Override + public boolean createIndex(String index) throws IOException { + final CreateIndexRequest createIndexRequest = new CreateIndexRequest.Builder() + .index(index) + .settings(new IndexSettings.Builder() + .numberOfShards(config.getIndexNumberOfShards() + "") + .numberOfReplicas(config.getIndexNumberOfReplicas() + "") + .build() + ) + .build(); + try { + final CreateIndexResponse createIndexResponse = client.indices().create(createIndexRequest); + if ((createIndexResponse.acknowledged() != null && createIndexResponse.acknowledged()) + && createIndexResponse.shardsAcknowledged()) { + return true; + } + throw new IOException("Unable to create index, acknowledged: " + createIndexResponse.acknowledged() + + " shardsAcknowledged: " + createIndexResponse.shardsAcknowledged()); + } catch (ElasticsearchException ex) { + if (ex.response().error().type().contains("resource_already_exists_exception")) { + return false; + } + throw ex; + } + } + + @Override + public boolean deleteIndex(String index) throws IOException { + return client.indices().delete(new DeleteIndexRequest.Builder().index(index).build()).acknowledged(); + } + + @Override + public boolean deleteDocument(String index, String documentId) throws IOException { + final DeleteRequest req = new + DeleteRequest.Builder() + .index(config.getIndexName()) + .id(documentId) + .build(); + + DeleteResponse deleteResponse = client.delete(req); + if (deleteResponse.result().equals(Result.Deleted) || deleteResponse.result().equals(Result.NotFound)) { + return true; + } else { + return false; + } + } + + @Override + public boolean indexDocument(String index, String documentId, String documentSource) throws IOException { + final Map mapped = objectMapper.readValue(documentSource, Map.class); + final IndexRequest indexRequest = new IndexRequest.Builder<>() + .index(config.getIndexName()) + .document(mapped) + .id(documentId) + .build(); + final IndexResponse indexResponse = client.index(indexRequest); + + if (indexResponse.result().equals(Result.Created) || indexResponse.result().equals(Result.Updated)) { + return true; + } else { + return false; + } + } + + @VisibleForTesting + public SearchResponse search(String indexName) throws IOException { + final RefreshRequest refreshRequest = new RefreshRequest.Builder().index(indexName).build(); + client.indices().refresh(refreshRequest); + + return client.search(new SearchRequest.Builder().index(indexName) + .q("*:*") + .build(), Map.class); + } + + @Override + public long totalHits(String indexName) throws IOException { + final SearchResponse searchResponse = search(indexName); + return searchResponse.hits().total().value(); + } + + @Override + public BulkProcessor getBulkProcessor() { + if (bulkProcessor == null) { + throw new IllegalStateException("bulkProcessor not enabled"); + } + return bulkProcessor; + } + + @Override + public void close() { + super.close(); + try { + if (bulkProcessor != null) { + bulkProcessor.awaitClose(5000L, TimeUnit.MILLISECONDS); + } + } catch (InterruptedException e) { + log.warn("Elasticsearch bulk processor close error:", e); + } + client.shutdown(); + + } +} diff --git a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/client/opensearch/OpenSearchHighLevelRestClient.java b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/client/opensearch/OpenSearchHighLevelRestClient.java new file mode 100644 index 0000000000000..9101a3156a6a6 --- /dev/null +++ b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/client/opensearch/OpenSearchHighLevelRestClient.java @@ -0,0 +1,294 @@ +/** + * 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.client.opensearch; + +import co.elastic.clients.elasticsearch.ElasticsearchClient; +import co.elastic.clients.json.jackson.JacksonJsonpMapper; +import co.elastic.clients.transport.ElasticsearchTransport; +import com.google.common.annotations.VisibleForTesting; +import lombok.extern.slf4j.Slf4j; +import org.apache.http.HttpHost; +import org.apache.pulsar.client.api.schema.GenericObject; +import org.apache.pulsar.io.elasticsearch.ElasticSearchConfig; +import org.apache.pulsar.io.elasticsearch.RandomExponentialRetry; +import org.apache.pulsar.io.elasticsearch.client.BulkProcessor; +import org.apache.pulsar.io.elasticsearch.client.RestClient; +import org.elasticsearch.client.Node; +import org.opensearch.action.DocWriteRequest; +import org.opensearch.action.DocWriteResponse; +import org.opensearch.action.admin.indices.delete.DeleteIndexRequest; +import org.opensearch.action.admin.indices.refresh.RefreshRequest; +import org.opensearch.action.bulk.BulkItemResponse; +import org.opensearch.action.bulk.BulkRequest; +import org.opensearch.action.bulk.BulkResponse; +import org.opensearch.action.delete.DeleteRequest; +import org.opensearch.action.delete.DeleteResponse; +import org.opensearch.action.index.IndexRequest; +import org.opensearch.action.index.IndexResponse; +import org.opensearch.action.search.SearchRequest; +import org.opensearch.action.search.SearchResponse; +import org.opensearch.client.RequestOptions; +import org.opensearch.client.Requests; +import org.opensearch.client.RestClientBuilder; +import org.opensearch.client.RestHighLevelClient; +import org.opensearch.client.indices.CreateIndexRequest; +import org.opensearch.client.indices.CreateIndexResponse; +import org.opensearch.client.indices.GetIndexRequest; +import org.opensearch.common.Strings; +import org.opensearch.common.settings.Settings; +import org.opensearch.common.unit.ByteSizeUnit; +import org.opensearch.common.unit.ByteSizeValue; +import org.opensearch.common.unit.TimeValue; +import org.opensearch.common.xcontent.XContentType; +import org.opensearch.index.query.QueryBuilders; +import org.opensearch.search.builder.SearchSourceBuilder; + +import java.io.IOException; +import java.net.MalformedURLException; +import java.net.URL; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; + +@Slf4j +public class OpenSearchHighLevelRestClient extends RestClient implements BulkProcessor { + + private RestHighLevelClient client; + private org.opensearch.action.bulk.BulkProcessor internalBulkProcessor; + private final ConcurrentMap, Long> bulkRequestMappings = new ConcurrentHashMap<>(); + + public OpenSearchHighLevelRestClient(ElasticSearchConfig elasticSearchConfig, BulkProcessor.Listener bulkProcessorListener) throws MalformedURLException { + super(elasticSearchConfig, bulkProcessorListener); + URL url = new URL(config.getElasticSearchUrl()); + log.info("ElasticSearch URL {}", url); + RestClientBuilder builder = org.opensearch.client.RestClient.builder(new HttpHost(url.getHost(), url.getPort(), url.getProtocol())) + .setRequestConfigCallback(builder1 -> builder1 + .setContentCompressionEnabled(config.isCompressionEnabled()) + .setConnectionRequestTimeout(config.getConnectionRequestTimeoutInMs()) + .setConnectTimeout(config.getConnectTimeoutInMs()) + .setSocketTimeout(config.getSocketTimeoutInMs())) + .setHttpClientConfigCallback(this.configCallback) + .setFailureListener(new org.opensearch.client.RestClient.FailureListener() { + public void onFailure(Node node) { + log.warn("Node host={} failed", node.getHost()); + } + }); + client = new RestHighLevelClient(builder); + + if (config.isBulkEnabled()) { + org.opensearch.action.bulk.BulkProcessor.Builder bulkBuilder = org.opensearch.action.bulk.BulkProcessor.builder( + (bulkRequest, bulkResponseActionListener) + -> client.bulkAsync(bulkRequest, RequestOptions.DEFAULT, bulkResponseActionListener), + new org.opensearch.action.bulk.BulkProcessor.Listener() { + + private List + convertBulkRequest(BulkRequest bulkRequest) { + return bulkRequest.requests().stream().map(docWriteRequest -> { + long requestId = bulkRequestMappings.get(docWriteRequest); + return BulkProcessor.BulkOperationRequest.builder() + .operationId(requestId) + .build(); + }).collect(Collectors.toList()); + } + + + private List + convertBulkResponse(BulkResponse bulkRequest) { + return Arrays.asList(bulkRequest.getItems()) + .stream() + .map(itemResponse -> + BulkProcessor.BulkOperationResult.builder() + .error(itemResponse.getFailureMessage()) + .index(itemResponse.getIndex()) + .documentId(itemResponse.getId()) + .build()) + .collect(Collectors.toList()); + } + @Override + public void beforeBulk(long l, BulkRequest bulkRequest) { + } + + @Override + public void afterBulk(long l, BulkRequest bulkRequest, + BulkResponse bulkResponse) { + bulkProcessorListener.afterBulk(l, convertBulkRequest(bulkRequest), + convertBulkResponse(bulkResponse)); + } + + @Override + public void afterBulk(long l, BulkRequest bulkRequest, Throwable throwable) { + bulkProcessorListener.afterBulk(l, convertBulkRequest(bulkRequest), + throwable); + } + } + ) + .setBulkActions(config.getBulkActions()) + .setBulkSize(new ByteSizeValue(config.getBulkSizeInMb(), ByteSizeUnit.MB)) + .setConcurrentRequests(config.getBulkConcurrentRequests()) + .setBackoffPolicy(new RandomExponentialBackoffPolicy( + new RandomExponentialRetry(elasticSearchConfig.getMaxRetryTimeInSec()), + config.getRetryBackoffInMs(), + config.getMaxRetries() + )); + if (config.getBulkFlushIntervalInMs() > 0) { + bulkBuilder.setFlushInterval(new TimeValue(config.getBulkFlushIntervalInMs(), TimeUnit.MILLISECONDS)); + } + this.internalBulkProcessor = bulkBuilder.build(); + } else { + this.internalBulkProcessor = null; + } + + + } + + @Override + public boolean indexExists(String index) throws IOException { + final GetIndexRequest request = new GetIndexRequest(index); + return client.indices().exists(request, RequestOptions.DEFAULT); + } + + @Override + public boolean createIndex(String index) throws IOException { + final CreateIndexRequest cireq = new CreateIndexRequest(index); + cireq.settings(Settings.builder() + .put("index.number_of_shards", config.getIndexNumberOfShards()) + .put("index.number_of_replicas", config.getIndexNumberOfReplicas())); + CreateIndexResponse resp = client.indices().create(cireq, RequestOptions.DEFAULT); + if (!resp.isAcknowledged() || !resp.isShardsAcknowledged()) { + throw new IOException("Unable to create index."); + } + return true; + } + + @Override + public boolean deleteIndex(String index) throws IOException { + return client.indices().delete(new DeleteIndexRequest(index), RequestOptions.DEFAULT).isAcknowledged(); + } + + @Override + public boolean indexDocument(String index, String documentId, String documentSource) throws IOException { + IndexRequest indexRequest = Requests.indexRequest(index); + if (!Strings.isNullOrEmpty(documentId)) { + indexRequest.id(documentId); + } + indexRequest.type(config.getTypeName()); + indexRequest.source(documentSource, XContentType.JSON); + + IndexResponse indexResponse = client.index(indexRequest, RequestOptions.DEFAULT); + if (indexResponse.getResult().equals(DocWriteResponse.Result.CREATED) + || indexResponse.getResult().equals(DocWriteResponse.Result.UPDATED)) { + return true; + } else { + return false; + } + } + + @Override + public boolean deleteDocument(String index, String documentId) throws IOException { + DeleteRequest deleteRequest = Requests.deleteRequest(index); + deleteRequest.id(documentId); + deleteRequest.type(config.getTypeName()); + DeleteResponse deleteResponse = client.delete(deleteRequest, RequestOptions.DEFAULT); + if (log.isDebugEnabled()) { + log.debug("delete result {}", deleteResponse.getResult()); + } + if (deleteResponse.getResult().equals(DocWriteResponse.Result.DELETED) + || deleteResponse.getResult().equals(DocWriteResponse.Result.NOT_FOUND)) { + return true; + } + return false; + } + + @Override + public long totalHits(String indexName) throws IOException { + return search(indexName).getHits().getTotalHits().value; + } + + @VisibleForTesting + public SearchResponse search(String indexName) throws IOException { + client.indices().refresh(new RefreshRequest(indexName), RequestOptions.DEFAULT); + return client.search( + new SearchRequest() + .indices(indexName) + .source(new SearchSourceBuilder().query(QueryBuilders.matchAllQuery())), + RequestOptions.DEFAULT); + } + @Override + public BulkProcessor getBulkProcessor() { + return this; + } + + + @Override + public void appendIndexRequest(BulkProcessor.BulkIndexRequest request) throws IOException { + IndexRequest indexRequest = Requests.indexRequest(request.getIndex()); + if (!Strings.isNullOrEmpty(request.getDocumentId())) { + indexRequest.id(request.getDocumentId()); + } + indexRequest.type(config.getTypeName()); + indexRequest.source(request.getDocumentSource(), XContentType.JSON); + bulkRequestMappings.put(indexRequest, request.getRequestId()); + internalBulkProcessor.add(indexRequest); + } + + @Override + public void appendDeleteRequest(BulkProcessor.BulkDeleteRequest request) throws IOException { + DeleteRequest deleteRequest = Requests.deleteRequest(request.getIndex()); + deleteRequest.id(request.getDocumentId()); + deleteRequest.type(config.getTypeName()); + bulkRequestMappings.put(deleteRequest, request.getRequestId()); + internalBulkProcessor.add(deleteRequest); + } + + @Override + public void flush() { + internalBulkProcessor.flush(); + } + + @Override + public void awaitClose(long timeout, TimeUnit unit) throws InterruptedException { + super.close(); + try { + if (internalBulkProcessor != null) { + internalBulkProcessor.awaitClose(5000L, TimeUnit.MILLISECONDS); + internalBulkProcessor = null; + } + } catch (InterruptedException e) { + log.warn("Elasticsearch bulk processor close error:", e); + } + try { + if (this.client != null) { + this.client.close(); + this.client = null; + } + } catch (IOException e) { + log.warn("Elasticsearch client close error:", e); + } + } + + public RestHighLevelClient getClient() { + return client; + } +} diff --git a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/RandomExponentialBackoffPolicy.java b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/client/opensearch/RandomExponentialBackoffPolicy.java similarity index 88% rename from pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/RandomExponentialBackoffPolicy.java rename to pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/client/opensearch/RandomExponentialBackoffPolicy.java index 3ac1dc23093d2..a4c681c8ae623 100644 --- a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/RandomExponentialBackoffPolicy.java +++ b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/client/opensearch/RandomExponentialBackoffPolicy.java @@ -16,10 +16,12 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.pulsar.io.elasticsearch; +package org.apache.pulsar.io.elasticsearch.client.opensearch; import java.util.Iterator; import java.util.NoSuchElementException; + +import org.apache.pulsar.io.elasticsearch.RandomExponentialRetry; import org.opensearch.action.bulk.BackoffPolicy; import org.opensearch.common.unit.TimeValue; @@ -28,7 +30,8 @@ public class RandomExponentialBackoffPolicy extends BackoffPolicy { private final long start; private final int numberOfElements; - public RandomExponentialBackoffPolicy(RandomExponentialRetry randomExponentialRetry, long start, int numberOfElements) { + public RandomExponentialBackoffPolicy(RandomExponentialRetry randomExponentialRetry, + long start, int numberOfElements) { this.randomExponentialRetry = randomExponentialRetry; assert start >= 0; assert numberOfElements >= -1; @@ -58,10 +61,11 @@ public TimeValue next() { if (!this.hasNext()) { throw new NoSuchElementException("Only up to " + this.numberOfElements + " elements"); } else { - long result = RandomExponentialBackoffPolicy.this.randomExponentialRetry.randomWaitInMs(this.currentlyConsumed, start); + long result = RandomExponentialBackoffPolicy.this.randomExponentialRetry + .randomWaitInMs(this.currentlyConsumed, start); ++this.currentlyConsumed; return TimeValue.timeValueMillis(result); } } } -} +} \ No newline at end of file diff --git a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchBWCTests.java b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchBWCTests.java index 86b9df3229f50..04a92be11b956 100644 --- a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchBWCTests.java +++ b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchBWCTests.java @@ -36,7 +36,7 @@ public void testGenericRecord() throws Exception { String json = "{\"c\":\"1\",\"d\":1,\"e\":{\"a\":\"a\",\"b\":true,\"d\":1.0,\"f\":1.0,\"i\":1,\"l\":10}}"; ElasticSearchSink elasticSearchSink = new ElasticSearchSink(); - elasticSearchSink.open(ImmutableMap.of("elasticSearchUrl", "http://localhost:9200", "schemaEnable", "true"), null); + elasticSearchSink.open(ImmutableMap.of("elasticSearchUrl", "http://localhost:9200", "schemaEnable", "true", "compatibilityMode", "ELASTICSEARCH"), null); Pair pair = elasticSearchSink.extractIdAndDocument(new Record() { @Override public GenericObject getValue() { diff --git a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClientSslTests.java b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClientSslTests.java index ee2d2ba4596d1..e59f247da93f2 100644 --- a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClientSslTests.java +++ b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClientSslTests.java @@ -30,13 +30,17 @@ import static org.testng.Assert.assertTrue; // see https://www.elastic.co/guide/en/elasticsearch/reference/current/security-settings.html#ssl-tls-settings -public class ElasticSearchClientSslTests extends ElasticSearchTestBase { +public abstract class ElasticSearchClientSslTests extends ElasticSearchTestBase { final static String INDEX = "myindex"; final static String sslResourceDir = MountableFile.forClasspathResource("ssl").getFilesystemPath(); final static String configDir = "/usr/share/elasticsearch/config"; + public ElasticSearchClientSslTests(String elasticImageName) { + super(elasticImageName); + } + @Test public void testSslBasic() throws IOException { try (ElasticsearchContainer container = createElasticsearchContainer() diff --git a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClientTests.java b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClientTests.java index f02d6be732506..e807f90fa21ec 100644 --- a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClientTests.java +++ b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClientTests.java @@ -23,16 +23,19 @@ import org.apache.commons.lang3.tuple.Pair; import org.apache.pulsar.client.api.schema.GenericObject; import org.apache.pulsar.functions.api.Record; +import org.apache.pulsar.io.elasticsearch.client.elastic.ElasticSearchJavaRestClient; +import org.apache.pulsar.io.elasticsearch.client.opensearch.OpenSearchHighLevelRestClient; import org.apache.pulsar.io.elasticsearch.testcontainers.ElasticToxiproxiContainer; import org.awaitility.Awaitility; import org.testcontainers.containers.Network; import org.testcontainers.elasticsearch.ElasticsearchContainer; import org.testng.annotations.AfterClass; -import org.testng.annotations.BeforeClass; +import org.testng.annotations.BeforeMethod; import org.testng.annotations.Test; import java.io.IOException; +import java.util.concurrent.TimeUnit; import java.util.UUID; import static org.hamcrest.Matchers.greaterThanOrEqualTo; @@ -45,7 +48,7 @@ import static org.testng.Assert.assertTrue; @Slf4j -public class ElasticSearchClientTests extends ElasticSearchTestBase { +public abstract class ElasticSearchClientTests extends ElasticSearchTestBase { public final static String INDEX = "myindex"; static ElasticsearchContainer container; @@ -53,8 +56,15 @@ public class ElasticSearchClientTests extends ElasticSearchTestBase { static ElasticSearchClient client; static Network network = Network.newNetwork(); - @BeforeClass - public static final void initBeforeClass() throws IOException { + public ElasticSearchClientTests(String elasticImageName) { + super(elasticImageName); + } + + @BeforeMethod(alwaysRun = true) + public void initBeforeClass() throws IOException { + if (container != null) { + return; + } container = createElasticsearchContainer().withNetwork(network); container.start(); @@ -63,6 +73,12 @@ public static final void initBeforeClass() throws IOException { config.setIndexName(INDEX); client = new ElasticSearchClient(config); + if (elasticImageName.equals(OPENSEARCH) || elasticImageName.equals(ELASTICSEARCH_7)) { + assertTrue(client.getRestClient() instanceof OpenSearchHighLevelRestClient); + } else { + assertTrue(client.getRestClient() instanceof ElasticSearchJavaRestClient); + } + } @AfterClass(alwaysRun = true) @@ -98,12 +114,57 @@ public void testIndexDelete() throws Exception { client.indexDocument(mockRecord, Pair.of("1","{ \"a\":1}")); assertEquals(mockRecord.acked, 1); assertEquals(mockRecord.failed, 0); - assertEquals(client.totalHits(INDEX), 1); + assertEquals(client.getRestClient().totalHits(INDEX), 1); client.deleteDocument(mockRecord, "1"); assertEquals(mockRecord.acked, 2); assertEquals(mockRecord.failed, 0); - assertEquals(client.totalHits(INDEX), 0); + assertEquals(client.getRestClient().totalHits(INDEX), 0); + } + + @Test + public void testIndexDeleteWithRetry() throws Exception { + try (ElasticToxiproxiContainer toxiproxy = new ElasticToxiproxiContainer(container, network)) + { + toxiproxy.start(); + + final String index = "indexretry-" + UUID.randomUUID(); + ElasticSearchConfig config = new ElasticSearchConfig() + .setElasticSearchUrl("http://" + toxiproxy.getHttpHostAddress()) + .setIndexName(index) + .setMaxRetries(1000); + + try (ElasticSearchClient client = new ElasticSearchClient(config)) { + try { + assertTrue(client.createIndexIfNeeded(index)); + + log.info("starting the toxic"); + toxiproxy.getProxy().setConnectionCut(false); + toxiproxy.getProxy().toxics().latency("elasticpause", ToxicDirection.DOWNSTREAM, 15000); + toxiproxy.removeToxicAfterDelay("elasticpause", 15000); + + MockRecord mockRecord = new MockRecord<>(); + client.indexDocument(mockRecord, Pair.of("1", "{\"a\":1}")); + Awaitility.await().atMost(20, TimeUnit.SECONDS).untilAsserted(() -> { + assertEquals(mockRecord.acked, 1); + assertEquals(mockRecord.failed, 0); + assertEquals(client.getRestClient().totalHits(index), 1); + }); + + toxiproxy.getProxy().toxics().latency("elasticpause", ToxicDirection.DOWNSTREAM, 15000); + toxiproxy.removeToxicAfterDelay("elasticpause", 15000); + + client.deleteDocument(mockRecord, "1"); + Awaitility.await().atMost(20, TimeUnit.SECONDS).untilAsserted(() -> { + assertEquals(mockRecord.acked, 2); + assertEquals(mockRecord.failed, 0); + assertEquals(client.getRestClient().totalHits(index), 0); + }); + } finally { + client.getRestClient().deleteIndex(index); + } + } + } } @Test @@ -192,7 +253,7 @@ public void testBulkRetry() throws Exception { client.bulkIndex(mockRecord, Pair.of("2", "{\"a\":2}")); assertEquals(mockRecord.acked, 2); assertEquals(mockRecord.failed, 0); - assertEquals(client.totalHits(index), 2); + assertEquals(client.getRestClient().totalHits(index), 2); log.info("starting the toxic"); toxiproxy.getProxy().setConnectionCut(false); @@ -202,14 +263,14 @@ public void testBulkRetry() throws Exception { client.bulkIndex(mockRecord, Pair.of("3", "{\"a\":3}")); assertEquals(mockRecord.acked, 2); assertEquals(mockRecord.failed, 0); - assertEquals(client.totalHits(index), 2); + assertEquals(client.getRestClient().totalHits(index), 2); client.flush(); assertEquals(mockRecord.acked, 3); assertEquals(mockRecord.failed, 0); - assertEquals(client.totalHits(index), 3); + assertEquals(client.getRestClient().totalHits(index), 3); } finally { - client.delete(index); + client.getRestClient().deleteIndex(index); } } } @@ -242,13 +303,13 @@ public void testBulkBlocking() throws Exception { Awaitility.await().untilAsserted(() -> { assertThat("acked record", mockRecord.acked, greaterThanOrEqualTo(4)); assertEquals(mockRecord.failed, 0); - assertThat("totalHits", client.totalHits(index), greaterThanOrEqualTo(4L)); + assertThat("totalHits", client.getRestClient().totalHits(index), greaterThanOrEqualTo(4L)); }); client.flush(); Awaitility.await().untilAsserted(() -> { assertEquals(mockRecord.failed, 0); assertEquals(mockRecord.acked, 5); - assertEquals(client.totalHits(index), 5); + assertEquals(client.getRestClient().totalHits(index), 5); }); log.info("starting the toxic"); @@ -273,7 +334,7 @@ public void testBulkBlocking() throws Exception { assertEquals(client.records.size(), 0); } finally { - client.delete(index); + client.getRestClient().deleteIndex(index); } } } diff --git a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchConfigTests.java b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchConfigTests.java index 68cb2511074fe..78857254ba0c6 100644 --- a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchConfigTests.java +++ b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchConfigTests.java @@ -102,6 +102,8 @@ public final void defaultValueTest() throws IOException { assertNull(config.getSsl().getCipherSuites()); assertEquals(config.getSsl().isHostnameVerification(), true); assertEquals(config.getSsl().getProtocols(), "TLSv1.2"); + + assertEquals(config.getCompatibilityMode(), ElasticSearchConfig.CompatibilityMode.AUTO); } @Test diff --git a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchExtractTests.java b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchExtractTests.java index 7d5342b06fbbc..8d7d77ee635d0 100644 --- a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchExtractTests.java +++ b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchExtractTests.java @@ -90,6 +90,7 @@ public GenericObject getValue() { ElasticSearchSink elasticSearchSink = new ElasticSearchSink(); elasticSearchSink.open(ImmutableMap.of( "elasticSearchUrl", "http://localhost:9200", + "compatibilityMode", "ELASTICSEARCH", "primaryFields","c", "schemaEnable", "true", "keyIgnore", "true"), null); @@ -101,6 +102,7 @@ public GenericObject getValue() { ElasticSearchSink elasticSearchSink2 = new ElasticSearchSink(); elasticSearchSink2.open(ImmutableMap.of( "elasticSearchUrl", "http://localhost:9200", + "compatibilityMode", "ELASTICSEARCH", "primaryFields","c,d", "schemaEnable", "true", "keyIgnore", "true"), null); diff --git a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSinkRawDataTests.java b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSinkRawDataTests.java index 42c918aef2ecd..bfccf97b66989 100644 --- a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSinkRawDataTests.java +++ b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSinkRawDataTests.java @@ -45,10 +45,14 @@ import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertNull; -public class ElasticSearchSinkRawDataTests extends ElasticSearchTestBase { +public abstract class ElasticSearchSinkRawDataTests extends ElasticSearchTestBase { private static ElasticsearchContainer container; + public ElasticSearchSinkRawDataTests(String elasticImageName) { + super(elasticImageName); + } + @Mock protected Record mockRecord; @@ -62,8 +66,11 @@ public class ElasticSearchSinkRawDataTests extends ElasticSearchTestBase { static Schema schema; - @BeforeClass - public static final void initBeforeClass() { + @BeforeMethod(alwaysRun = true) + public final void initBeforeClass() { + if (container != null) { + return; + } container = createElasticsearchContainer(); schema = Schema.BYTES; } diff --git a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSinkTests.java b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSinkTests.java index 3b8b351040cbd..274b80f5e5d0e 100644 --- a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSinkTests.java +++ b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSinkTests.java @@ -29,6 +29,8 @@ import org.apache.pulsar.common.schema.SchemaType; import org.apache.pulsar.functions.api.Record; import org.apache.pulsar.io.core.SinkContext; +import org.apache.pulsar.io.elasticsearch.client.elastic.ElasticSearchJavaRestClient; +import org.apache.pulsar.io.elasticsearch.client.opensearch.OpenSearchHighLevelRestClient; import org.apache.pulsar.io.elasticsearch.data.UserProfile; import org.mockito.Mock; import org.mockito.invocation.InvocationOnMock; @@ -36,7 +38,6 @@ import org.testcontainers.elasticsearch.ElasticsearchContainer; import org.testng.annotations.AfterClass; import org.testng.annotations.AfterMethod; -import org.testng.annotations.BeforeClass; import org.testng.annotations.BeforeMethod; import org.testng.annotations.Test; @@ -49,10 +50,14 @@ import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertNull; -public class ElasticSearchSinkTests extends ElasticSearchTestBase { +public abstract class ElasticSearchSinkTests extends ElasticSearchTestBase { private static ElasticsearchContainer container; + public ElasticSearchSinkTests(String elasticImageName) { + super(elasticImageName); + } + @Mock protected Record mockRecord; @@ -66,8 +71,8 @@ public class ElasticSearchSinkTests extends ElasticSearchTestBase { static GenericSchema genericSchema; static GenericRecord userProfile; - @BeforeClass - public static final void initBeforeClass() { + @BeforeMethod(alwaysRun = true) + public final void initBeforeClass() { container = createElasticsearchContainer(); valueSchema = Schema.JSON(UserProfile.class); @@ -171,8 +176,17 @@ public final void sendKeyIgnoreSingleField() throws Exception { sink.open(map, mockSinkContext); send(1); verify(mockRecord, times(1)).ack(); - assertEquals(sink.getElasticsearchClient().totalHits(index), 1L); - assertEquals(sink.getElasticsearchClient().search(index).getHits().getHits()[0].getId(), "bob"); + assertEquals(sink.getElasticsearchClient().getRestClient().totalHits(index), 1L); + + if (elasticImageName.equals(ELASTICSEARCH_8)) { + final ElasticSearchJavaRestClient restClient = (ElasticSearchJavaRestClient) + sink.getElasticsearchClient().getRestClient(); + assertEquals(restClient.search(index).hits().hits().get(0).id(), "bob"); + } else { + final OpenSearchHighLevelRestClient restClient = (OpenSearchHighLevelRestClient) + sink.getElasticsearchClient().getRestClient(); + assertEquals(restClient.search(index).getHits().getHits()[0].getId(), "bob"); + } } @Test(enabled = true) @@ -184,8 +198,16 @@ public final void sendKeyIgnoreMultipleFields() throws Exception { sink.open(map, mockSinkContext); send(1); verify(mockRecord, times(1)).ack(); - assertEquals(sink.getElasticsearchClient().totalHits(index), 1L); - assertEquals(sink.getElasticsearchClient().search(index).getHits().getHits()[0].getId(), "[\"bob\",\"boby\"]"); + assertEquals(sink.getElasticsearchClient().getRestClient().totalHits(index), 1L); + if (elasticImageName.equals(ELASTICSEARCH_8)) { + final ElasticSearchJavaRestClient restClient = (ElasticSearchJavaRestClient) + sink.getElasticsearchClient().getRestClient(); + assertEquals(restClient.search(index).hits().hits().get(0).id(), "[\"bob\",\"boby\"]"); + } else { + final OpenSearchHighLevelRestClient restClient = (OpenSearchHighLevelRestClient) + sink.getElasticsearchClient().getRestClient(); + assertEquals(restClient.search(index).getHits().getHits()[0].getId(), "[\"bob\",\"boby\"]"); + } } protected final void send(int numRecords) throws Exception { @@ -302,9 +324,9 @@ public Object getNativeObject() { }; } }); - assertEquals(sink.getElasticsearchClient().totalHits(index), 1L); + assertEquals(sink.getElasticsearchClient().getRestClient().totalHits(index), 1L); sink.write(new MockRecordNullValue()); - assertEquals(sink.getElasticsearchClient().totalHits(index), action.equals(ElasticSearchConfig.NullValueAction.DELETE) ? 0L : 1L); + assertEquals(sink.getElasticsearchClient().getRestClient().totalHits(index), action.equals(ElasticSearchConfig.NullValueAction.DELETE) ? 0L : 1L); assertNull(sink.getElasticsearchClient().irrecoverableError.get()); } } diff --git a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchTestBase.java b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchTestBase.java index 34e3fc21abb40..50c519a6a0566 100644 --- a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchTestBase.java +++ b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchTestBase.java @@ -18,19 +18,41 @@ */ package org.apache.pulsar.io.elasticsearch; +import org.testcontainers.containers.Network; import org.testcontainers.elasticsearch.ElasticsearchContainer; +import org.testcontainers.utility.DockerImageName; +import org.testng.annotations.DataProvider; import java.util.Optional; -public class ElasticSearchTestBase { +public abstract class ElasticSearchTestBase { - private static final String ELASTICSEARCH_IMAGE = Optional.ofNullable(System.getenv("ELASTICSEARCH_IMAGE")) + public static final String ELASTICSEARCH_8 = Optional.ofNullable(System.getenv("ELASTICSEARCH_IMAGE_V8")) + .orElse("docker.elastic.co/elasticsearch/elasticsearch:8.1.0"); + + public static final String ELASTICSEARCH_7 = Optional.ofNullable(System.getenv("ELASTICSEARCH_IMAGE_V7")) .orElse("docker.elastic.co/elasticsearch/elasticsearch:7.16.3-amd64"); - protected static ElasticsearchContainer createElasticsearchContainer() { - return new ElasticsearchContainer(ELASTICSEARCH_IMAGE) - .withEnv("ES_JAVA_OPTS", "-Xms128m -Xmx256m"); + public static final String OPENSEARCH = Optional.ofNullable(System.getenv("OPENSEARCH_IMAGE")) + .orElse("opensearchproject/opensearch:1.2.4"); + + protected final String elasticImageName; + public ElasticSearchTestBase(String elasticImageName) { + this.elasticImageName = elasticImageName; } + protected ElasticsearchContainer createElasticsearchContainer() { + if (elasticImageName.equals(OPENSEARCH)) { + DockerImageName dockerImageName = DockerImageName.parse(OPENSEARCH).asCompatibleSubstituteFor("docker.elastic.co/elasticsearch/elasticsearch"); + return new ElasticsearchContainer(dockerImageName) + .withEnv("OPENSEARCH_JAVA_OPTS", "-Xms128m -Xmx256m") + .withEnv("bootstrap.memory_lock", "true") + .withEnv("plugins.security.disabled", "true"); + } + return new ElasticsearchContainer(elasticImageName) + .withEnv("ES_JAVA_OPTS", "-Xms128m -Xmx256m") + .withEnv("xpack.security.enabled", "false") + .withEnv("xpack.security.http.ssl.enabled", "false"); + } } diff --git a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/client/RestClientFactoryTest.java b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/client/RestClientFactoryTest.java new file mode 100644 index 0000000000000..8c9b834f7d6e7 --- /dev/null +++ b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/client/RestClientFactoryTest.java @@ -0,0 +1,47 @@ +/** + * 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.client; + +import lombok.SneakyThrows; +import org.apache.pulsar.io.elasticsearch.ElasticSearchConfig; +import org.apache.pulsar.io.elasticsearch.client.elastic.ElasticSearchJavaRestClient; +import org.apache.pulsar.io.elasticsearch.client.opensearch.OpenSearchHighLevelRestClient; +import org.testng.annotations.Test; + +import static org.testng.Assert.*; + +public class RestClientFactoryTest { + + @Test + @SneakyThrows + public void testCompatibilityMode() { + final ElasticSearchConfig config = new ElasticSearchConfig(); + config.setElasticSearchUrl("http://localhost:9200"); + + config.setCompatibilityMode(ElasticSearchConfig.CompatibilityMode.ELASTICSEARCH_7); + assertTrue(RestClientFactory.createClient(config, null) instanceof OpenSearchHighLevelRestClient); + + config.setCompatibilityMode(ElasticSearchConfig.CompatibilityMode.OPENSEARCH); + assertTrue(RestClientFactory.createClient(config, null) instanceof OpenSearchHighLevelRestClient); + + config.setCompatibilityMode(ElasticSearchConfig.CompatibilityMode.ELASTICSEARCH); + assertTrue(RestClientFactory.createClient(config, null) instanceof ElasticSearchJavaRestClient); + + } +} \ No newline at end of file diff --git a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/opensearch/OpenSearchClientSslTests.java b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/opensearch/OpenSearchClientSslTests.java new file mode 100644 index 0000000000000..92bfd2509ffcf --- /dev/null +++ b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/opensearch/OpenSearchClientSslTests.java @@ -0,0 +1,149 @@ +/** + * 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.opensearch; + +import org.apache.pulsar.io.elasticsearch.ElasticSearchClient; +import org.apache.pulsar.io.elasticsearch.ElasticSearchConfig; +import org.apache.pulsar.io.elasticsearch.ElasticSearchSslConfig; +import org.apache.pulsar.io.elasticsearch.ElasticSearchTestBase; +import org.testcontainers.containers.wait.strategy.Wait; +import org.testcontainers.elasticsearch.ElasticsearchContainer; +import org.testcontainers.utility.MountableFile; +import org.testng.annotations.Test; + +import java.io.IOException; +import java.time.Duration; +import java.util.HashMap; +import java.util.Map; + +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertTrue; + +/*https://opensearch.org/docs/latest/opensearch/install/docker-security/*/ +public class OpenSearchClientSslTests extends ElasticSearchTestBase { + + final static String INDEX = "myindex"; + + final static String sslResourceDir = MountableFile.forClasspathResource("ssl").getFilesystemPath(); + final static String configDir = "/usr/share/opensearch/config"; + + public OpenSearchClientSslTests() { + super(OPENSEARCH); + } + + private static Map sslEnv() { + Map map = new HashMap<>(); + map.put("plugins.security.disabled", "false"); + map.put("plugins.security.ssl.http.enabled", "true"); + + map.put("plugins.security.ssl.http.enabled", "true"); + map.put("plugins.security.ssl.http.pemkey_filepath", configDir + "/ssl/elasticsearch.pem"); + map.put("plugins.security.ssl.http.pemcert_filepath", configDir + "/ssl/elasticsearch.crt"); + map.put("plugins.security.ssl.http.pemtrustedcas_filepath", configDir + "/ssl/cacert.pem"); + map.put("plugins.security.ssl.transport.enabled", "true"); + map.put("plugins.security.ssl.transport.pemkey_filepath", configDir + "/ssl/elasticsearch.pem"); + map.put("plugins.security.ssl.transport.pemcert_filepath", configDir + "/ssl/elasticsearch.crt"); + map.put("plugins.security.ssl.transport.pemtrustedcas_filepath", configDir + "/ssl/cacert.pem"); + return map; + } + + @Test + public void testSslBasic() throws IOException { + try (ElasticsearchContainer container = createElasticsearchContainer() + .withFileSystemBind(sslResourceDir, configDir + "/ssl") + .withEnv(sslEnv()) + .waitingFor(Wait.forLogMessage(".*Node started.*", 1) + .withStartupTimeout(Duration.ofMinutes(2)))) { + container.start(); + + ElasticSearchConfig config = new ElasticSearchConfig() + .setElasticSearchUrl("https://" + container.getHttpHostAddress()) + .setIndexName(INDEX) + .setUsername("admin") + .setPassword("admin") + .setSsl(new ElasticSearchSslConfig() + .setEnabled(true) + .setTruststorePath(sslResourceDir + "/truststore.jks") + .setTruststorePassword("changeit")); + ElasticSearchClient client = new ElasticSearchClient(config); + testIndexExists(client); + } + } + + @Test + public void testSslWithHostnameVerification() throws IOException { + try (ElasticsearchContainer container = createElasticsearchContainer() + .withFileSystemBind(sslResourceDir, configDir + "/ssl") + .withEnv(sslEnv()) + .withEnv("plugins.security.ssl.transport.enforce_hostname_verification", "true") + .waitingFor(Wait.forLogMessage(".*Node started.*", 1) + .withStartupTimeout(Duration.ofMinutes(2)))) { + container.start(); + + ElasticSearchConfig config = new ElasticSearchConfig() + .setElasticSearchUrl("https://" + container.getHttpHostAddress()) + .setIndexName(INDEX) + .setUsername("admin") + .setPassword("admin") + .setSsl(new ElasticSearchSslConfig() + .setEnabled(true) + .setProtocols("TLSv1.2") + .setHostnameVerification(true) + .setTruststorePath(sslResourceDir + "/truststore.jks") + .setTruststorePassword("changeit")); + ElasticSearchClient client = new ElasticSearchClient(config); + testIndexExists(client); + } + } + + @Test + public void testSslWithClientAuth() throws IOException { + try(ElasticsearchContainer container = createElasticsearchContainer() + .withFileSystemBind(sslResourceDir, configDir + "/ssl") + .withEnv(sslEnv()) + .waitingFor(Wait.forLogMessage(".*Node started.*", 1) + .withStartupTimeout(Duration.ofMinutes(3)))) { + container.start(); + + ElasticSearchConfig config = new ElasticSearchConfig() + .setElasticSearchUrl("https://" + container.getHttpHostAddress()) + .setIndexName(INDEX) + .setUsername("admin") + .setPassword("admin") + .setSsl(new ElasticSearchSslConfig() + .setEnabled(true) + .setHostnameVerification(true) + .setTruststorePath(sslResourceDir + "/truststore.jks") + .setTruststorePassword("changeit") + .setKeystorePath(sslResourceDir + "/keystore.jks") + .setKeystorePassword("changeit")); + ElasticSearchClient client = new ElasticSearchClient(config); + testIndexExists(client); + } + } + + + public void testIndexExists(ElasticSearchClient client) throws IOException { + assertFalse(client.indexExists("mynewindex")); + assertTrue(client.createIndexIfNeeded("mynewindex")); + assertTrue(client.indexExists("mynewindex")); + assertFalse(client.createIndexIfNeeded("mynewindex")); + } + +} diff --git a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/opensearch/OpenSearchClientTests.java b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/opensearch/OpenSearchClientTests.java new file mode 100644 index 0000000000000..2b7d72d6d5d54 --- /dev/null +++ b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/opensearch/OpenSearchClientTests.java @@ -0,0 +1,28 @@ +/** + * 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.opensearch; + +import org.apache.pulsar.io.elasticsearch.ElasticSearchClientTests; + +public class OpenSearchClientTests extends ElasticSearchClientTests { + + public OpenSearchClientTests() { + super(OPENSEARCH); + } +} diff --git a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/opensearch/OpenSearchSinkRawDataTests.java b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/opensearch/OpenSearchSinkRawDataTests.java new file mode 100644 index 0000000000000..ce18e6ea2b799 --- /dev/null +++ b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/opensearch/OpenSearchSinkRawDataTests.java @@ -0,0 +1,29 @@ +/** + * 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.opensearch; + +import org.apache.pulsar.io.elasticsearch.ElasticSearchSinkRawDataTests; + +public class OpenSearchSinkRawDataTests extends ElasticSearchSinkRawDataTests { + + public OpenSearchSinkRawDataTests() { + super(OPENSEARCH); + } +} + diff --git a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/opensearch/OpenSearchSinkTests.java b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/opensearch/OpenSearchSinkTests.java new file mode 100644 index 0000000000000..b224c7b69093d --- /dev/null +++ b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/opensearch/OpenSearchSinkTests.java @@ -0,0 +1,28 @@ +/** + * 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.opensearch; + +import org.apache.pulsar.io.elasticsearch.ElasticSearchSinkTests; + +public class OpenSearchSinkTests extends ElasticSearchSinkTests { + + public OpenSearchSinkTests() { + super(OPENSEARCH); + } +} diff --git a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/v7/ElasticSearch7ClientSslTests.java b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/v7/ElasticSearch7ClientSslTests.java new file mode 100644 index 0000000000000..9709b259e381b --- /dev/null +++ b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/v7/ElasticSearch7ClientSslTests.java @@ -0,0 +1,28 @@ +/** + * 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.v7; + +import org.apache.pulsar.io.elasticsearch.ElasticSearchClientSslTests; + +public class ElasticSearch7ClientSslTests extends ElasticSearchClientSslTests { + + public ElasticSearch7ClientSslTests() { + super(ELASTICSEARCH_7); + } +} diff --git a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/v7/ElasticSearch7ClientTests.java b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/v7/ElasticSearch7ClientTests.java new file mode 100644 index 0000000000000..d982ed9826030 --- /dev/null +++ b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/v7/ElasticSearch7ClientTests.java @@ -0,0 +1,28 @@ +/** + * 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.v7; + +import org.apache.pulsar.io.elasticsearch.ElasticSearchClientTests; + +public class ElasticSearch7ClientTests extends ElasticSearchClientTests { + + public ElasticSearch7ClientTests() { + super(ELASTICSEARCH_7); + } +} diff --git a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/v7/ElasticSearch7SinkRawDataTests.java b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/v7/ElasticSearch7SinkRawDataTests.java new file mode 100644 index 0000000000000..a193982ad388e --- /dev/null +++ b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/v7/ElasticSearch7SinkRawDataTests.java @@ -0,0 +1,29 @@ +/** + * 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.v7; + +import org.apache.pulsar.io.elasticsearch.ElasticSearchSinkRawDataTests; + +public class ElasticSearch7SinkRawDataTests extends ElasticSearchSinkRawDataTests { + + public ElasticSearch7SinkRawDataTests() { + super(ELASTICSEARCH_7); + } +} + diff --git a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/v7/ElasticSearch7SinkTests.java b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/v7/ElasticSearch7SinkTests.java new file mode 100644 index 0000000000000..fbd8979ac8023 --- /dev/null +++ b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/v7/ElasticSearch7SinkTests.java @@ -0,0 +1,28 @@ +/** + * 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.v7; + +import org.apache.pulsar.io.elasticsearch.ElasticSearchSinkTests; + +public class ElasticSearch7SinkTests extends ElasticSearchSinkTests { + + public ElasticSearch7SinkTests() { + super(ELASTICSEARCH_7); + } +} diff --git a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/v8/ElasticSearch8ClientSslTests.java b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/v8/ElasticSearch8ClientSslTests.java new file mode 100644 index 0000000000000..25155b0d0a232 --- /dev/null +++ b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/v8/ElasticSearch8ClientSslTests.java @@ -0,0 +1,28 @@ +/** + * 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.v8; + +import org.apache.pulsar.io.elasticsearch.ElasticSearchClientSslTests; + +public class ElasticSearch8ClientSslTests extends ElasticSearchClientSslTests { + + public ElasticSearch8ClientSslTests() { + super(ELASTICSEARCH_8); + } +} diff --git a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/v8/ElasticSearch8ClientTests.java b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/v8/ElasticSearch8ClientTests.java new file mode 100644 index 0000000000000..19546d45f960e --- /dev/null +++ b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/v8/ElasticSearch8ClientTests.java @@ -0,0 +1,28 @@ +/** + * 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.v8; + +import org.apache.pulsar.io.elasticsearch.ElasticSearchClientTests; + +public class ElasticSearch8ClientTests extends ElasticSearchClientTests { + + public ElasticSearch8ClientTests() { + super(ELASTICSEARCH_8); + } +} diff --git a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/v8/ElasticSearch8SinkRawDataTests.java b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/v8/ElasticSearch8SinkRawDataTests.java new file mode 100644 index 0000000000000..09343e947930d --- /dev/null +++ b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/v8/ElasticSearch8SinkRawDataTests.java @@ -0,0 +1,29 @@ +/** + * 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.v8; + +import org.apache.pulsar.io.elasticsearch.ElasticSearchSinkRawDataTests; + +public class ElasticSearch8SinkRawDataTests extends ElasticSearchSinkRawDataTests { + + public ElasticSearch8SinkRawDataTests() { + super(ELASTICSEARCH_8); + } +} + diff --git a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/v8/ElasticSearch8SinkTests.java b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/v8/ElasticSearch8SinkTests.java new file mode 100644 index 0000000000000..b2f293f64fb62 --- /dev/null +++ b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/v8/ElasticSearch8SinkTests.java @@ -0,0 +1,28 @@ +/** + * 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.v8; + +import org.apache.pulsar.io.elasticsearch.ElasticSearchSinkTests; + +public class ElasticSearch8SinkTests extends ElasticSearchSinkTests { + + public ElasticSearch8SinkTests() { + super(ELASTICSEARCH_8); + } +} diff --git a/pulsar-io/elastic-search/src/test/resources/ssl/elasticsearch.pem b/pulsar-io/elastic-search/src/test/resources/ssl/elasticsearch.pem new file mode 100644 index 0000000000000..25b2a0f56cd60 --- /dev/null +++ b/pulsar-io/elastic-search/src/test/resources/ssl/elasticsearch.pem @@ -0,0 +1,28 @@ +-----BEGIN PRIVATE KEY----- +MIIEvwIBADANBgkqhkiG9w0BAQEFAASCBKkwggSlAgEAAoIBAQCGUyeOQB1GUP2t +nvIyNNkr4Y59eM2Ixk5Gu0hFKYgu44fD0bES1JpZcW8M8Phr4mWghgr83cJXku8T +/fNwHmO/r9f9M4BrP555A4yP1AgOO1fWJrhgs57cWpV/e8Hab2logzq6DuyANYjF +iakvcdhN8vgHuxIiexDOkSsMMgfUiq68KjakdGCAo9sCAn3yk5oIuC9jYlEt7iiP +YQwHR9ikTNFGDuZCCRqXfDQLJKZq0e8aoPDZ2CEeOQ100WXWGeKS4TYdG9zH4yYD +kANIF9BTohkoB62i5A4yjcWibZ5KCgYuJPv52VKgEBrFSeGgPEcjpaQ11pNKOoLc +CUXvEgbNAgMBAAECggEAGSDNMYtE7raP/j7RyBqi1daoNUi4CnIuioTWjDmsxDjE +WRmNKpyjE0BpKd6aWnRL1q+8tnk1tfRCzVagh8TnnCBlI3TS/T01hCXp0a40o5Tu +ZZ1wuIZhnY9EkEiyhw54hZinD89i5skdAPczszB06oAdbjTMu66SAlKd3vYtCNFz +M+NQol/l0ajqLsx3ckdoshsWJjU0LE+ParZJ/hNtTve2x0z+zQO+uFYNem5IYtBZ +GTCi81a+7dpVaWm6sZRihc2ZTD+BYIt5VC5Nc7GLcLkRXGKBaFcUrD1bR99CfByW +V600jT0zbQjDoLa/ERK0VEHJU/DIH4NLZNwJO5+FgQKBgQDHyh4oqQe9h0iN+Bd7 +e4b7jG0H6AGmJsYSt3jXTwcFmt7U6fdF31/wg9mG2CdISxFhrnxrdFwaX4aVqfXg +OjDJpfpJrOPSU4MJgufgQQQioNlDglCUgsP7KdMbV13QWHEcpjGRl3XOtN2yNF4F +yV36OjnFdgkNpe4y5ckEHgxJ+QKBgQCsHe2OvYWwjk5GR4DN3w9lkwQ8wOH/GspL +QEaoDuvPZTNmNGxb8ANgHWQ/Gts7MDDkwY0W+ontbLvr5UTXz4wiJp1GRXAigiOZ +4PeeBCyrkByQtcEqxejYFKFzed2Our3cqbQu8q5Wr46a0QSWGdE90NI8rWIxXDq2 +MRBFVOP4dQKBgQCe9TofobUN38gjZKPSVIsmMylApCBDwQ/RLncP38m3dOwcPPbH +eh1MMKtu9SX0B/4RWRGXMSJivWRISczeFY5hshQ8cDlnS5izhZrVuwT+RDn404Mn +Vg92E2XqmA2FSgjdAYHo07gguZi2Q6IXOoryH0d9yxcS69VkW50fEIU4iQKBgQCb +afa84iMalpTVXvFOc/EqLcMwvJYUzPMHWgWy8K47Ok1cJ1AFAEd7/W4skSqOWmh7 +1s14h0gODBXv3rj4Cd+mYqm27zJe5pYQ95N/qpNPYzR38rZ20ff7TT2v0MWfgL25 +x778eYO0oJcq8juq8ar/n1SHF6RHn9kf9FOV1x52lQKBgQCa7AhlAlqS33QdD0Zj +g2UFFckBqKU/MmrlzmNexyJ8lZs+tAWgpie4F8a4JYgqpSstwpu42XQfTkyMK//a +o395VBbb3hrHwVQjNbu1Tpp/Zi4hR5dQzp9HTxqCo364RkwISKXyFUmwRaSnikKm +DZJs75+QGdGSCSpgKlapT0F8Rw== +-----END PRIVATE KEY----- diff --git a/tests/integration/pom.xml b/tests/integration/pom.xml index ca0a0d21fb1e2..6c19c64f106a4 100644 --- a/tests/integration/pom.xml +++ b/tests/integration/pom.xml @@ -149,7 +149,20 @@ org.opensearch.client opensearch-rest-high-level-client test - + + + + co.elastic.clients + elasticsearch-java + test + + + + org.testcontainers + elasticsearch + test + + com.rabbitmq diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/ElasticSearchContainer.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/ElasticSearchContainer.java deleted file mode 100644 index aa743e77b9a4a..0000000000000 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/ElasticSearchContainer.java +++ /dev/null @@ -1,47 +0,0 @@ -/** - * 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.tests.integration.containers; - -import org.testcontainers.containers.wait.strategy.HostPortWaitStrategy; - -public class ElasticSearchContainer extends ChaosContainer { - - public static final String NAME = "ElasticSearch"; - static final Integer[] PORTS = { 9200, 9300 }; - - private static final String IMAGE_NAME = "docker.elastic.co/elasticsearch/elasticsearch:7.13.3"; - - public ElasticSearchContainer(String clusterName) { - super(clusterName, IMAGE_NAME); - } - - @Override - protected void configure() { - super.configure(); - this.withNetworkAliases(NAME) - .withExposedPorts(PORTS) - .withEnv("discovery.type", "single-node") - .withCreateContainerCmdModifier(createContainerCmd -> { - createContainerCmd.withHostName(NAME); - createContainerCmd.withName(clusterName + "-" + NAME); - }) - .waitingFor(new HostPortWaitStrategy()); - } - -} diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/ElasticSearch7SinkTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/ElasticSearch7SinkTester.java new file mode 100644 index 0000000000000..699f99d491350 --- /dev/null +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/ElasticSearch7SinkTester.java @@ -0,0 +1,36 @@ +/** + * 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.tests.integration.io.sinks; + +import org.apache.pulsar.tests.integration.topologies.PulsarCluster; +import org.testcontainers.elasticsearch.ElasticsearchContainer; + +public class ElasticSearch7SinkTester extends ElasticSearchSinkTester { + + public ElasticSearch7SinkTester(boolean schemaEnable) { + super(schemaEnable); + } + + @Override + protected ElasticsearchContainer createSinkService(PulsarCluster cluster) { + return new ElasticsearchContainer("docker.elastic.co/elasticsearch/elasticsearch:7.16.3-amd64") + .withEnv("ES_JAVA_OPTS", "-Xms128m -Xmx256m"); + } + +} diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/ElasticSearch8SinkTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/ElasticSearch8SinkTester.java new file mode 100644 index 0000000000000..e24f23fac89ec --- /dev/null +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/ElasticSearch8SinkTester.java @@ -0,0 +1,38 @@ +/** + * 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.tests.integration.io.sinks; + +import org.apache.pulsar.tests.integration.topologies.PulsarCluster; +import org.testcontainers.elasticsearch.ElasticsearchContainer; + +public class ElasticSearch8SinkTester extends ElasticSearchSinkTester { + + public ElasticSearch8SinkTester(boolean schemaEnable) { + super(schemaEnable); + } + + @Override + protected ElasticsearchContainer createSinkService(PulsarCluster cluster) { + return new ElasticsearchContainer("docker.elastic.co/elasticsearch/elasticsearch:8.1.0") + .withEnv("ES_JAVA_OPTS", "-Xms128m -Xmx256m") + .withEnv("xpack.security.enabled", "false") + .withEnv("xpack.security.http.ssl.enabled", "false"); + } + +} diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/ElasticSearchSinkTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/ElasticSearchSinkTester.java index 490e7bdcf95bf..06aa4174b8aab 100644 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/ElasticSearchSinkTester.java +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/ElasticSearchSinkTester.java @@ -25,6 +25,13 @@ import java.util.LinkedHashMap; import java.util.Map; +import co.elastic.clients.elasticsearch.ElasticsearchClient; +import co.elastic.clients.elasticsearch.core.SearchRequest; +import co.elastic.clients.elasticsearch.core.SearchResponse; +import co.elastic.clients.json.jackson.JacksonJsonpMapper; +import co.elastic.clients.transport.ElasticsearchTransport; +import co.elastic.clients.transport.rest_client.RestClientTransport; +import io.vertx.core.http.RequestOptions; import lombok.AllArgsConstructor; import lombok.Cleanup; import lombok.Data; @@ -35,19 +42,17 @@ import org.apache.pulsar.common.schema.KeyValue; import org.apache.pulsar.common.schema.KeyValueEncodingType; import org.apache.pulsar.common.util.ObjectMapperFactory; -import org.apache.pulsar.tests.integration.containers.ElasticSearchContainer; import org.apache.pulsar.tests.integration.topologies.PulsarCluster; import org.awaitility.Awaitility; -import org.opensearch.action.search.SearchRequest; -import org.opensearch.action.search.SearchResponse; -import org.opensearch.client.RequestOptions; -import org.opensearch.client.RestClient; -import org.opensearch.client.RestClientBuilder; -import org.opensearch.client.RestHighLevelClient; +import org.elasticsearch.client.RestClient; +import org.elasticsearch.client.RestClientBuilder; +import org.testcontainers.elasticsearch.ElasticsearchContainer; -public class ElasticSearchSinkTester extends SinkTester { +public abstract class ElasticSearchSinkTester extends SinkTester { - private RestHighLevelClient elasticClient; + private static final String NAME = "elastic-search"; + + private ElasticsearchClient elasticClient; private boolean schemaEnable; private final Schema> kvSchema; @@ -73,9 +78,9 @@ public Schema getInputTopicSchema() { } public ElasticSearchSinkTester(boolean schemaEnable) { - super(ElasticSearchContainer.NAME, SinkType.ELASTIC_SEARCH); + super(NAME, SinkType.ELASTIC_SEARCH); - sinkConfig.put("elasticSearchUrl", "http://" + ElasticSearchContainer.NAME + ":9200"); + sinkConfig.put("elasticSearchUrl", "http://" + NAME + ":9200"); sinkConfig.put("indexName", "test-index"); this.schemaEnable = schemaEnable; if (schemaEnable) { @@ -89,11 +94,6 @@ public ElasticSearchSinkTester(boolean schemaEnable) { } - @Override - protected ElasticSearchContainer createSinkService(PulsarCluster cluster) { - return new ElasticSearchContainer(cluster.getClusterName()); - } - @Override public void prepareSink() throws Exception { RestClientBuilder builder = RestClient.builder( @@ -101,16 +101,18 @@ public void prepareSink() throws Exception { "localhost", serviceContainer.getMappedPort(9200), "http")); - elasticClient = new RestHighLevelClient(builder); + ElasticsearchTransport transport = new RestClientTransport(builder.build(), + new JacksonJsonpMapper()); + elasticClient = new ElasticsearchClient(transport); } @Override public void validateSinkResult(Map kvs) { - SearchRequest searchRequest = new SearchRequest("test-index"); - Awaitility.await().untilAsserted(() -> { - SearchResponse searchResult = elasticClient.search(searchRequest, RequestOptions.DEFAULT); - assertTrue(searchResult.getHits().getTotalHits().value > 0, searchResult.toString()); + SearchResponse searchResult = elasticClient.search(new SearchRequest.Builder().index("test-index") + .q("*:*") + .build(), Map.class); + assertTrue(searchResult.hits().total().value() > 0, searchResult.toString()); }); } diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/OpenSearchSinkTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/OpenSearchSinkTester.java new file mode 100644 index 0000000000000..5710a438a8152 --- /dev/null +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/OpenSearchSinkTester.java @@ -0,0 +1,81 @@ +/** + * 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.tests.integration.io.sinks; + +import co.elastic.clients.elasticsearch.ElasticsearchClient; +import co.elastic.clients.json.jackson.JacksonJsonpMapper; +import co.elastic.clients.transport.ElasticsearchTransport; +import co.elastic.clients.transport.rest_client.RestClientTransport; +import org.apache.http.HttpHost; +import org.apache.pulsar.tests.integration.topologies.PulsarCluster; +import org.awaitility.Awaitility; +import org.opensearch.action.search.SearchRequest; +import org.opensearch.action.search.SearchResponse; +import org.opensearch.client.RequestOptions; +import org.opensearch.client.RestClient; +import org.opensearch.client.RestClientBuilder; +import org.opensearch.client.RestHighLevelClient; +import org.testcontainers.elasticsearch.ElasticsearchContainer; +import org.testcontainers.utility.DockerImageName; + +import java.util.Map; + +import static org.testng.Assert.assertTrue; + +public class OpenSearchSinkTester extends ElasticSearchSinkTester { + + private RestHighLevelClient elasticClient; + + + public OpenSearchSinkTester(boolean schemaEnable) { + super(schemaEnable); + } + + @Override + protected ElasticsearchContainer createSinkService(PulsarCluster cluster) { + DockerImageName dockerImageName = DockerImageName.parse("opensearchproject/opensearch:1.2.4") + .asCompatibleSubstituteFor("docker.elastic.co/elasticsearch/elasticsearch"); + return new ElasticsearchContainer(dockerImageName) + .withEnv("OPENSEARCH_JAVA_OPTS", "-Xms128m -Xmx256m") + .withEnv("bootstrap.memory_lock", "true") + .withEnv("plugins.security.disabled", "true"); + } + + @Override + public void prepareSink() throws Exception { + RestClientBuilder builder = RestClient.builder( + new HttpHost( + "localhost", + serviceContainer.getMappedPort(9200), + "http")); + elasticClient = new RestHighLevelClient(builder); + } + + @Override + public void validateSinkResult(Map kvs) { + org.opensearch.action.search.SearchRequest searchRequest = new SearchRequest("test-index"); + + Awaitility.await().untilAsserted(() -> { + SearchResponse searchResult = elasticClient.search(searchRequest, RequestOptions.DEFAULT); + assertTrue(searchResult.getHits().getTotalHits().value > 0, searchResult.toString()); + }); + } + + +} diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/PulsarSinksTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/PulsarSinksTest.java index f0ca3155abc95..8dbef586665b5 100644 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/PulsarSinksTest.java +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/PulsarSinksTest.java @@ -55,13 +55,33 @@ public void testJdbcSink() throws Exception { } @Test(groups = "sink") - public void testElasticSearchSinkRawData() throws Exception { - testSink(new ElasticSearchSinkTester(false), true); + public void testElasticSearch7SinkRawData() throws Exception { + testSink(new ElasticSearch7SinkTester(false), true); } @Test(groups = "sink") - public void testElasticSearchSinkSchemaEnabled() throws Exception { - testSink(new ElasticSearchSinkTester(true), true); + public void testElasticSearchSink7SchemaEnabled() throws Exception { + testSink(new ElasticSearch7SinkTester(true), true); + } + + @Test(groups = "sink") + public void testElasticSearch8SinkRawData() throws Exception { + testSink(new ElasticSearch8SinkTester(false), true); + } + + @Test(groups = "sink") + public void testElasticSearch8SinkSchemaEnabled() throws Exception { + testSink(new ElasticSearch8SinkTester(true), true); + } + + @Test(groups = "sink") + public void testOpenSearchSinkRawData() throws Exception { + testSink(new OpenSearchSinkTester(false), true); + } + + @Test(groups = "sink") + public void testOpenSearchSinkSchemaEnabled() throws Exception { + testSink(new OpenSearchSinkTester(true), true); } @Test(enabled = false, groups = "sink")