Merge branch 'upgrade-flink' into 'master'

Upgrade Flink benchmarks to Flink 1.13

Closes #302

See merge request !226

theodolite-benchmarks/buildSrc/src/main/groovy/theodolite.flink.gradle

@@ -20,7 +20,7 @@ shadowJar {
 tasks.distZip.enabled = false
 
 ext {
-  flinkVersion = '1.12.2'
+  flinkVersion = '1.13.5'
   scalaBinaryVersion = '2.12'
 }
 

theodolite-benchmarks/flink-commons/build.gradle

@@ -3,7 +3,7 @@ plugins {
 }
 
 ext {
-    flinkVersion = '1.12.0'
+    flinkVersion = '1.13.5'
     scalaBinaryVersion = '2.12'
 }
 

theodolite-benchmarks/uc1-flink/Dockerfile

-FROM flink:1.12-scala_2.12-java11
+FROM flink:1.13-java11
 
 ADD build/libs/uc1-flink-all.jar /opt/flink/usrlib/artifacts/uc1-flink-all.jar

theodolite-benchmarks/uc2-flink/Dockerfile

-FROM flink:1.12-scala_2.12-java11
+FROM flink:1.13-java11
 
 ADD build/libs/uc2-flink-all.jar /opt/flink/usrlib/artifacts/uc2-flink-all.jar
\ No newline at end of file

theodolite-benchmarks/uc2-flink/src/main/java/theodolite/uc2/application/HistoryServiceFlinkJob.java

@@ -5,7 +5,6 @@ import org.apache.commons.configuration2.Configuration;
 import org.apache.flink.api.common.typeinfo.Types;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.runtime.state.StateBackend;
-import org.apache.flink.streaming.api.TimeCharacteristic;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
 import org.apache.flink.streaming.api.windowing.time.Time;
@@ -48,8 +47,6 @@ public final class HistoryServiceFlinkJob {
   }
 
   private void configureEnv() {
-    this.env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
-
     final boolean checkpointing = this.config.getBoolean(ConfigurationKeys.CHECKPOINTING, true);
     final int commitIntervalMs = this.config.getInt(ConfigurationKeys.COMMIT_INTERVAL_MS);
     if (checkpointing) {

theodolite-benchmarks/uc3-flink/Dockerfile

-FROM flink:1.12-scala_2.12-java11
+FROM flink:1.13-java11
 
 ADD build/libs/uc3-flink-all.jar /opt/flink/usrlib/artifacts/uc3-flink-all.jar
\ No newline at end of file

theodolite-benchmarks/uc3-flink/src/main/java/theodolite/uc3/application/HistoryServiceFlinkJob.java

@@ -9,7 +9,6 @@ import org.apache.flink.api.common.typeinfo.Types;
 import org.apache.flink.api.java.functions.KeySelector;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.runtime.state.StateBackend;
-import org.apache.flink.streaming.api.TimeCharacteristic;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
 import org.apache.flink.streaming.api.windowing.time.Time;
@@ -55,8 +54,6 @@ public final class HistoryServiceFlinkJob {
   }
 
   private void configureEnv() {
-    this.env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
-
     final boolean checkpointing = this.config.getBoolean(ConfigurationKeys.CHECKPOINTING, true);
     final int commitIntervalMs = this.config.getInt(ConfigurationKeys.COMMIT_INTERVAL_MS);
     if (checkpointing) {

theodolite-benchmarks/uc4-flink/Dockerfile

-FROM flink:1.12-scala_2.12-java11
+FROM flink:1.13-java11
 
 ADD build/libs/uc4-flink-all.jar /opt/flink/usrlib/artifacts/uc4-flink-all.jar
\ No newline at end of file

theodolite-benchmarks/uc4-flink/src/main/java/theodolite/uc4/application/AggregationServiceFlinkJob.java

@@ -9,7 +9,6 @@ import org.apache.flink.api.common.typeinfo.Types;
 import org.apache.flink.api.java.functions.KeySelector;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.runtime.state.StateBackend;
-import org.apache.flink.streaming.api.TimeCharacteristic;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
@@ -69,8 +68,6 @@ public final class AggregationServiceFlinkJob {
   }
 
   private void configureEnv() {
-    this.env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
-
     final boolean checkpointing = this.config.getBoolean(ConfigurationKeys.CHECKPOINTING, true);
     final int commitIntervalMs = this.config.getInt(ConfigurationKeys.COMMIT_INTERVAL_MS);
     if (checkpointing) {