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/uc1-beam-flink/src/main/java/application/Uc1ApplicationBeam.java

@@ -20,6 +20,8 @@ import org.apache.beam.sdk.values.PBegin;
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.common.serialization.StringDeserializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import titan.ccp.model.records.ActivePowerRecord;
 
 /**
@@ -30,25 +32,46 @@ import titan.ccp.model.records.ActivePowerRecord;
  * ${workspace_loc:/uc1-application-samza/eclipseConsoleLogs.log} as Output File under Standard
  * Input Output in Common in the Run Configuration Start via Eclipse Run.
  */
-public class Uc1ApplicationBeam {
-  @SuppressWarnings({"unchecked", "rawtypes", "serial"})
+public final class Uc1ApplicationBeam {
+  private static final Logger LOGGER = LoggerFactory.getLogger(Uc1ApplicationBeam.class);
+  private static final String BOOTSTRAP = "KAFKA_BOOTSTRAP_SERVERS";
+  private static final String INPUT = "INPUT";
+  private static final String SCHEMA_REGISTRY = "SCHEMA_REGISTRY_URL";
+  private static final String YES = "true";
+  private static final String USE_AVRO_READER = YES;
+  private static final String AUTO_COMMIT_CONFIG = YES;
+
+  /**
+   * Private constructor to avoid instantiation.
+   */
+  private Uc1ApplicationBeam() {
+    throw new UnsupportedOperationException();
+  }
+
+  /**
+   * Main method.
+   *
+   */
+  @SuppressWarnings({"unchecked", "rawtypes","unused"})
   public static void main(final String[] args) {
+
     // Set Configuration for Kafka
     final String bootstrapServer =
-        System.getenv("KAFKA_BOOTSTRAP_SERVERS") != null ? System.getenv("KAFKA_BOOTSTRAP_SERVERS")
-            : "my-confluent-cp-kafka:9092";
-    final String inputTopic = System.getenv("INPUT") != null ? System.getenv("INPUT") : "input";
-    final String schemaRegistryURL =
-        System.getenv("SCHEMA_REGISTRY_URL") != null ? System.getenv("SCHEMA_REGISTRY_URL")
-            : "http://my-confluent-cp-schema-registry:8081";
+        System.getenv(BOOTSTRAP) == null ? "my-confluent-cp-kafka:9092"
+            : System.getenv(BOOTSTRAP);
+    final String inputTopic = System.getenv(INPUT) == null ? "input" : System.getenv(INPUT);
+    final String schemaRegistryUrl =
+        System.getenv(SCHEMA_REGISTRY) == null ? "http://my-confluent-cp-schema-registry:8081"
+            : System.getenv(SCHEMA_REGISTRY);
     // Set consumer configuration for the schema registry and commits back to Kafka
     final HashMap<String, Object> consumerConfig = new HashMap<>();
-    consumerConfig.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
+    consumerConfig.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, AUTO_COMMIT_CONFIG);
     consumerConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
-    consumerConfig.put("schema.registry.url", schemaRegistryURL);
-    consumerConfig.put("specific.avro.reader", "true");
+    consumerConfig.put("schema.registry.url", schemaRegistryUrl);
+    consumerConfig.put("specific.avro.reader", USE_AVRO_READER);
     consumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, "uc-application");
 
+    final LogKeyValue logKeyValue = new LogKeyValue();
 
     final PipelineOptions options = PipelineOptionsFactory.fromArgs(args).create();
     options.setJobName("ucapplication");
@@ -70,6 +93,7 @@ public class Uc1ApplicationBeam {
                 AvroCoder.of(ActivePowerRecord.class))
             .withConsumerConfigUpdates(consumerConfig)
             .withoutMetadata();
+
     // Apply pipeline transformations
     // Read from Kafka
     pipeline.apply(kafka)
@@ -77,7 +101,7 @@ public class Uc1ApplicationBeam {
         .apply(MapElements
             .via(
                 new SimpleFunction<KV<String, ActivePowerRecord>, KV<String, String>>() {
-                  transient Gson gsonObj = new Gson();
+                  private transient Gson gsonObj = new Gson();
 
                   @Override
                   public KV<String, String> apply(
@@ -90,18 +114,25 @@ public class Uc1ApplicationBeam {
                   }
                 }))
         // Print to console
-        .apply(ParDo.of(new DoFn<KV<String, String>, KV<String, String>>() {
-          @ProcessElement
-          public void processElement(@Element final KV<String, String> kv,
-              final OutputReceiver<KV<String, String>> out) {
-            System.out.println("Key: " + kv.getKey() + "Value: " + kv.getValue());
-//            out.output(kv);
-          }
-        }));
-
+        .apply(ParDo.of(logKeyValue));
     // Submit job and start execution
     pipeline.run().waitUntilFinish();
+  }
+
+  /**
+   * Logs all Key Value pairs.
+   */
+  @SuppressWarnings({"unused"})
+  private static class LogKeyValue extends DoFn<KV<String, String>,KV<String, String>> {
+    private static final long serialVersionUID = 4328743;
 
+    @ProcessElement
+    public void processElement(@Element final KV<String, String> kv,
+                               final OutputReceiver<KV<String, String>> out) {
+      if (LOGGER.isInfoEnabled()) {
+        LOGGER.info("Key: " + kv.getKey() + "Value: " + kv.getValue());
+      }
+    }
   }
 }