This is an automated email from the ASF dual-hosted git repository. eskabetxe pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/bahir-flink.git
commit 5f287b9abe1025dae55d3e8179c34cebafdaccb6 Author: Joao Boto <[email protected]> AuthorDate: Wed Mar 8 17:21:45 2023 +0100 [BAHIR-308] Bump flink version to 1.16.1 --- .github/workflows/maven-ci.yml | 2 +- .../streaming/connectors/flume/FlumeSinkTest.java | 8 +++---- .../connectors/redis/RedisDescriptorTest.java | 26 ++++++++++++---------- .../flink/streaming/siddhi/SiddhiCEPITCase.java | 4 +++- pom.xml | 2 +- 5 files changed, 22 insertions(+), 20 deletions(-) diff --git a/.github/workflows/maven-ci.yml b/.github/workflows/maven-ci.yml index ef344b9..6488728 100644 --- a/.github/workflows/maven-ci.yml +++ b/.github/workflows/maven-ci.yml @@ -31,7 +31,7 @@ jobs: fail-fast: false matrix: java: ['8', '11'] - flink-version: ['1.15.3'] + flink-version: ['1.16.1'] connector: [ 'flink-connector-activemq', 'flink-connector-akka', diff --git a/flink-connector-flume/src/test/java/org/apache/flink/streaming/connectors/flume/FlumeSinkTest.java b/flink-connector-flume/src/test/java/org/apache/flink/streaming/connectors/flume/FlumeSinkTest.java index c212fdf..f59bde8 100644 --- a/flink-connector-flume/src/test/java/org/apache/flink/streaming/connectors/flume/FlumeSinkTest.java +++ b/flink-connector-flume/src/test/java/org/apache/flink/streaming/connectors/flume/FlumeSinkTest.java @@ -22,9 +22,7 @@ import org.apache.flume.Event; import org.apache.flume.event.EventBuilder; import org.junit.jupiter.api.Test; -import java.nio.charset.Charset; - -import static org.apache.flink.test.util.TestUtils.tryExecute; +import java.nio.charset.StandardCharsets; public class FlumeSinkTest extends FlumeServerTest { @@ -35,7 +33,7 @@ public class FlumeSinkTest extends FlumeServerTest { FlumeEventBuilder<String> flumeEventBuilder = new FlumeEventBuilder<String>() { @Override public Event createFlumeEvent(String value, RuntimeContext ctx) { - return EventBuilder.withBody(value, Charset.forName("UTF-8")); + return EventBuilder.withBody(value, StandardCharsets.UTF_8); } }; @@ -43,7 +41,7 @@ public class FlumeSinkTest extends FlumeServerTest { environment.fromElements("string1", "string2").addSink(flumeSink); - tryExecute(environment, "FlumeTest"); + environment.executeAsync( "FlumeTest"); } } diff --git a/flink-connector-redis/src/test/java/org/apache/flink/streaming/connectors/redis/RedisDescriptorTest.java b/flink-connector-redis/src/test/java/org/apache/flink/streaming/connectors/redis/RedisDescriptorTest.java index abb1b2e..3d424f3 100644 --- a/flink-connector-redis/src/test/java/org/apache/flink/streaming/connectors/redis/RedisDescriptorTest.java +++ b/flink-connector-redis/src/test/java/org/apache/flink/streaming/connectors/redis/RedisDescriptorTest.java @@ -49,7 +49,9 @@ public class RedisDescriptorTest extends RedisITCaseBase { .inStreamingMode() .build(); StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env, settings); - tableEnvironment.registerDataStream("t1", source, "k, v"); + + Table table = tableEnvironment.fromDataStream(source); + tableEnvironment.createTemporaryView("t1", table); /*Redis redis = new Redis() .mode(RedisValidator.REDIS_CLUSTER) @@ -57,17 +59,17 @@ public class RedisDescriptorTest extends RedisITCaseBase { .ttl(100000) .property(RedisValidator.REDIS_NODES, REDIS_HOST+ ":" + REDIS_PORT);*/ - tableEnvironment.executeSql("create table redis " + - "(k string, " + - "v bigint) " + - "with (" + - "'connector.type'='redis'," + - "'redis-mode'='cluster'," + - "'cluster-nodes'='"+String.format("%s:%s",REDIS_HOST, REDIS_PORT)+"'," + - "'command'='INCRBY_EX'," + - "'key.ttl'='100000')"); - - tableEnvironment.executeSql("insert into redis select k, v from t1"); + tableEnvironment.executeSql( + "create table redis (key string, number bigint) " + + "with (" + + "'connector.type'='redis'," + + "'redis-mode'='cluster'," + + "'cluster-nodes'='"+String.format("%s:%s",REDIS_HOST, REDIS_PORT)+"'," + + "'command'='INCRBY_EX'," + + "'key.ttl'='100000'" + + ")"); + + tableEnvironment.executeSql("insert into redis select * from t1"); } @Test diff --git a/flink-library-siddhi/src/test/java/org/apache/flink/streaming/siddhi/SiddhiCEPITCase.java b/flink-library-siddhi/src/test/java/org/apache/flink/streaming/siddhi/SiddhiCEPITCase.java index a0d9935..d59110f 100755 --- a/flink-library-siddhi/src/test/java/org/apache/flink/streaming/siddhi/SiddhiCEPITCase.java +++ b/flink-library-siddhi/src/test/java/org/apache/flink/streaming/siddhi/SiddhiCEPITCase.java @@ -47,8 +47,9 @@ import java.util.LinkedList; import java.util.List; import java.util.Map; +import static org.apache.flink.test.util.TestBaseUtils.compareResultsByLinesInMemory; +import static org.apache.flink.test.util.TestBaseUtils.readAllResultLines; import static org.junit.Assert.assertEquals; - /** * Flink-siddhi library integration test cases */ @@ -402,4 +403,5 @@ public class SiddhiCEPITCase extends AbstractTestBase implements Serializable { output.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE); env.execute(); } + } diff --git a/pom.xml b/pom.xml index 8bfd3ee..9d5ae98 100644 --- a/pom.xml +++ b/pom.xml @@ -113,7 +113,7 @@ <maven.compiler.target>${java.version}</maven.compiler.target> <!-- Flink version --> - <flink.version>1.15.3</flink.version> + <flink.version>1.16.1</flink.version> <scala.binary.version>2.12</scala.binary.version> <scala.version>2.12.8</scala.version>
