Merge branch 'master' into theodolite-kotlin

theodolite-benchmarks/uc1-flink/src/main/java/theodolite/uc1/application/ConfigurationKeys.java

@@ -19,6 +19,8 @@ public final class ConfigurationKeys {
 
   public static final String CHECKPOINTING = "checkpointing";
 
+  public static final String PARALLELISM = "parallelism";
+
   private ConfigurationKeys() {}
 
 }

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

 package theodolite.uc1.application;
 
 import org.apache.commons.configuration2.Configuration;
+import org.apache.flink.api.common.typeinfo.Types;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
@@ -42,6 +43,14 @@ public final class HistoryServiceFlinkJob {
     if (checkpointing) {
       this.env.enableCheckpointing(commitIntervalMs);
     }
+
+    // Parallelism
+    final Integer parallelism = this.config.getInteger(ConfigurationKeys.PARALLELISM, null);
+    if (parallelism != null) {
+      LOGGER.error("Set parallelism: {}.", parallelism);
+      this.env.setParallelism(parallelism);
+    }
+
   }
 
   private void buildPipeline() {
@@ -61,7 +70,8 @@ public final class HistoryServiceFlinkJob {
     stream
         .rebalance()
         .map(new GsonMapper())
-        .flatMap((record, c) -> LOGGER.info("Record: {}", record));
+        .flatMap((record, c) -> LOGGER.info("Record: {}", record))
+        .returns(Types.GENERIC(Object.class)); // Will never be used
   }
 
   /**

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

@@ -30,6 +30,8 @@ public final class ConfigurationKeys {
 
   public static final String CHECKPOINTING = "checkpointing";
 
+  public static final String PARALLELISM = "parallelism";
+
   private ConfigurationKeys() {}
 
 }

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

@@ -56,6 +56,13 @@ public final class HistoryServiceFlinkJob {
       this.env.enableCheckpointing(commitIntervalMs);
     }
 
+    // Parallelism
+    final Integer parallelism = this.config.getInteger(ConfigurationKeys.PARALLELISM, null);
+    if (parallelism != null) {
+      LOGGER.error("Set parallelism: {}.", parallelism);
+      this.env.setParallelism(parallelism);
+    }
+
     // State Backend
     final StateBackend stateBackend = StateBackends.fromConfiguration(this.config);
     this.env.setStateBackend(stateBackend);

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

@@ -34,6 +34,8 @@ public final class ConfigurationKeys {
 
   public static final String CHECKPOINTING = "checkpointing";
 
+  public static final String PARALLELISM = "parallelism";
+
   private ConfigurationKeys() {}
 
 }

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

@@ -63,6 +63,13 @@ public final class HistoryServiceFlinkJob {
       this.env.enableCheckpointing(commitIntervalMs);
     }
 
+    // Parallelism
+    final Integer parallelism = this.config.getInteger(ConfigurationKeys.PARALLELISM, null);
+    if (parallelism != null) {
+      LOGGER.error("Set parallelism: {}.", parallelism);
+      this.env.setParallelism(parallelism);
+    }
+
     // State Backend
     final StateBackend stateBackend = StateBackends.fromConfiguration(this.config);
     this.env.setStateBackend(stateBackend);

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

@@ -76,6 +76,13 @@ public final class AggregationServiceFlinkJob {
       this.env.enableCheckpointing(commitIntervalMs);
     }
 
+    // Parallelism
+    final Integer parallelism = this.config.getInteger(ConfigurationKeys.PARALLELISM, null);
+    if (parallelism != null) {
+      LOGGER.error("Set parallelism: {}.", parallelism);
+      this.env.setParallelism(parallelism);
+    }
+
     // State Backend
     final StateBackend stateBackend = StateBackends.fromConfiguration(this.config);
     this.env.setStateBackend(stateBackend);

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

@@ -15,7 +15,7 @@ public final class ConfigurationKeys {
   public static final String KAFKA_OUTPUT_TOPIC = "kafka.output.topic";
 
   public static final String KAFKA_INPUT_TOPIC = "kafka.input.topic";
-  
+
   public static final String SCHEMA_REGISTRY_URL = "schema.registry.url";
 
   public static final String WINDOW_SIZE_MS = "window.size.ms";
@@ -28,13 +28,15 @@ public final class ConfigurationKeys {
 
   public static final String FLINK_STATE_BACKEND_PATH = "flink.state.backend.path";
 
-  public static final String FLINK_STATE_BACKEND_MEMORY_SIZE = //NOPMD
+  public static final String FLINK_STATE_BACKEND_MEMORY_SIZE = // NOPMD
       "flink.state.backend.memory.size";
 
   public static final String DEBUG = "debug";
 
   public static final String CHECKPOINTING = "checkpointing";
 
+  public static final String PARALLELISM = "parallelism";
+
   private ConfigurationKeys() {}
 
 }