Migrate Beam benchmark implementation

Co-authored-by: Jan Bensien stu128012@mail.uni-kiel.de

This merge request will fix #147 (closed). It is based on the existing implementation in the https://git.se.informatik.uni-kiel.de/stu128012/theodolite/-/tree/beam_current_theodolite fork 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 Titan Control Center Commons 0.1.0 (Avro records and Schema Registry are already used)
• Fix code quality issues if any
• Support Docker builds (already available in the base branch)
• Build and push Docker images in CI pipeline
• Update Docker Compose files in docker-test
• Share code among Beam runners (either create subprojects or dedicated Gradle tasks or dedicated Dockerfiles or environment variables or ...)
  • Current approach is to use different subprojects. Other approaches would increase the complexity and is currently not possible because of:
    • https://issues.apache.org/jira/browse/BEAM-9830
    • https://stackoverflow.com/questions/61441333/conflict-with-runner-dependencies-in-beam
    • build beam commons project (for builing configurations, common PT transformations)
    • outsource uc1-4 pipelines into single projects
    • outsource uc1-4 beam samza into single projects
    • outsource uc1-4 beam flink into single projects
• Clean up Gradle build files
  • Build Gradle plugins:
    • Build beam commons plugin
    • Build beam flink plugin
    • Build beam samza plugin
  • Check if same versions of dependecies are used (slf4, etc)
• Provide corresponding Kubernetes benchmark definitions -> Moved to #281 (closed)
• Solve #278 (closed)
• Fix #293 (closed)
• Solve #291 (not part of this MR)
• Make samza configurable through environment variables
• Resolve ./gradlew build --warning-mode all
• @she Image names in Docker Compose files must be replaced when merging and pushing images

Depends on !188 (merged)

theodolite-benchmarks/beam-commons/src/main/java/theodolite/commons/beam/kafka/KafkaActivePowerRecordReader.java

+package theodolite.commons.beam.kafka;
+
+import io.confluent.kafka.serializers.KafkaAvroDeserializer;
+import java.util.Properties;
+import org.apache.beam.sdk.coders.AvroCoder;
+import org.apache.beam.sdk.io.kafka.KafkaIO;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PBegin;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import titan.ccp.model.records.ActivePowerRecord;
+
+/**
+ * Simple {@link PTransform} that read from Kafka using {@link KafkaIO}.
+ */
+public class KafkaActivePowerRecordReader extends
+    PTransform<PBegin, PCollection<KV<String, ActivePowerRecord>>> {
+
+  private static final long serialVersionUID = 2603286150183186115L;
+  private final PTransform<PBegin, PCollection<KV<String, ActivePowerRecord>>> reader;
+
+
+  /**
+   * Instantiates a {@link PTransform} that reads from Kafka with the given Configuration.
+   */
+  @SuppressWarnings({"unchecked", "rawtypes"})
+  public KafkaActivePowerRecordReader(final String bootstrapServer, final String inputTopic,
+                                      final Properties consumerConfig) {
+    super();
+
+    // Check if boostrap server and inputTopic are defined
+    if (bootstrapServer.isEmpty() || inputTopic.isEmpty()) {
+      throw new IllegalArgumentException("bootstrapServer or inputTopic missing");
+    }
+
+    reader =
+        KafkaIO.<String, ActivePowerRecord>read()
+            .withBootstrapServers(bootstrapServer)
+            .withTopic(inputTopic)
+            .withKeyDeserializer(StringDeserializer.class)
+            .withValueDeserializerAndCoder((Class) KafkaAvroDeserializer.class,
+                AvroCoder.of(ActivePowerRecord.class))
+            .withConsumerConfigUpdates(consumerConfig)
+            .withoutMetadata();
+  }
+
+  @Override
+  public PCollection<KV<String, ActivePowerRecord>> expand(final PBegin input) {
+    return input.apply(this.reader);
+  }
+
+}