Update versions of CI

apache · Jun 13, 2024 · 5067eb0 · 5067eb0
1 parent 8e89e7f
commit 5067eb0
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 5 deletions.
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
@@ -47,7 +47,7 @@ jobs:
   compile_and_test:
     strategy:
       matrix:
-        flink: [ 1.18.1, 1.19.0 ]
+        flink: [ 1.19.0 ]
     uses: apache/flink-connector-shared-utils/.github/workflows/ci.yml@ci_utils
     with:
       flink_version: ${{ matrix.flink }}

diff --git a/...-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriterTest.java b/...-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriterTest.java
@@ -25,7 +25,8 @@
 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.InitContext;
+import org.apache.flink.api.connector.sink2.InitContext;
+import org.apache.flink.api.connector.sink2.Sink;
 import org.apache.flink.api.connector.sink2.SinkWriter;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.connector.base.DeliveryGuarantee;
@@ -60,7 +61,9 @@
 
 import java.util.Collection;
 import java.util.List;
+import java.util.Optional;
 import java.util.OptionalLong;
+import java.util.function.Consumer;
 
 import static java.util.Collections.singletonList;
 import static org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
@@ -98,10 +101,10 @@ private void writeMessageAndVerify(
         TopicRouter<String> router = new DynamicTopicRouter<>(configuration, topic);
         PulsarSerializationSchema<String> schema = new PulsarSchemaWrapper<>(STRING);
         FixedMessageDelayer<String> delayer = MessageDelayer.never();
-        MockInitContext initContext = new MockInitContext();
+        Sink.InitContext initContext = new MockInitContext();
 
         PulsarWriter<String> writer =
-                new PulsarWriter<>(
+                new PulsarWriter<String>(
                         configuration,
                         schema,
                         listener,
@@ -163,7 +166,7 @@ public TopicPartition route(
         }
     }
 
-    private static class MockInitContext implements InitContext {
+    private static class MockInitContext implements InitContext, Sink.InitContext {
 
         private final MetricListener metricListener;
         private final OperatorIOMetricGroup ioMetricGroup;
@@ -225,6 +228,11 @@ public <IN> TypeSerializer<IN> createInputSerializer() {
             return null;
         }
 
+        @Override
+        public <MetaT> Optional<Consumer<MetaT>> metadataConsumer() {
+            return Sink.InitContext.super.metadataConsumer();
+        }
+
         // @Override
         public JobID getJobId() {
             return null;