Migrate Flink benchmark implementation

This merge request will fix #20 (closed). It is based on the existing implementation in the apache-flink branch but has to tackle the following requirements:

• Migrate the individual benchmarks with their new enumeration
  • Migrate UC1
  • Migrate UC2 (f.k.a. UC3)
  • Migrate UC3 (f.k.a. UC4)
  • Migrate UC4 (f.k.a. UC2)
    • Migrate to latest architecture including dedicated feedback loop
      • Moved to #173
• Migrate to Titan Control Center Commons 0.1.0
• Fix code quality issues
• Support Docker builds (already available in apache-flink)
• Build and push Docker images in CI pipeline
• Update Docker Compose files in docker-test
• Enhance Gradle configuration sharing (see #107 (closed), might be tackled in a follow up MR)
  • Moved to !95 (merged)
• Clean up Gradle build files (see #59, might be tackled in a follow up MR)
  • Moved to #59
• Rename Gradle subprojects. Envisaged naming is already available in settings.gradle files as comments)
  • Moved to #174 (closed)
• Provide corresponding Kubernetes benchmark definitions (might be tackled in a follow up ticket)
  • Moved to #176 (closed)

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

 package theodolite.uc1.application;
 
+import java.util.Properties;
 import org.apache.commons.configuration2.Configuration;
 import org.apache.flink.api.common.serialization.DeserializationSchema;
 import org.apache.flink.formats.avro.registry.confluent.ConfluentRegistryAvroDeserializationSchema;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
-import titan.ccp.common.configuration.Configurations;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import titan.ccp.common.configuration.ServiceConfigurations;
 import titan.ccp.model.records.ActivePowerRecord;
 
-import java.util.Properties;
-
 /**
- * The History Microservice Flink Job.
+ * The History microservice implemented as a Flink job.
  */
 public class HistoryServiceFlinkJob {
 
-  private final Configuration config = Configurations.create();
+  private static final Logger LOGGER = LoggerFactory.getLogger(HistoryServiceFlinkJob.class);
+
+  private final Configuration config = ServiceConfigurations.createWithDefaults();
 
   private void run() {
     final String applicationName = this.config.getString(ConfigurationKeys.APPLICATION_NAME);
@@ -32,12 +35,6 @@ public class HistoryServiceFlinkJob {
     kafkaProps.setProperty("bootstrap.servers", kafkaBroker);
     kafkaProps.setProperty("group.id", applicationId);
 
-    /*
-     * final DeserializationSchema<ActivePowerRecord> serde = new
-     * FlinkMonitoringRecordSerde<>(inputTopic, ActivePowerRecord.class,
-     * ActivePowerRecordFactory.class);
-     */
-
     final DeserializationSchema<ActivePowerRecord> serde =
         ConfluentRegistryAvroDeserializationSchema.forSpecific(
             ActivePowerRecord.class,
@@ -46,12 +43,14 @@ public class HistoryServiceFlinkJob {
     final FlinkKafkaConsumer<ActivePowerRecord> kafkaConsumer =
         new FlinkKafkaConsumer<>(inputTopic, serde, kafkaProps);
     kafkaConsumer.setStartFromGroupOffsets();
-    if (checkpointing)
+    if (checkpointing) {
       kafkaConsumer.setCommitOffsetsOnCheckpoints(true);
+    }
 
     final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-    if (checkpointing)
+    if (checkpointing) {
       env.enableCheckpointing(commitIntervalMs);
+    }
 
     final DataStream<ActivePowerRecord> stream = env.addSource(kafkaConsumer);
 
@@ -65,8 +64,8 @@ public class HistoryServiceFlinkJob {
 
     try {
       env.execute(applicationId);
-    } catch (Exception e) {
-      e.printStackTrace();
+    } catch (final Exception e) { // NOPMD Execution thrown by Flink
+      LOGGER.error("An error occured while running this job.", e);
     }
   }