Revert "Upgraded ElasticSearch to get rid of CVEs. (apache#13747)"

This reverts commit 421424e.
datastax · Jan 21, 2022 · ded46cd · ded46cd
1 parent 06d0a9d
commit ded46cd
Show file tree

Hide file tree

Showing 9 changed files with 101 additions and 238 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.16.3</elasticsearch.version>
+    <elasticsearch.version>7.9.1</elasticsearch.version>
     <presto.version>334</presto.version>
     <scala.binary.version>2.13</scala.binary.version>
     <scala-library.version>2.13.6</scala-library.version>
@@ -1311,7 +1311,6 @@ 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>
@@ -1819,7 +1818,6 @@ 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,14 +28,6 @@
   <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,12 +68,6 @@
       <version>${project.version}</version>
     </dependency>
 
-    <dependency>
-      <groupId>org.awaitility</groupId>
-      <artifactId>awaitility</artifactId>
-      <scope>test</scope>
-    </dependency>
-
   	<dependency>
   		<groupId>org.elasticsearch.client</groupId>
   		<artifactId>elasticsearch-rest-high-level-client</artifactId>
@@ -90,11 +76,7 @@
     <dependency>
       <groupId>org.testcontainers</groupId>
       <artifactId>elasticsearch</artifactId>
-      <scope>test</scope>
-    </dependency>
-    <dependency>
-      <groupId>org.testcontainers</groupId>
-      <artifactId>toxiproxy</artifactId>
+      <version>1.15.3</version>
       <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
@@ -48,7 +48,6 @@
 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;
@@ -67,11 +66,11 @@
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.ByteSizeUnit;
 import org.elasticsearch.common.unit.ByteSizeValue;
-import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.index.query.QueryBuilders;
 import org.elasticsearch.search.SearchHit;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
-import org.elasticsearch.xcontent.XContentType;
 
 import javax.net.ssl.HostnameVerifier;
 import javax.net.ssl.SSLContext;
@@ -90,7 +89,7 @@
 import java.util.concurrent.atomic.AtomicReference;
 
 @Slf4j
-public class ElasticSearchClient implements AutoCloseable {
+public class ElasticSearchClient {
 
     static final String[] malformedErrors = {
             "mapper_parsing_exception",
@@ -345,7 +344,6 @@ public void flush() {
         bulkProcessor.flush();
     }
 
-    @Override
     public void close() {
         try {
             if (bulkProcessor != null) {
@@ -469,11 +467,6 @@ protected org.elasticsearch.action.search.SearchResponse search(String indexName
                 RequestOptions.DEFAULT);
     }
 
-    @VisibleForTesting
-    protected org.elasticsearch.action.support.master.AcknowledgedResponse delete(String indexName) throws IOException {
-        return client.indices().delete(new DeleteIndexRequest(indexName), RequestOptions.DEFAULT);
-    }
-
     private <T> T retry(Callable<T> callable, String source) {
         try {
             return backoffRetry.retry(callable, config.getMaxRetries(), config.getRetryBackoffInMs(), source);

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
@@ -19,7 +19,7 @@
 package org.apache.pulsar.io.elasticsearch;
 
 import org.elasticsearch.action.bulk.BackoffPolicy;
-import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.common.unit.TimeValue;
 
 import java.util.Iterator;
 import java.util.NoSuchElementException;

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
@@ -35,7 +35,7 @@
 public class ElasticSearchClientSslTests {
 
     public static final String ELASTICSEARCH_IMAGE = Optional.ofNullable(System.getenv("ELASTICSEARCH_IMAGE"))
-            .orElse("docker.elastic.co/elasticsearch/elasticsearch:7.16.3-amd64");
+            .orElse("docker.elastic.co/elasticsearch/elasticsearch:7.10.2-amd64");
 
     final static String INDEX = "myindex";
 
@@ -45,8 +45,9 @@ 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")
-                .withPassword("elastic")
+                .withEnv("ELASTIC_PASSWORD","elastic")  // boostrap password
                 .withEnv("xpack.license.self_generated.type", "trial")
                 .withEnv("xpack.security.enabled", "true")
                 .withEnv("xpack.security.http.ssl.enabled", "true")
@@ -80,8 +81,9 @@ 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")
-                .withPassword("elastic")
+                .withEnv("ELASTIC_PASSWORD","elastic")  // boostrap password
                 .withEnv("xpack.license.self_generated.type", "trial")
                 .withEnv("xpack.security.enabled", "true")
                 .withEnv("xpack.security.http.ssl.enabled", "true")
@@ -118,8 +120,9 @@ 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")
-                .withPassword("elastic")
+                .withEnv("ELASTIC_PASSWORD","elastic")  // boostrap password
                 .withEnv("xpack.license.self_generated.type", "trial")
                 .withEnv("xpack.security.enabled", "true")
                 .withEnv("xpack.security.http.ssl.enabled", "true")