From bcc4e07b574d4e13ee91f3f281ead7cf4343c213 Mon Sep 17 00:00:00 2001 From: GaMalhot <106246924+GaMalhot@users.noreply.github.com> Date: Tue, 6 Feb 2024 16:46:54 +0530 Subject: [PATCH] [issue-730] Upgrade to flink 1.18.1 (#731) Upgrade of Flink version to 1.18.1 on master branch. Changes w.r.t FLINK-32376 for Sink Init Context. Changes w.r.t FLINK-31972 for fixing the ambiguous usage of assertThat(). Signed-off-by: root --- README.md | 3 ++- gradle.properties | 2 +- .../connectors/flink/PravegaCatalogITCase.java | 2 +- .../flink/sink/PravegaSinkWriterITCase.java | 17 +++++++++++++++++ 4 files changed, 21 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 986c1da4..fff07077 100644 --- a/README.md +++ b/README.md @@ -36,7 +36,8 @@ supported versions of Flink and Pravega. | Git Branch | Pravega Version | Flink Version | Status | Artifact Link | |-------------------------------------------------------------------------------------|-----------------|---------------|-------------------|--------------------------------------------------------------------------------------| -| [master](https://github.com/pravega/flink-connectors) | 0.14 | 1.17 | Under Development | https://github.com/pravega/flink-connectors/packages/1441637 | +| [master](https://github.com/pravega/flink-connectors) | 0.14 | 1.18 | Under Development | https://github.com/pravega/flink-connectors/packages/19676441 | +| [r0.14-flink1.17](https://github.com/pravega/flink-connectors/tree/r0.14-flink1.17) | 0.14 | 1.17 | Under Development | https://github.com/pravega/flink-connectors/packages/1441637 | | [r0.14-flink1.16](https://github.com/pravega/flink-connectors/tree/r0.14-flink1.16) | 0.14 | 1.16 | Under Development | https://github.com/pravega/flink-connectors/packages/1704300 | | [r0.13](https://github.com/pravega/flink-connectors/tree/r0.13) | 0.13 | 1.16 | Released | https://repo1.maven.org/maven2/io/pravega/pravega-connectors-flink-1.16_2.12/0.13.0/ | | [r0.13-flink1.15](https://github.com/pravega/flink-connectors/tree/r0.13-flink1.15) | 0.13 | 1.15 | Released | https://repo1.maven.org/maven2/io/pravega/pravega-connectors-flink-1.15_2.12/0.13.0/ | diff --git a/gradle.properties b/gradle.properties index d2b61365..a42d9cca 100644 --- a/gradle.properties +++ b/gradle.properties @@ -17,7 +17,7 @@ # 3rd party Versions. assertjVersion=3.23.1 checkstyleToolVersion=7.1 -flinkVersion=1.17.1 +flinkVersion=1.18.1 flinkScalaVersion=2.12 jacksonVersion=2.8.9 twitterMvnRepoVersion=4.3.4-TWTTR diff --git a/src/test/java/io/pravega/connectors/flink/PravegaCatalogITCase.java b/src/test/java/io/pravega/connectors/flink/PravegaCatalogITCase.java index 0740b7d5..3a9254e3 100644 --- a/src/test/java/io/pravega/connectors/flink/PravegaCatalogITCase.java +++ b/src/test/java/io/pravega/connectors/flink/PravegaCatalogITCase.java @@ -135,7 +135,7 @@ public void testCreateCatalogFromFactory() { assertThat(actualCatalog instanceof PravegaCatalog).isTrue(); assertThat(((PravegaCatalog) actualCatalog).getName()).isEqualTo(CATALOG.getName()); assertThat(((PravegaCatalog) actualCatalog).getDefaultDatabase()).isEqualTo(CATALOG.getDefaultDatabase()); - assertThat(Whitebox.getInternalState(actualCatalog, "properties")) + assertThat((Map) Whitebox.getInternalState(actualCatalog, "properties")) .isEqualTo(Whitebox.getInternalState(CATALOG, "properties")); } diff --git a/src/test/java/io/pravega/connectors/flink/sink/PravegaSinkWriterITCase.java b/src/test/java/io/pravega/connectors/flink/sink/PravegaSinkWriterITCase.java index 3185bf6c..5bccd793 100644 --- a/src/test/java/io/pravega/connectors/flink/sink/PravegaSinkWriterITCase.java +++ b/src/test/java/io/pravega/connectors/flink/sink/PravegaSinkWriterITCase.java @@ -21,10 +21,12 @@ import io.pravega.connectors.flink.utils.runtime.PravegaRuntime; import io.pravega.connectors.flink.utils.runtime.PravegaRuntimeOperator; import org.apache.commons.lang3.RandomStringUtils; +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.time.Time; +import org.apache.flink.api.common.typeutils.TypeSerializer; import org.apache.flink.api.connector.sink2.Sink; import org.apache.flink.api.connector.sink2.SinkWriter; import org.apache.flink.connector.base.DeliveryGuarantee; @@ -148,6 +150,21 @@ private static class SinkInitContext implements Sink.InitContext { this.metricGroup = metricGroup; } + @Override + public JobID getJobId() { + return null; + } + + @Override + public TypeSerializer createInputSerializer() { + return null; + } + + @Override + public boolean isObjectReuseEnabled() { + return false; + } + @Override public UserCodeClassLoader getUserCodeClassLoader() { throw new UnsupportedOperationException("Not implemented.");