[FLINK-34113] Update flink-connector-elasticsearch to be compatible w…

…ith updated SinkV2 interfaces
apache · Jan 26, 2024 · 0b7eabb · 0b7eabb
1 parent 153b8fc
commit 0b7eabb
Show file tree

Hide file tree

Showing 5 changed files with 34 additions and 97 deletions.
diff --git a/flink-connector-elasticsearch-base/pom.xml b/flink-connector-elasticsearch-base/pom.xml
@@ -131,6 +131,14 @@ under the License.
 			<type>test-jar</type>
 		</dependency>
 
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-connector-base</artifactId>
+			<version>${flink.version}</version>
+			<type>test-jar</type>
+			<scope>test</scope>
+		</dependency>
+
 		<!-- Elasticsearch table descriptor testing -->
 		<dependency>
 			<groupId>org.apache.flink</groupId>

diff --git a/.../java/org/apache/flink/connector/elasticsearch/sink/ElasticsearchSinkBuilderBaseTest.java b/.../java/org/apache/flink/connector/elasticsearch/sink/ElasticsearchSinkBuilderBaseTest.java
@@ -17,26 +17,12 @@
 
 package org.apache.flink.connector.elasticsearch.sink;
 
-import org.apache.flink.api.common.JobID;
-import org.apache.flink.api.common.operators.MailboxExecutor;
-import org.apache.flink.api.common.operators.ProcessingTimeService;
-import org.apache.flink.api.common.serialization.SerializationSchema;
-import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.api.connector.sink2.Sink;
-import org.apache.flink.api.connector.sink2.Sink.InitContext;
 import org.apache.flink.connector.base.DeliveryGuarantee;
+import org.apache.flink.connector.base.sink.writer.TestSinkInitContext;
 import org.apache.flink.connector.elasticsearch.sink.BulkResponseInspector.BulkResponseInspectorFactory;
 import org.apache.flink.connector.elasticsearch.sink.ElasticsearchWriter.DefaultBulkResponseInspector;
 import org.apache.flink.metrics.MetricGroup;
-import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
-import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
-import org.apache.flink.runtime.metrics.groups.InternalSinkWriterMetricGroup;
-import org.apache.flink.streaming.runtime.tasks.TestProcessingTimeService;
-import org.apache.flink.util.FlinkRuntimeException;
-import org.apache.flink.util.SimpleUserCodeClassLoader;
 import org.apache.flink.util.TestLoggerExtension;
-import org.apache.flink.util.UserCodeClassLoader;
-import org.apache.flink.util.function.ThrowingRunnable;
 
 import org.apache.http.HttpHost;
 import org.junit.jupiter.api.DynamicTest;
@@ -45,7 +31,6 @@
 import org.junit.jupiter.api.TestInstance;
 import org.junit.jupiter.api.extension.ExtendWith;
 
-import java.util.OptionalLong;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Stream;
 
