WIP: Hazelcast Jet Benchmark/Docker-Compose/Kubernetes implementation

Hazelcast Jet implementation for benchmark use cases (eventually 1-3 or 1-4) with an additional commons project; docker-compose implementation for hazelcast jet; kubernetes implementation for hazelcast jet;

Open points:

• Add benchmark implementations. This includes
  • UC1
  • UC2
  • UC3
  • UC4
• Move common logic to a "hazelcastjet-commons" project, similar to those of the other implementations
• Add docker-compose files for all Hazelcast Jet implementations (might be tackled by a follow-up merge request)
• Add Kubernetes benchmark definitions for all Hazelcast Jet implementations (might be tackled by a follow-up merge request). This includes:
  • Kubernetes Deployments + Services + maybe other resources
  • A Kubernetes benchmark resource of kind benchmarks.theodolite.com
• Ensure that record lag can be monitored via Prometheus and Kafka Lag Exporter (@she can assist here)
• Provide appropriate documentation

TODO for @she:

• Remove commits from flink-benchmark-def-with-rm-rebalance Branch

theodolite-benchmarks/uc1-hazelcastjet/src/main/java/theodolite/uc1/application/HistoryService.java

@@ -5,6 +5,7 @@ import com.hazelcast.config.JoinConfig;
 import com.hazelcast.internal.json.JsonObject;
 import com.hazelcast.jet.Jet;
 import com.hazelcast.jet.JetInstance;
+import com.hazelcast.jet.config.JobConfig;
 import com.hazelcast.jet.kafka.KafkaSources;
 import com.hazelcast.jet.pipeline.Pipeline;
 import com.hazelcast.jet.pipeline.Sinks;
@@ -104,13 +105,14 @@ public class HistoryService {
   }
 
   /** Set Pipeline Kafka Properties. */
-  public HistoryService setKafkaPropertiesForPipeline(final Properties kafkaPropertiesForPipeline) {
+  public HistoryService setKafkaPropertiesForPipeline(// NOPMD
+      final Properties kafkaPropertiesForPipeline) {
     this.kafkaPropertiesForPipeline = kafkaPropertiesForPipeline;
     return this;
   }
 
   /** Set Kafka Input topic used to build the pipeline. */
-  public HistoryService setKafkaInputTopic(final String kafkaInputTopic) {
+  public HistoryService setKafkaInputTopic(final String kafkaInputTopic) { // NOPMD
     this.kafkaInputTopic = kafkaInputTopic;
     return this;
   }
@@ -148,7 +150,8 @@ public class HistoryService {
     final Pipeline p = Pipeline.create();
     p.readFrom(KafkaSources.<String, ActivePowerRecord>kafka(this.kafkaPropertiesForPipeline,
         this.kafkaInputTopic))
-        .withoutTimestamps()
+        .withNativeTimestamps(0)
+        .setLocalParallelism(1)
         .setName("Log content")
         .map(record -> {
           final ActivePowerRecord powerRecord = record.getValue();
@@ -167,12 +170,9 @@ public class HistoryService {
         .setPort(this.clusterConfig.getPort())
         .setPortAutoIncrement(this.clusterConfig.isPortAutoIncrement())
         .getJoin();
-
-    System.out.println("Port: " + this.clusterConfig.getPort());
     joinConfig.getMulticastConfig().setEnabled(false);
     if (this.clusterConfig.hasBootstrapServer()) {
       joinConfig.getTcpIpConfig().addMember(this.clusterConfig.getBootstrapServer());
-      System.out.println("Added Member to port: " + this.clusterConfig.getBootstrapServer());
     } else if (this.clusterConfig.hasKubernetesDnsName()) {
       joinConfig.getKubernetesConfig()
           .setEnabled(true)
@@ -183,7 +183,9 @@ public class HistoryService {
     // Add config and add pipeline as the job
     final JetInstance jet = Jet.newJetInstance();
     jet.getConfig().setHazelcastConfig(config);
-    jet.newJob(p).join();
+    final JobConfig jobConfig = new JobConfig();
+    jobConfig.setName("uc1-hazelcastjet");
+    jet.newJobIfAbsent(p, jobConfig).join();
   }