[FLINK-24325][connectors/elasticsearch] Remove Elasticsearch 5 from c…

…onnectors
martin-g · Nov 29, 2021 · 9649f45 · 9649f45
1 parent f8f6935
commit 9649f45
Show file tree

Hide file tree

Showing 34 changed files with 304 additions and 2,346 deletions.
diff --git a/flink-connectors/flink-connector-elasticsearch-base/pom.xml b/flink-connectors/flink-connector-elasticsearch-base/pom.xml
@@ -37,7 +37,7 @@ under the License.
 
 	<!-- Allow users to pass custom connector versions -->
 	<properties>
-		<elasticsearch.version>5.3.3</elasticsearch.version>
+		<elasticsearch.version>7.5.1</elasticsearch.version>
 	</properties>
 
 	<dependencies>
@@ -85,6 +85,20 @@ under the License.
 
 		<!-- Tests -->
 
+		<!-- For PrebuiltTransportClient use in test -->
+		<dependency>
+			<groupId>org.elasticsearch.client</groupId>
+			<artifactId>transport</artifactId>
+			<version>${elasticsearch.version}</version>
+			<scope>test</scope>
+		</dependency>
+
+		<dependency>
+			<groupId>org.testcontainers</groupId>
+			<artifactId>elasticsearch</artifactId>
+			<scope>test</scope>
+		</dependency>
+
 		<dependency>
 			<groupId>org.apache.flink</groupId>
 			<artifactId>flink-test-utils</artifactId>

diff --git a/.../java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchApiCallBridge.java b/.../java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchApiCallBridge.java
@@ -100,13 +100,10 @@ void configureBulkProcessorBackoff(
      * Creates a {@link RequestIndexer} that is able to work with {@link BulkProcessor} binary
      * compatible.
      */
-    default RequestIndexer createBulkProcessorIndexer(
+    RequestIndexer createBulkProcessorIndexer(
             BulkProcessor bulkProcessor,
             boolean flushOnCheckpoint,
-            AtomicLong numPendingRequestsRef) {
-        return new PreElasticsearch6BulkProcessorIndexer(
-                bulkProcessor, flushOnCheckpoint, numPendingRequestsRef);
-    }
+            AtomicLong numPendingRequestsRef);
 
     /** Perform any necessary state cleanup. */
     default void cleanup() {

diff --git a/...org/apache/flink/streaming/connectors/elasticsearch/index/AbstractTimeIndexGenerator.java b/...org/apache/flink/streaming/connectors/elasticsearch/index/AbstractTimeIndexGenerator.java
diff --git a/...c/main/java/org/apache/flink/streaming/connectors/elasticsearch/index/IndexGenerator.java b/...c/main/java/org/apache/flink/streaming/connectors/elasticsearch/index/IndexGenerator.java
diff --git a/...in/java/org/apache/flink/streaming/connectors/elasticsearch/index/IndexGeneratorBase.java b/...in/java/org/apache/flink/streaming/connectors/elasticsearch/index/IndexGeneratorBase.java