Upgraded ElasticSearch to get rid of CVEs (and switched client to Ope…

…nSearch one) (#13867) * Upgraded ElasticSearch to get rid of CVEs. (#13747) * Upgraded ElasticSearch to get rid of CVEs. CVE-2020-7020 CVE-2020-7021 CVE-2021-22132 CVE-2021-22134 CVE-2021-22144 CVE-2021-22147 * Elastic search client version >= 7.11 no longer works with OSS Elastic images (and elastic.co no longer releases OSS images) * Fixed tests for Elasticsearch * pom cleanup * Switched to OpenSearch client for Elastic (Apache 2 licensed)
apache · Jan 21, 2022 · bef3071 · bef3071
1 parent 662c1c4
commit bef3071
Show file tree

Hide file tree

Showing 12 changed files with 255 additions and 229 deletions.
diff --git a/pom.xml b/pom.xml
@@ -153,7 +153,7 @@ flexible messaging model and an intuitive client API.</description>
     <mariadb-jdbc.version>2.6.0</mariadb-jdbc.version>
     <hdfs-offload-version3>3.3.1</hdfs-offload-version3>
     <json-smart.version>2.4.7</json-smart.version>
-    <elasticsearch.version>7.9.1</elasticsearch.version>
+    <opensearch.version>1.2.4</opensearch.version>
     <presto.version>332</presto.version>
     <scala.binary.version>2.13</scala.binary.version>
     <scala-library.version>2.13.6</scala-library.version>
@@ -1116,9 +1116,9 @@ flexible messaging model and an intuitive client API.</description>
       </dependency>
 
       <dependency>
-        <groupId>org.elasticsearch.client</groupId>
-        <artifactId>elasticsearch-rest-high-level-client</artifactId>
-        <version>${elasticsearch.version}</version>
+        <groupId>org.opensearch.client</groupId>
+        <artifactId>opensearch-rest-high-level-client</artifactId>
+        <version>${opensearch.version}</version>
       </dependency>
 
       <dependency>
@@ -1328,6 +1328,7 @@ flexible messaging model and an intuitive client API.</description>
         </dependencies>
       </plugin>
       <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-compiler-plugin</artifactId>
         <configuration>
           <encoding>UTF-8</encoding>
@@ -1845,6 +1846,7 @@ flexible messaging model and an intuitive client API.</description>
         <pluginManagement>
           <plugins>
             <plugin>
+              <groupId>org.apache.maven.plugins</groupId>
               <artifactId>maven-compiler-plugin</artifactId>
               <configuration>
                 <!-- for some reason, setting maven.compiler.release property alone doesn't work -->

diff --git a/pulsar-io/elastic-search/pom.xml b/pulsar-io/elastic-search/pom.xml
@@ -28,6 +28,14 @@
   <artifactId>pulsar-io-elastic-search</artifactId>
   <name>Pulsar IO :: ElasticSearch</name>
 
+  <properties>
+    <!--
+     Work-around for "Container exited with code 137" (OOM)
+    -->
+    <testReuseFork>false</testReuseFork>
+    <testForkCount>1</testForkCount>
+  </properties>
+
   <dependencies>
 
     <dependency>
@@ -76,14 +84,18 @@
     </dependency>
 
     <dependency>
-      <groupId>org.elasticsearch.client</groupId>
-      <artifactId>elasticsearch-rest-high-level-client</artifactId>
+      <groupId>org.opensearch.client</groupId>
+      <artifactId>opensearch-rest-high-level-client</artifactId>
     </dependency>
 
     <dependency>
       <groupId>org.testcontainers</groupId>
       <artifactId>elasticsearch</artifactId>
-      <version>1.15.3</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.testcontainers</groupId>
+      <artifactId>toxiproxy</artifactId>
       <scope>test</scope>
     </dependency>
 

diff --git a/.../elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClient.java b/.../elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClient.java
@@ -72,35 +72,37 @@
 import org.apache.http.ssl.SSLContexts;
 import org.apache.pulsar.client.api.schema.GenericObject;
 import org.apache.pulsar.functions.api.Record;
-import org.elasticsearch.action.DocWriteRequest;
-import org.elasticsearch.action.DocWriteResponse;
-import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
-import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
-import org.elasticsearch.action.bulk.BulkItemResponse;
-import org.elasticsearch.action.bulk.BulkProcessor;
-import org.elasticsearch.action.bulk.BulkRequest;
-import org.elasticsearch.action.bulk.BulkResponse;
-import org.elasticsearch.action.delete.DeleteRequest;
-import org.elasticsearch.action.delete.DeleteResponse;
-import org.elasticsearch.action.index.IndexRequest;
-import org.elasticsearch.action.index.IndexResponse;
-import org.elasticsearch.action.search.SearchRequest;
-import org.elasticsearch.client.Node;
-import org.elasticsearch.client.RequestOptions;
-import org.elasticsearch.client.Requests;
-import org.elasticsearch.client.RestClient;
-import org.elasticsearch.client.RestClientBuilder;
-import org.elasticsearch.client.RestHighLevelClient;
-import org.elasticsearch.client.indices.CreateIndexRequest;
-import org.elasticsearch.client.indices.CreateIndexResponse;
-import org.elasticsearch.client.indices.GetIndexRequest;
-import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.unit.ByteSizeUnit;
-import org.elasticsearch.common.unit.ByteSizeValue;
-import org.elasticsearch.common.unit.TimeValue;
-import org.elasticsearch.common.xcontent.XContentType;
-import org.elasticsearch.index.query.QueryBuilders;
-import org.elasticsearch.search.builder.SearchSourceBuilder;
+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.builder.SearchSourceBuilder;
 
 @Slf4j
 public class ElasticSearchClient implements AutoCloseable {
@@ -464,7 +466,7 @@ protected long totalHits(String indexName) throws IOException {
     }
 
     @VisibleForTesting
-    protected org.elasticsearch.action.search.SearchResponse search(String indexName) throws IOException {
+    protected SearchResponse search(String indexName) throws IOException {
         client.indices().refresh(new RefreshRequest(indexName), RequestOptions.DEFAULT);
         return client.search(
                 new SearchRequest()
@@ -474,7 +476,7 @@ protected org.elasticsearch.action.search.SearchResponse search(String indexName
     }
 
     @VisibleForTesting
-    protected org.elasticsearch.action.support.master.AcknowledgedResponse delete(String indexName) throws IOException {
+    protected AcknowledgedResponse delete(String indexName) throws IOException {
         return client.indices().delete(new DeleteIndexRequest(indexName), RequestOptions.DEFAULT);
     }
 

diff --git a/...arch/src/main/java/org/apache/pulsar/io/elasticsearch/RandomExponentialBackoffPolicy.java b/...arch/src/main/java/org/apache/pulsar/io/elasticsearch/RandomExponentialBackoffPolicy.java
@@ -20,8 +20,8 @@
 
 import java.util.Iterator;
 import java.util.NoSuchElementException;
-import org.elasticsearch.action.bulk.BackoffPolicy;
-import org.elasticsearch.common.unit.TimeValue;
+import org.opensearch.action.bulk.BackoffPolicy;
+import org.opensearch.common.unit.TimeValue;
 
 public class RandomExponentialBackoffPolicy extends BackoffPolicy {
     private final RandomExponentialRetry randomExponentialRetry;

diff --git a/...-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClientSslTests.java b/...-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClientSslTests.java
@@ -34,7 +34,7 @@
 public class ElasticSearchClientSslTests {
 
     public static final String ELASTICSEARCH_IMAGE = Optional.ofNullable(System.getenv("ELASTICSEARCH_IMAGE"))
-            .orElse("docker.elastic.co/elasticsearch/elasticsearch:7.10.2-amd64");
+            .orElse("docker.elastic.co/elasticsearch/elasticsearch:7.16.3-amd64");
 
     final static String INDEX = "myindex";
 
@@ -44,9 +44,8 @@ public class ElasticSearchClientSslTests {
     @Test
     public void testSslBasic() throws IOException {
         try(ElasticsearchContainer container = new ElasticsearchContainer(ELASTICSEARCH_IMAGE)
-                .withCreateContainerCmdModifier(c -> c.withName("elasticsearch"))
                 .withFileSystemBind(sslResourceDir, configDir + "/ssl")
-                .withEnv("ELASTIC_PASSWORD","elastic")  // boostrap password
+                .withPassword("elastic")
                 .withEnv("xpack.license.self_generated.type", "trial")
                 .withEnv("xpack.security.enabled", "true")
                 .withEnv("xpack.security.http.ssl.enabled", "true")
@@ -80,9 +79,8 @@ public void testSslBasic() throws IOException {
     @Test
     public void testSslWithHostnameVerification() throws IOException {
         try(ElasticsearchContainer container = new ElasticsearchContainer(ELASTICSEARCH_IMAGE)
-                .withCreateContainerCmdModifier(c -> c.withName("elasticsearch"))
                 .withFileSystemBind(sslResourceDir, configDir + "/ssl")
-                .withEnv("ELASTIC_PASSWORD","elastic")  // boostrap password
+                .withPassword("elastic")
                 .withEnv("xpack.license.self_generated.type", "trial")
                 .withEnv("xpack.security.enabled", "true")
                 .withEnv("xpack.security.http.ssl.enabled", "true")
@@ -119,9 +117,8 @@ public void testSslWithHostnameVerification() throws IOException {
     @Test
     public void testSslWithClientAuth() throws IOException {
         try(ElasticsearchContainer container = new ElasticsearchContainer(ELASTICSEARCH_IMAGE)
-                .withCreateContainerCmdModifier(c -> c.withName("elasticsearch"))
                 .withFileSystemBind(sslResourceDir, configDir + "/ssl")
-                .withEnv("ELASTIC_PASSWORD","elastic")  // boostrap password
+                .withPassword("elastic")
                 .withEnv("xpack.license.self_generated.type", "trial")
                 .withEnv("xpack.security.enabled", "true")
                 .withEnv("xpack.security.http.ssl.enabled", "true")