@@ -125,7 +110,7 @@ void testOverrideFailureHandler() {
         final ElasticsearchSink<Object> sink =
                 createMinimalBuilder().setFailureHandler(failureHandler).build();
 
-        final InitContext sinkInitContext = new MockInitContext();
+        final TestSinkInitContext sinkInitContext = new TestSinkInitContext();
         final BulkResponseInspector bulkResponseInspector =
                 sink.getBulkResponseInspectorFactory().apply(sinkInitContext::metricGroup);
         assertThat(bulkResponseInspector)
@@ -150,7 +135,7 @@ void testOverrideBulkResponseInspectorFactory() {
                         .setBulkResponseInspectorFactory(bulkResponseInspectorFactory)
                         .build();
 
-        final InitContext sinkInitContext = new MockInitContext();
+        final TestSinkInitContext sinkInitContext = new TestSinkInitContext();
 
         assertThatCode(() -> sink.createWriter(sinkInitContext)).doesNotThrowAnyException();
         assertThat(called).isTrue();
@@ -159,83 +144,4 @@ void testOverrideBulkResponseInspectorFactory() {
     abstract B createEmptyBuilder();
 
     abstract B createMinimalBuilder();
-
-    private static class DummyMailboxExecutor implements MailboxExecutor {
-        private DummyMailboxExecutor() {}
-
-        public void execute(
-                ThrowingRunnable<? extends Exception> command,
-                String descriptionFormat,
-                Object... descriptionArgs) {}
-
-        public void yield() throws InterruptedException, FlinkRuntimeException {}
-
-        public boolean tryYield() throws FlinkRuntimeException {
-            return false;
-        }
-    }
-
-    private static class MockInitContext
-            implements Sink.InitContext, SerializationSchema.InitializationContext {
-
-        public UserCodeClassLoader getUserCodeClassLoader() {
-            return SimpleUserCodeClassLoader.create(
-                    ElasticsearchSinkBuilderBaseTest.class.getClassLoader());
-        }
-
-        public MailboxExecutor getMailboxExecutor() {
-            return new ElasticsearchSinkBuilderBaseTest.DummyMailboxExecutor();
-        }
-
-        public ProcessingTimeService getProcessingTimeService() {
-            return new TestProcessingTimeService();
-        }
-
-        public int getSubtaskId() {
-            return 0;
-        }
-
-        public int getNumberOfParallelSubtasks() {
-            return 0;
-        }
-
-        public int getAttemptNumber() {
-            return 0;
-        }
-
-        public SinkWriterMetricGroup metricGroup() {
-            return InternalSinkWriterMetricGroup.wrap(
-                    new TestingSinkWriterMetricGroup.Builder()
-                            .setIoMetricGroupSupplier(
-                                    UnregisteredMetricsGroup::createOperatorIOMetricGroup)
-                            .setParentMetricGroup(
-                                    UnregisteredMetricsGroup.createOperatorMetricGroup())
-                            .build());
-        }
-
-        public MetricGroup getMetricGroup() {
-            return this.metricGroup();
-        }
-
-        public OptionalLong getRestoredCheckpointId() {
-            return OptionalLong.empty();
-        }
-
-        public SerializationSchema.InitializationContext
-                asSerializationSchemaInitializationContext() {
-            return this;
-        }
-
-        public boolean isObjectReuseEnabled() {
-            return false;
-        }
-
-        public <IN> TypeSerializer<IN> createInputSerializer() {
-            throw new UnsupportedOperationException();
-        }
-
-        public JobID getJobId() {
-            throw new UnsupportedOperationException();
-        }
-    }
 }
diff --git a/flink-connector-elasticsearch6/pom.xml b/flink-connector-elasticsearch6/pom.xml
@@ -121,6 +121,14 @@ under the License.
 			<type>test-jar</type>
 		</dependency>
 
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-connector-base</artifactId>
+			<version>${flink.version}</version>
+			<type>test-jar</type>
+			<scope>test</scope>
+		</dependency>
+
 		<dependency>
 			<groupId>org.apache.flink</groupId>
 			<artifactId>flink-connector-elasticsearch-base</artifactId>

diff --git a/flink-connector-elasticsearch7/pom.xml b/flink-connector-elasticsearch7/pom.xml
@@ -128,6 +128,14 @@ under the License.
 			<type>test-jar</type>
 		</dependency>
 
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-connector-base</artifactId>
+			<version>${flink.version}</version>
+			<type>test-jar</type>
+			<scope>test</scope>
+		</dependency>
+
 		<dependency>
 			<groupId>org.apache.flink</groupId>
 			<artifactId>flink-connector-elasticsearch-base</artifactId>

diff --git a/pom.xml b/pom.xml
@@ -339,6 +339,13 @@ under the License.
 				<version>2.1</version>
 			</dependency>
 
+			<!-- For dependency convergence  -->
+			<dependency>
+				<groupId>org.yaml</groupId>
+				<artifactId>snakeyaml</artifactId>
+				<version>1.31</version>
+			</dependency>
+
 			<!-- For dependency convergence on mockito/powermock mismatch -->
 			<dependency>
 				<groupId>net.bytebuddy</groupId>