diff --git a/.settings/qa.eclipse.plugin.pmd.prefs b/.settings/qa.eclipse.plugin.pmd.prefs
index 04ac39484810c88779a38e11662b0bab0961a22e..a8950d58ca40d8762a665ce3174deda7b2d89b85 100644
--- a/.settings/qa.eclipse.plugin.pmd.prefs
+++ b/.settings/qa.eclipse.plugin.pmd.prefs
@@ -1,4 +1,4 @@
 customRulesJars=
 eclipse.preferences.version=1
-enabled=true
+enabled=false
 ruleSetFilePath=config/pmd.xml
diff --git a/execution/README.md b/execution/README.md
index fb864fcdfb5c0791befd802fe7e2fad38383f29e..f7df1a32ad8b7898ba34b792095957a8e1362fc8 100644
--- a/execution/README.md
+++ b/execution/README.md
@@ -68,19 +68,6 @@ below), we provide a [patch](https://github.com/SoerenHenning/cp-helm-charts)
 for these helm charts. Note that this patch is only required for observation and
 not for the actual benchmark execution and evaluation.
 
-<<<<<<< HEAD
-**TODO** Add required configuration, installation
-
-### The Kafka Lag Exporter
-
-Lightbend's Kafka Lag Exporter can be installed via helm:
-
-``sh
-helm install kafka-lag-exporter https://github.com/lightbend/kafka-lag-exporter/releases/download/v0.6.0/kafka-lag-exporter-0.6.0.tgz
-``
-
-**TODO** Add configuration + ServiceMonitor
-=======
 #### Our patched Confluent Helm Charts
 
 To use our patched Confluent Helm Charts clone the
@@ -118,17 +105,10 @@ To let Prometheus scrape Kafka lag metrics, deploy a ServiceMonitor:
 ```sh
 kubectl apply -f infrastructure/kafka-lag-exporter/service-monitor.yaml
 ```
->>>>>>> 624692753eb09684dd3dda3926482e9b56ada0d6
 
 
 ## Python 3.7
 
-<<<<<<< HEAD
-For executing benchmarks and analyzing their results, a Python 3.7 installation
-is required. We suggest to use a virtual environment placed in the `.venv` directory.
-
-**TODO** Show how to install requirements
-=======
 For executing benchmarks and analyzing their results, a **Python 3.7** installation
 is required. We suggest to use a virtual environment placed in the `.venv` directory.
 
@@ -169,4 +149,3 @@ The `./run_loop.sh` is the entrypoint for all benchmark executions. Is has to be
 * `<memory-limit>`: Kubernetes memory limit. Optional. Default `4Gi`.
 * `<commit-interval>`: Kafka Streams' commit interval in milliseconds. Optional. Default `100`.
 * `<duration>`: Duration in minutes subexperiments should be executed for. Optional. Default `5`.
->>>>>>> 624692753eb09684dd3dda3926482e9b56ada0d6
diff --git a/execution/cluster-setup.sh b/execution/cluster-setup.sh
new file mode 100755
index 0000000000000000000000000000000000000000..a0e271deda23106f3c734f546a6059d09995d08c
--- /dev/null
+++ b/execution/cluster-setup.sh
@@ -0,0 +1,60 @@
+#!/bin/bash
+
+CP_HELM_PATH='../../cp-helm-charts'
+
+## minikube ##
+# if minikube stop responding after few minutes
+
+# minikube delete 
+
+# cd 
+# rm -r .minikube
+# cd Dokumente/Master-2-SoSe-2020/project/spesb/execution/
+
+#minikube config set memory  4046
+#minikube delete
+#minikube config set cpus 4 
+#minikube delete
+#minikube start --vm-driver=virtualbox
+
+## kind ## 
+kind delete cluster 
+kind create cluster # --config  infrastructure/cluster/kind-configuration.yaml 
+
+# K8s dashboard
+# Token: eyJhbGciOiJSUzI1NiIsImtpZCI6ImdGNlU1U3BnN01XcS14RnlWUFRBODlaTzNpeUtxa1hTV3VKNTVmVGVrZ2MifQ.eyJpc3MiOiJrdWJlcm5ldGVzL3NlcnZpY2VhY2NvdW50Iiwia3ViZXJuZXRlcy5pby9zZXJ2aWNlYWNjb3VudC9uYW1lc3BhY2UiOiJrdWJlcm5ldGVzLWRhc2hib2FyZCIsImt1YmVybmV0ZXMuaW8vc2VydmljZWFjY291bnQvc2VjcmV0Lm5hbWUiOiJkZWZhdWx0LXRva2VuLW5rbnc1Iiwia3ViZXJuZXRlcy5pby9zZXJ2aWNlYWNjb3VudC9zZXJ2aWNlLWFjY291bnQubmFtZSI6ImRlZmF1bHQiLCJrdWJlcm5ldGVzLmlvL3NlcnZpY2VhY2NvdW50L3NlcnZpY2UtYWNjb3VudC51aWQiOiI2NDE5NTUzYy0yY2RkLTQ2OGUtYTMwNS03YzZlNWQ5NjhmMzMiLCJzdWIiOiJzeXN0ZW06c2VydmljZWFjY291bnQ6a3ViZXJuZXRlcy1kYXNoYm9hcmQ6ZGVmYXVsdCJ9.xkkZJViw4Q7RUbpzWjmTUGIGlgHfIhC94GuJ_bmGId3w8UxCP5PK6eq0yPNTTMJMT-yGr3qH7D1f616UNDJM1SrcoervG1fXyzw0XYmdbXluemW1LIm3WyzukhBs4dF4s93RrxUd9iHFjCrQanssXOSDDCZO-2V4BrpYEZ4TLvgMz9pAy4_k4-1gL4QKu8FBgCydBa2SBVOZ8tFy_5r38KH7j9eX0OJD8kyugcmPz0ARaIZhyZERyTHz3wmxY5E-W_qSe1GY12EeR9c5KlWeGYYIzheyBr-TpcyLuoQQguxmI7Ico917k0zG2YBSEYdfTo1I2LPXCYzbp__MAIV_TQ
+# kubectl apply -f https://raw.githubusercontent.com/kubernetes/dashboard/v2.0.0/aio/deploy/recommended.yaml
+# kubectl proxy & 
+
+# prometheus
+helm install prometheus-operator stable/prometheus-operator -f infrastructure/prometheus/helm-values.yaml
+kubectl apply -f infrastructure/prometheus/service-account.yaml
+kubectl apply -f infrastructure/prometheus/cluster-role.yaml
+kubectl apply -f infrastructure/prometheus/cluster-role-binding.yaml
+kubectl apply -f infrastructure/prometheus/prometheus.yaml
+
+# grafana 
+kubectl apply -f infrastructure/grafana/prometheus-datasource-config-map.yaml
+kubectl apply -f infrastructure/grafana/dashboard-config-map.yaml
+helm install grafana stable/grafana -f infrastructure/grafana/values.yaml
+
+# kafka + lag-exporter
+helm install my-confluent $CP_HELM_PATH  -f infrastructure/kafka/values.yaml
+kubectl apply -f $CP_HELM_PATH/examples/kafka-client.yaml 
+kubectl apply -f infrastructure/kafka/service-monitor.yaml
+
+helm install kafka-lag-exporter https://github.com/lightbend/kafka-lag-exporter/releases/download/v0.6.0/kafka-lag-exporter-0.6.0.tgz -f infrastructure/kafka-lag-exporter/values.yaml
+kubectl apply -f infrastructure/kafka-lag-exporter/service-monitor.yaml
+
+# port fowarding kubectl "port-forward --namespace monitoring <pod name> <local port>:<container port>"
+sleep 3m # wait for grafana and prometheus pods
+kubectl port-forward $(kubectl get pods  -o name | grep grafana) 3000:3000 & 
+kubectl port-forward $(kubectl get pods  -o name | grep prometheus-prometheus) 9090:9090 &
+
+# open web interfaces
+# xdg-open http://localhost:8001/api/v1/namespaces/kubernetes-dashboard/services/https:kubernetes-dashboard:/proxy/. 
+xdg-open http://localhost:9090
+xdg-open http://localhost:3000
+
+# grafana token
+# kubectl get secret --namespace default grafana -o jsonpath="{.data.admin-password}" | base64 --decode ; echo
diff --git a/execution/exp_counter.txt b/execution/exp_counter.txt
index 573541ac9702dd3969c9bc859d2b91ec1f7e6e56..ec635144f60048986bc560c5576355344005e6e7 100644
--- a/execution/exp_counter.txt
+++ b/execution/exp_counter.txt
@@ -1 +1 @@
-0
+9
diff --git a/execution/infrastructure/kafka/values.yaml b/execution/infrastructure/kafka/values.yaml
index a33a6f21f8529377162704fbabc7a381706ad64e..ea7c0327f90e3027030d72a4efb92c19ea671d97 100644
--- a/execution/infrastructure/kafka/values.yaml
+++ b/execution/infrastructure/kafka/values.yaml
@@ -3,7 +3,7 @@
 ## ------------------------------------------------------
 cp-zookeeper:
   enabled: true
-  servers: 3
+  servers: 1
   image: confluentinc/cp-zookeeper
   imageTag: 5.4.0
   ## Optionally specify an array of imagePullSecrets. Secrets must be manually created in the namespace.
@@ -38,7 +38,7 @@ cp-zookeeper:
 ## ------------------------------------------------------
 cp-kafka:
   enabled: true
-  brokers: 10
+  brokers: 1
   image: confluentinc/cp-enterprise-kafka
   imageTag: 5.4.0
   ## Optionally specify an array of imagePullSecrets. Secrets must be manually created in the namespace.
@@ -61,7 +61,7 @@ cp-kafka:
   #   cpu: 100m
   #   memory: 128Mi
   configurationOverrides:
-    #"offsets.topic.replication.factor": "3"
+    offsets.topic.replication.factor: 1
     "message.max.bytes": "134217728" # 128 MB
     "replica.fetch.max.bytes": "134217728" # 128 MB
     # "default.replication.factor": 3
diff --git a/execution/infrastructure/prometheus/cluster-role-binding.yaml b/execution/infrastructure/prometheus/cluster-role-binding.yaml
index 5369e02aac84440053b3be5485f0644419d981d1..db2717cddcea180f84bb68377ba6daad37c33296 100644
--- a/execution/infrastructure/prometheus/cluster-role-binding.yaml
+++ b/execution/infrastructure/prometheus/cluster-role-binding.yaml
@@ -9,4 +9,4 @@ roleRef:
 subjects:
 - kind: ServiceAccount
   name: prometheus
-  namespace: titan-scalability
\ No newline at end of file
+  namespace: default
\ No newline at end of file
diff --git a/execution/run_loop.sh b/execution/run_loop.sh
index e63c0ecdfc54d27456afd720cc66303bfb143b28..4c7d49bbce67de29f81f0771bd4f1c84e5b35404 100755
--- a/execution/run_loop.sh
+++ b/execution/run_loop.sh
@@ -4,8 +4,10 @@ UC=$1
 IFS=', ' read -r -a DIM_VALUES <<< "$2"
 IFS=', ' read -r -a REPLICAS <<< "$3"
 PARTITIONS=${4:-40}
-CPU_LIMIT=${5:-1000m}
-MEMORY_LIMIT=${6:-4Gi}
+CPU_LIMIT=${5:-200m}
+MEMORY_LIMIT=${6:-400Mi}
+#CPU_LIMIT=${5:-1000m}
+#MEMORY_LIMIT=${6:-4Gi}
 KAFKA_STREAMS_COMMIT_INTERVAL_MS=${7:-100}
 EXECUTION_MINUTES=${8:-5}
 
diff --git a/execution/run_uc1-new.sh b/execution/run_uc1-new.sh
index 0edb75d002861393ce9a4b1b59c21e5871c651eb..ed58972667dca9b5783cf38cc380e038d61596e2 100755
--- a/execution/run_uc1-new.sh
+++ b/execution/run_uc1-new.sh
@@ -26,7 +26,8 @@ kubectl exec kafka-client -- bash -c "kafka-topics --zookeeper my-confluent-cp-z
 
 # Start workload generator
 NUM_SENSORS=$DIM_VALUE
-WL_MAX_RECORDS=150000
+#WL_MAX_RECORDS=150000
+WL_MAX_RECORDS=25
 WL_INSTANCES=$(((NUM_SENSORS + (WL_MAX_RECORDS -1 ))/ WL_MAX_RECORDS))
 
 WORKLOAD_GENERATOR_YAML=$(sed "s/{{NUM_SENSORS}}/$NUM_SENSORS/g; s/{{INSTANCES}}/$WL_INSTANCES/g" uc1-workload-generator/deployment.yaml)
diff --git a/execution/uc1-workload-generator/deployment.yaml b/execution/uc1-workload-generator/deployment.yaml
index a0fde4bbf9765b2bb56bd36acde430d97169f34b..968496b8d1dbcec88c89d5967b54e24fefd00a47 100644
--- a/execution/uc1-workload-generator/deployment.yaml
+++ b/execution/uc1-workload-generator/deployment.yaml
@@ -16,8 +16,13 @@ spec:
       terminationGracePeriodSeconds: 0
       containers:
       - name: workload-generator
-        image: soerenhenning/uc1-wg:latest 
+  #      image: soerenhenning/uc1-wg:latest 
+        image: sehrenstein/uc1-wg:latest 
         env:
+        - name: ZK_HOST
+          value: "my-confluent-cp-zookeeper"
+        - name: ZK_PORT
+          value: "2181"
         - name: KAFKA_BOOTSTRAP_SERVERS
           value: "my-confluent-cp-kafka:9092"
         - name: NUM_SENSORS
diff --git a/test-workload-generator/.settings/qa.eclipse.plugin.pmd.prefs b/test-workload-generator/.settings/qa.eclipse.plugin.pmd.prefs
index efbcb8c9e5d449194a48ca1ea42b7d807b573db9..40bfd0ecdbbe324bb54e4b9f9f32ba95cf5b0c2a 100644
--- a/test-workload-generator/.settings/qa.eclipse.plugin.pmd.prefs
+++ b/test-workload-generator/.settings/qa.eclipse.plugin.pmd.prefs
@@ -1,4 +1,4 @@
 customRulesJars=
 eclipse.preferences.version=1
-enabled=true
+enabled=false
 ruleSetFilePath=../config/pmd.xml
diff --git a/test-workload-generator/docker-compose.yml b/test-workload-generator/docker-compose.yml
index a1d7533d2f46cb334e63738cb501d9223c309668..b34f9239f27b54ae25948c7397f374a3cfd94610 100644
--- a/test-workload-generator/docker-compose.yml
+++ b/test-workload-generator/docker-compose.yml
@@ -1,7 +1,20 @@
 version: '3.1'
 
+services:
+  version: '3.1'
+
 services:
   zookeeper:
     image: zookeeper
     ports:
-      - 2181:2181
+      - "2181:2181"
+  kafka:
+    image: wurstmeister/kafka
+    ports:
+      - "9092:9092"
+    environment:
+      KAFKA_ADVERTISED_HOST_NAME: localhost # Replace with docker network
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+      KAFKA_ZOOKEEPER_CONNECTION_TIMEOUT_MS: 30000
+      KAFKA_CREATE_TOPICS: "input:3:1"
+
diff --git a/test-workload-generator/src/main/java/test/Main.java b/test-workload-generator/src/main/java/test/Main.java
index 29ff7ded9460d8cab4a138d85c854f473670b906..65ed8e8742b39174841bcb6765552ddb8f2c956f 100644
--- a/test-workload-generator/src/main/java/test/Main.java
+++ b/test-workload-generator/src/main/java/test/Main.java
@@ -1,20 +1,28 @@
 package test;
 
-import common.KafkaWorkloadGenerator;
-import common.KafkaWorkloadGeneratorBuilder;
 import common.dimensions.Duration;
 import common.dimensions.KeySpace;
 import common.dimensions.Period;
+import common.generators.KafkaWorkloadGenerator;
+import common.generators.KafkaWorkloadGeneratorBuilder;
 import common.messages.OutputMessage;
+import common.misc.ZooKeeper;
+import communication.kafka.KafkaRecordSender;
 import java.util.concurrent.TimeUnit;
+import kieker.common.record.IMonitoringRecord;
 import titan.ccp.models.records.ActivePowerRecord;
 
 public class Main {
 
   public static void main(final String[] args) {
 
-    final KafkaWorkloadGenerator generator =
+    final KafkaRecordSender<IMonitoringRecord> recordSender =
+        new KafkaRecordSender<>("localhost:9092", "input");
+
+    final KafkaWorkloadGenerator<IMonitoringRecord> generator =
         KafkaWorkloadGeneratorBuilder.builder()
+            .setZooKeeper(new ZooKeeper("127.0.0.1", 2181))
+            .setKafkaRecordSender(recordSender)
             .setBeforeAction(() -> {
               System.out.println("Before Hook");
             })
@@ -26,10 +34,6 @@ public class Main {
                     new ActivePowerRecord(key, 0L, 100d)))
             .build();
 
-
-    // dwhedhwedherbfherf ferufer e u uebvhebzvbjkr fjkebhr erfberf rt gtr grt gtr
-    // gebuwbfuzerfuzerzgfer fe rf er fe rferhfveurfgerzfgzuerf erf erf ethvrif
-
     generator.start();
   }
 
diff --git a/uc1-workload-generator/build.gradle b/uc1-workload-generator/build.gradle
index d934bd09de1d64cadac982669d7cab5b564f0dd5..cba0ffcbcb3b74a40d45e27ac6874c7006fc9726 100644
--- a/uc1-workload-generator/build.gradle
+++ b/uc1-workload-generator/build.gradle
@@ -1 +1,5 @@
 mainClassName = "spesb.uc1.workloadgenerator.LoadGenerator"
+
+dependencies {
+    compile project(':workload-generator-common')
+}
\ No newline at end of file
diff --git a/uc1-workload-generator/src/main/java/spesb/uc1/workloadgenerator/LoadGenerator.java b/uc1-workload-generator/src/main/java/spesb/uc1/workloadgenerator/LoadGenerator.java
index 9eb95f0c104ee3a5cd497f735f839cdb474af6a9..ef024e71677910517d6d23ad7286b55e11e2164d 100644
--- a/uc1-workload-generator/src/main/java/spesb/uc1/workloadgenerator/LoadGenerator.java
+++ b/uc1-workload-generator/src/main/java/spesb/uc1/workloadgenerator/LoadGenerator.java
@@ -1,38 +1,39 @@
 package spesb.uc1.workloadgenerator;
 
+import common.dimensions.Duration;
+import common.dimensions.KeySpace;
+import common.dimensions.Period;
+import common.generators.KafkaWorkloadGenerator;
+import common.generators.KafkaWorkloadGeneratorBuilder;
+import common.messages.OutputMessage;
+import common.misc.ZooKeeper;
+import communication.kafka.KafkaRecordSender;
 import java.io.IOException;
-import java.util.List;
 import java.util.Objects;
 import java.util.Properties;
-import java.util.Random;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
-import java.util.regex.Pattern;
-import java.util.stream.Collectors;
-import java.util.stream.IntStream;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import spesb.kafkasender.KafkaRecordSender;
 import titan.ccp.models.records.ActivePowerRecord;
 
 public class LoadGenerator {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(LoadGenerator.class);
 
-  private static final int WL_MAX_RECORDS = 150_000;
-
   public static void main(final String[] args) throws InterruptedException, IOException {
     LOGGER.info("Start workload generator for use case UC1.");
 
     final int numSensors =
         Integer.parseInt(Objects.requireNonNullElse(System.getenv("NUM_SENSORS"), "10"));
-    final int instanceId = getInstanceId();
     final int periodMs =
         Integer.parseInt(Objects.requireNonNullElse(System.getenv("PERIOD_MS"), "1000"));
     final int value = Integer.parseInt(Objects.requireNonNullElse(System.getenv("VALUE"), "10"));
-    final int threads = Integer.parseInt(Objects.requireNonNullElse(System.getenv("THREADS"), "4"));
+    final int threads = Integer.parseInt(Objects.requireNonNullElse(System.getenv("THREADS"),
+        "4"));
+    final String zooKeeperHost = Objects.requireNonNullElse(System.getenv("ZK_HOST"), "localhost");
+    final int zooKeeperPort =
+        Integer.parseInt(Objects.requireNonNullElse(System.getenv("ZK_PORT"), "2181"));
     final String kafkaBootstrapServers =
         Objects.requireNonNullElse(System.getenv("KAFKA_BOOTSTRAP_SERVERS"), "localhost:9092");
     final String kafkaInputTopic =
@@ -41,13 +42,6 @@ public class LoadGenerator {
     final String kafkaLingerMs = System.getenv("KAFKA_LINGER_MS");
     final String kafkaBufferMemory = System.getenv("KAFKA_BUFFER_MEMORY");
 
-    final int idStart = instanceId * WL_MAX_RECORDS;
-    final int idEnd = Math.min((instanceId + 1) * WL_MAX_RECORDS, numSensors);
-    LOGGER.info("Generating data for sensors with IDs from {} to {} (exclusive).", idStart, idEnd);
-    final List<String> sensors = IntStream.range(idStart, idEnd)
-        .mapToObj(i -> "s_" + i)
-        .collect(Collectors.toList());
-
     final Properties kafkaProperties = new Properties();
     // kafkaProperties.put("acks", this.acknowledges);
     kafkaProperties.compute(ProducerConfig.BATCH_SIZE_CONFIG, (k, v) -> kafkaBatchSize);
@@ -60,33 +54,18 @@ public class LoadGenerator {
         r -> r.getTimestamp(),
         kafkaProperties);
 
-    final ScheduledExecutorService executor = Executors.newScheduledThreadPool(threads);
-    final Random random = new Random();
-
-    for (final String sensor : sensors) {
-      final int initialDelay = random.nextInt(periodMs);
-      executor.scheduleAtFixedRate(() -> {
-        kafkaRecordSender.write(new ActivePowerRecord(sensor, System.currentTimeMillis(), value));
-      }, initialDelay, periodMs, TimeUnit.MILLISECONDS);
-    }
-
-    System.out.println("Wait for termination...");
-    executor.awaitTermination(30, TimeUnit.DAYS);
-    System.out.println("Will terminate now");
+    final KafkaWorkloadGenerator<ActivePowerRecord> workloadGenerator =
+        KafkaWorkloadGeneratorBuilder.<ActivePowerRecord>builder()
+            .setKeySpace(new KeySpace("s_", numSensors))
+            .setThreads(threads)
+            .setPeriod(new Period(periodMs, TimeUnit.MILLISECONDS))
+            .setDuration(new Duration(100, TimeUnit.SECONDS))
+            .setGeneratorFunction(sensor -> new OutputMessage<>(sensor,
+                new ActivePowerRecord(sensor, System.currentTimeMillis(), value)))
+            .setZooKeeper(new ZooKeeper(zooKeeperHost, zooKeeperPort))
+            .setKafkaRecordSender(kafkaRecordSender)
+            .build();
 
+    workloadGenerator.start();
   }
-
-  private static int getInstanceId() {
-    final String podName = System.getenv("POD_NAME");
-    if (podName == null) {
-      return 0;
-    } else {
-      return Pattern.compile("-")
-          .splitAsStream(podName)
-          .reduce((p, x) -> x)
-          .map(Integer::parseInt)
-          .orElse(0);
-    }
-  }
-
 }
diff --git a/uc3-application/src/main/java/uc3/streamprocessing/KafkaStreamsBuilder.java b/uc3-application/src/main/java/uc3/streamprocessing/KafkaStreamsBuilder.java
index 02d0953a38b610887ceaa6bd4fa698df718bc597..9c496fba2188758d37adf098e022f479e5129aec 100644
--- a/uc3-application/src/main/java/uc3/streamprocessing/KafkaStreamsBuilder.java
+++ b/uc3-application/src/main/java/uc3/streamprocessing/KafkaStreamsBuilder.java
@@ -5,6 +5,7 @@ import java.util.Objects;
 import java.util.Properties;
 import org.apache.kafka.streams.KafkaStreams;
 import org.apache.kafka.streams.StreamsConfig;
+import spesb.uc3.streamprocessing.TopologyBuilder;
 import titan.ccp.common.kafka.streams.PropertiesBuilder;
 
 /**
diff --git a/workload-generator-common/build.gradle b/workload-generator-common/build.gradle
index 7e1c2aebe2786c7fe525850613c1ee9105b66826..7654647879ad9579c204c0504578e376bcbe308d 100644
--- a/workload-generator-common/build.gradle
+++ b/workload-generator-common/build.gradle
@@ -13,7 +13,6 @@ sourceCompatibility = "1.11"
 targetCompatibility = "1.11"
 
 dependencies {
-    compile project(':')
     compile 'org.apache.curator:curator-recipes:4.3.0'
     compile 'org.slf4j:slf4j-simple:1.6.1'
 
diff --git a/workload-generator-common/src/main/java/common/dimensions/Period.java b/workload-generator-common/src/main/java/common/dimensions/Period.java
index a7d4087c045608f422ad670a53c0e78093bf53a0..0aa129fd0b328a34bb0508dc7f91ae11e7454ffb 100644
--- a/workload-generator-common/src/main/java/common/dimensions/Period.java
+++ b/workload-generator-common/src/main/java/common/dimensions/Period.java
@@ -13,7 +13,7 @@ public class Period extends Dimension {
 
   /**
    * Define a new period.
-   * 
+   *
    * @param period the period
    * @param timeUnit the time unit that applies to the specified {@code period}
    */
@@ -23,7 +23,7 @@ public class Period extends Dimension {
     this.timeUnit = timeUnit;
   }
 
-  public int getDuration() {
+  public int getPeriod() {
     return this.period;
   }
 
diff --git a/workload-generator-common/src/main/java/common/dimensions/copy/Dimension.java b/workload-generator-common/src/main/java/common/dimensions/copy/Dimension.java
deleted file mode 100644
index 9ed031df72c2fb45653720bf8d18b805d121010c..0000000000000000000000000000000000000000
--- a/workload-generator-common/src/main/java/common/dimensions/copy/Dimension.java
+++ /dev/null
@@ -1,8 +0,0 @@
-package common.dimensions.copy;
-
-/*
- * Base class for workload dimensions.
- */
-public abstract class Dimension {
-
-}
diff --git a/workload-generator-common/src/main/java/common/dimensions/copy/Duration.java b/workload-generator-common/src/main/java/common/dimensions/copy/Duration.java
deleted file mode 100644
index 8105347f3af4091a6f724dd58af8a5e7d2d74b39..0000000000000000000000000000000000000000
--- a/workload-generator-common/src/main/java/common/dimensions/copy/Duration.java
+++ /dev/null
@@ -1,34 +0,0 @@
-package common.dimensions.copy;
-
-import java.util.concurrent.TimeUnit;
-import common.generators.WorkloadGenerator;
-
-/**
- * Wrapper class for the definition of the duration for the {@link WorkloadGenerator}.
- */
-public class Duration extends Dimension {
-
-  private final int duration;
-  private final TimeUnit timeUnit;
-
-  /**
-   * Define a new duration.
-   *
-   * @param duration the duration
-   * @param timeUnit the time unit that applies to the specified {@code duration}
-   */
-  public Duration(final int duration, final TimeUnit timeUnit) {
-    super();
-    this.duration = duration;
-    this.timeUnit = timeUnit;
-  }
-
-  public int getDuration() {
-    return this.duration;
-  }
-
-  public TimeUnit getTimeUnit() {
-    return this.timeUnit;
-  }
-
-}
diff --git a/workload-generator-common/src/main/java/common/dimensions/copy/KeySpace.java b/workload-generator-common/src/main/java/common/dimensions/copy/KeySpace.java
deleted file mode 100644
index 5c2c1776d84aa57f3296d144220b06e83d7ff566..0000000000000000000000000000000000000000
--- a/workload-generator-common/src/main/java/common/dimensions/copy/KeySpace.java
+++ /dev/null
@@ -1,56 +0,0 @@
-package common.dimensions.copy;
-
-import common.generators.WorkloadGenerator;
-
-/**
- * Wrapper class for the definition of the Keys that should be used by the
- * {@link WorkloadGenerator}.
- */
-public class KeySpace extends Dimension {
-
-  private final String prefix;
-  private final int min;
-  private final int max;
-
-
-  /**
-   * Create a new key space. All keys will have the prefix {@code prefix}. The remaining part of
-   * each key will be determined by a number of the interval ({@code min}, {@code max}-1).
-   *
-   * @param prefix the prefix to use for all keys
-   * @param min the lower bound (inclusive) to start counting from
-   * @param max the upper bound (exclusive) to count to
-   */
-  public KeySpace(final String prefix, final int min, final int max) {
-    if (prefix == null || prefix.contains(";")) {
-      throw new IllegalArgumentException(
-          "The prefix must not be null and must not contain the ';' character.");
-    }
-    this.prefix = prefix;
-    this.min = min;
-    this.max = max;
-
-  }
-
-  public KeySpace(final String prefix, final int numberOfKeys) {
-    this(prefix, 0, numberOfKeys - 1);
-  }
-
-  public KeySpace(final int numberOfKeys) {
-    this("sensor_", 0, numberOfKeys - 1);
-  }
-
-  public String getPrefix() {
-    return this.prefix;
-  }
-
-
-  public int getMin() {
-    return this.min;
-  }
-
-
-  public int getMax() {
-    return this.max;
-  }
-}
diff --git a/workload-generator-common/src/main/java/common/dimensions/copy/Period.java b/workload-generator-common/src/main/java/common/dimensions/copy/Period.java
deleted file mode 100644
index 6c44543f3188f6f5a6b2ad95968990aece75d663..0000000000000000000000000000000000000000
--- a/workload-generator-common/src/main/java/common/dimensions/copy/Period.java
+++ /dev/null
@@ -1,34 +0,0 @@
-package common.dimensions.copy;
-
-import java.util.concurrent.TimeUnit;
-import common.generators.WorkloadGenerator;
-
-/**
- * Wrapper class for the definition of period to use for the {@link WorkloadGenerator}.
- */
-public class Period extends Dimension {
-
-  private final int period;
-  private final TimeUnit timeUnit;
-
-  /**
-   * Define a new period.
-   * 
-   * @param period the period
-   * @param timeUnit the time unit that applies to the specified {@code period}
-   */
-  public Period(final int period, final TimeUnit timeUnit) {
-    super();
-    this.period = period;
-    this.timeUnit = timeUnit;
-  }
-
-  public int getDuration() {
-    return this.period;
-  }
-
-  public TimeUnit getTimeUnit() {
-    return this.timeUnit;
-  }
-
-}
diff --git a/workload-generator-common/src/main/java/common/dimensions/copy2/Dimension.java b/workload-generator-common/src/main/java/common/dimensions/copy2/Dimension.java
deleted file mode 100644
index 84b94088fa2065807c665f438d7374215959bbcc..0000000000000000000000000000000000000000
--- a/workload-generator-common/src/main/java/common/dimensions/copy2/Dimension.java
+++ /dev/null
@@ -1,8 +0,0 @@
-package common.dimensions.copy2;
-
-/*
- * Base class for workload dimensions.
- */
-public abstract class Dimension {
-
-}
diff --git a/workload-generator-common/src/main/java/common/dimensions/copy2/Duration.java b/workload-generator-common/src/main/java/common/dimensions/copy2/Duration.java
deleted file mode 100644
index fb31929cf212084a007e61fece4813689f232c45..0000000000000000000000000000000000000000
--- a/workload-generator-common/src/main/java/common/dimensions/copy2/Duration.java
+++ /dev/null
@@ -1,34 +0,0 @@
-package common.dimensions.copy2;
-
-import java.util.concurrent.TimeUnit;
-import common.generators.WorkloadGenerator;
-
-/**
- * Wrapper class for the definition of the duration for the {@link WorkloadGenerator}.
- */
-public class Duration extends Dimension {
-
-  private final int duration;
-  private final TimeUnit timeUnit;
-
-  /**
-   * Define a new duration.
-   *
-   * @param duration the duration
-   * @param timeUnit the time unit that applies to the specified {@code duration}
-   */
-  public Duration(final int duration, final TimeUnit timeUnit) {
-    super();
-    this.duration = duration;
-    this.timeUnit = timeUnit;
-  }
-
-  public int getDuration() {
-    return this.duration;
-  }
-
-  public TimeUnit getTimeUnit() {
-    return this.timeUnit;
-  }
-
-}
diff --git a/workload-generator-common/src/main/java/common/dimensions/copy2/KeySpace.java b/workload-generator-common/src/main/java/common/dimensions/copy2/KeySpace.java
deleted file mode 100644
index 311dffa62ad09a7a7f68ee43e31fdbe6acfe9a46..0000000000000000000000000000000000000000
--- a/workload-generator-common/src/main/java/common/dimensions/copy2/KeySpace.java
+++ /dev/null
@@ -1,56 +0,0 @@
-package common.dimensions.copy2;
-
-import common.generators.WorkloadGenerator;
-
-/**
- * Wrapper class for the definition of the Keys that should be used by the
- * {@link WorkloadGenerator}.
- */
-public class KeySpace extends Dimension {
-
-  private final String prefix;
-  private final int min;
-  private final int max;
-
-
-  /**
-   * Create a new key space. All keys will have the prefix {@code prefix}. The remaining part of
-   * each key will be determined by a number of the interval ({@code min}, {@code max}-1).
-   *
-   * @param prefix the prefix to use for all keys
-   * @param min the lower bound (inclusive) to start counting from
-   * @param max the upper bound (exclusive) to count to
-   */
-  public KeySpace(final String prefix, final int min, final int max) {
-    if (prefix == null || prefix.contains(";")) {
-      throw new IllegalArgumentException(
-          "The prefix must not be null and must not contain the ';' character.");
-    }
-    this.prefix = prefix;
-    this.min = min;
-    this.max = max;
-
-  }
-
-  public KeySpace(final String prefix, final int numberOfKeys) {
-    this(prefix, 0, numberOfKeys - 1);
-  }
-
-  public KeySpace(final int numberOfKeys) {
-    this("sensor_", 0, numberOfKeys - 1);
-  }
-
-  public String getPrefix() {
-    return this.prefix;
-  }
-
-
-  public int getMin() {
-    return this.min;
-  }
-
-
-  public int getMax() {
-    return this.max;
-  }
-}
diff --git a/workload-generator-common/src/main/java/common/dimensions/copy2/Period.java b/workload-generator-common/src/main/java/common/dimensions/copy2/Period.java
deleted file mode 100644
index d89a426156a4eda2edd90a52164633238872787b..0000000000000000000000000000000000000000
--- a/workload-generator-common/src/main/java/common/dimensions/copy2/Period.java
+++ /dev/null
@@ -1,34 +0,0 @@
-package common.dimensions.copy2;
-
-import java.util.concurrent.TimeUnit;
-import common.generators.WorkloadGenerator;
-
-/**
- * Wrapper class for the definition of period to use for the {@link WorkloadGenerator}.
- */
-public class Period extends Dimension {
-
-  private final int period;
-  private final TimeUnit timeUnit;
-
-  /**
-   * Define a new period.
-   * 
-   * @param period the period
-   * @param timeUnit the time unit that applies to the specified {@code period}
-   */
-  public Period(final int period, final TimeUnit timeUnit) {
-    super();
-    this.period = period;
-    this.timeUnit = timeUnit;
-  }
-
-  public int getDuration() {
-    return this.period;
-  }
-
-  public TimeUnit getTimeUnit() {
-    return this.timeUnit;
-  }
-
-}
diff --git a/workload-generator-common/src/main/java/common/functions/copy/BeforeAction.java b/workload-generator-common/src/main/java/common/functions/copy/BeforeAction.java
deleted file mode 100644
index 55542255ac3b5dff2b3e69676be1e42299c0d88f..0000000000000000000000000000000000000000
--- a/workload-generator-common/src/main/java/common/functions/copy/BeforeAction.java
+++ /dev/null
@@ -1,8 +0,0 @@
-package common.functions.copy;
-
-@FunctionalInterface
-public interface BeforeAction {
-
-  public void run();
-  
-}
diff --git a/workload-generator-common/src/main/java/common/functions/copy/MessageGenerator.java b/workload-generator-common/src/main/java/common/functions/copy/MessageGenerator.java
deleted file mode 100644
index fb347210d961c955ca9cd90b7f17204f3c234085..0000000000000000000000000000000000000000
--- a/workload-generator-common/src/main/java/common/functions/copy/MessageGenerator.java
+++ /dev/null
@@ -1,11 +0,0 @@
-package common.functions.copy;
-
-import common.messages.OutputMessage;
-import kieker.common.record.IMonitoringRecord;
-
-@FunctionalInterface
-public interface MessageGenerator<T extends IMonitoringRecord> {
-
-  OutputMessage<T> generateMessage(final String key);
-
-}
diff --git a/workload-generator-common/src/main/java/common/functions/copy/Transport.java b/workload-generator-common/src/main/java/common/functions/copy/Transport.java
deleted file mode 100644
index 6b76a1791cbcf954f930849e0e26fdc710d8ac11..0000000000000000000000000000000000000000
--- a/workload-generator-common/src/main/java/common/functions/copy/Transport.java
+++ /dev/null
@@ -1,11 +0,0 @@
-package common.functions.copy;
-
-import common.messages.OutputMessage;
-import kieker.common.record.IMonitoringRecord;
-
-@FunctionalInterface
-public interface Transport<T extends IMonitoringRecord> {
-  
-  public void transport(final OutputMessage<T> message);
-
-}
diff --git a/workload-generator-common/src/main/java/common/generators/KafkaWorkloadGenerator.java b/workload-generator-common/src/main/java/common/generators/KafkaWorkloadGenerator.java
index 42f26db80de6ea8e3a0ccefa8b6b3252d3fea16a..66553cc1982397d329775b20a42db4336395e9f5 100644
--- a/workload-generator-common/src/main/java/common/generators/KafkaWorkloadGenerator.java
+++ b/workload-generator-common/src/main/java/common/generators/KafkaWorkloadGenerator.java
@@ -5,6 +5,7 @@ import common.dimensions.KeySpace;
 import common.dimensions.Period;
 import common.functions.BeforeAction;
 import common.functions.MessageGenerator;
+import common.misc.ZooKeeper;
 import communication.kafka.KafkaRecordSender;
 import kieker.common.record.IMonitoringRecord;
 
@@ -19,6 +20,7 @@ public class KafkaWorkloadGenerator<T extends IMonitoringRecord> extends Workloa
    * Create a new workload generator.
    *
    * @param keySpace the key space to generate the workload for.
+   * @param threads tha amount of threads to use per instance.
    * @param period the period how often a message is generated for each key specified in the
    *        {@code keySpace}
    * @param duration the duration how long the workload generator will emit messages.
@@ -29,22 +31,23 @@ public class KafkaWorkloadGenerator<T extends IMonitoringRecord> extends Workloa
    * @param recordSender the record sender which is used to send the generated messages to kafka.
    */
   public KafkaWorkloadGenerator(
+      final ZooKeeper zooKeeper,
       final KeySpace keySpace,
+      final int threads,
       final Period period,
       final Duration duration,
       final BeforeAction beforeAction,
       final MessageGenerator<T> generatorFunction,
       final KafkaRecordSender<T> recordSender) {
-    super(keySpace, period, duration, beforeAction, generatorFunction, o -> {
-      System.out.println(o.getKey());
-    });
+    super(zooKeeper, keySpace, threads, period, duration, beforeAction, generatorFunction,
+        recordSender);
     this.recordSender = recordSender;
   }
 
 
   @Override
   public void stop() {
-    // this.recordSender.terminate();
+    this.recordSender.terminate();
 
     super.stop();
   }
diff --git a/workload-generator-common/src/main/java/common/generators/KafkaWorkloadGeneratorBuilder.java b/workload-generator-common/src/main/java/common/generators/KafkaWorkloadGeneratorBuilder.java
index 38132686d847a889cdde0165f6825ed844d46312..d6ce8894a5675aa208ce6d804dc0a51ba5d501ae 100644
--- a/workload-generator-common/src/main/java/common/generators/KafkaWorkloadGeneratorBuilder.java
+++ b/workload-generator-common/src/main/java/common/generators/KafkaWorkloadGeneratorBuilder.java
@@ -6,13 +6,18 @@ import common.dimensions.KeySpace;
 import common.dimensions.Period;
 import common.functions.BeforeAction;
 import common.functions.MessageGenerator;
+import common.misc.ZooKeeper;
 import communication.kafka.KafkaRecordSender;
 import kieker.common.record.IMonitoringRecord;
 
 public class KafkaWorkloadGeneratorBuilder<T extends IMonitoringRecord> {
 
+  private ZooKeeper zooKeeper;
+
   private KeySpace keySpace;
 
+  private int threads;
+
   private Period period;
 
   private Duration duration;
@@ -32,10 +37,15 @@ public class KafkaWorkloadGeneratorBuilder<T extends IMonitoringRecord> {
    *
    * @return the builder.
    */
-  public static KafkaWorkloadGeneratorBuilder<IMonitoringRecord> builder() {
+  public static <T extends IMonitoringRecord> KafkaWorkloadGeneratorBuilder<T> builder() {
     return new KafkaWorkloadGeneratorBuilder<>();
   }
 
+  public KafkaWorkloadGeneratorBuilder<T> setZooKeeper(final ZooKeeper zooKeeper) {
+    this.zooKeeper = zooKeeper;
+    return this;
+  }
+
   /**
    * Set the before action for the {@link KafkaWorkloadGenerator}.
    *
@@ -58,6 +68,17 @@ public class KafkaWorkloadGeneratorBuilder<T extends IMonitoringRecord> {
     return this;
   }
 
+  /**
+   * Set the key space for the {@link KafkaWorkloadGenerator}.
+   *
+   * @param keySpace the {@link KeySpace}.
+   * @return the builder.
+   */
+  public KafkaWorkloadGeneratorBuilder<T> setThreads(final int threads) {
+    this.threads = threads;
+    return this;
+  }
+
   /**
    * Set the period for the {@link KafkaWorkloadGenerator}.
    *
@@ -118,15 +139,24 @@ public class KafkaWorkloadGeneratorBuilder<T extends IMonitoringRecord> {
    * @return the built instance of the {@link KafkaWorkloadGenerator}.
    */
   public KafkaWorkloadGenerator<T> build() {
+    Objects.requireNonNull(this.zooKeeper, "Please specify the ZooKeeper instance.");
+    this.threads = Objects.requireNonNullElse(this.threads, 1);
     Objects.requireNonNull(this.keySpace, "Please specify the key space.");
     Objects.requireNonNull(this.period, "Please specify the period.");
     Objects.requireNonNull(this.duration, "Please specify the duration.");
-    final BeforeAction beforeAction = Objects.requireNonNullElse(this.beforeAction, () -> {
+    this.beforeAction = Objects.requireNonNullElse(this.beforeAction, () -> {
     });
     Objects.requireNonNull(this.generatorFunction, "Please specify the generator function.");
-    // Objects.requireNonNull(this.kafkaRecordSender, "Please specify the kafka record sender.");
-
-    return new KafkaWorkloadGenerator<>(this.keySpace, this.period, this.duration, beforeAction,
-        this.generatorFunction, this.kafkaRecordSender);
+    Objects.requireNonNull(this.kafkaRecordSender, "Please specify the kafka record sender.");
+
+    return new KafkaWorkloadGenerator<>(
+        this.zooKeeper,
+        this.keySpace,
+        this.threads,
+        this.period,
+        this.duration,
+        this.beforeAction,
+        this.generatorFunction,
+        this.kafkaRecordSender);
   }
 }
diff --git a/workload-generator-common/src/main/java/common/generators/WorkloadGenerator.java b/workload-generator-common/src/main/java/common/generators/WorkloadGenerator.java
index b012c812f84bf934bbf274cbab228e35b6e114cf..859a19833af3a289ccf83221dd1c288e97fcd99e 100644
--- a/workload-generator-common/src/main/java/common/generators/WorkloadGenerator.java
+++ b/workload-generator-common/src/main/java/common/generators/WorkloadGenerator.java
@@ -7,6 +7,8 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.function.BiConsumer;
 import java.util.function.BiFunction;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import common.dimensions.Duration;
 import common.dimensions.KeySpace;
 import common.dimensions.Period;
@@ -17,13 +19,20 @@ import common.messages.OutputMessage;
 import common.misc.Worker;
 import common.misc.WorkloadDefinition;
 import common.misc.WorkloadEntity;
+import common.misc.ZooKeeper;
 import communication.zookeeper.WorkloadDistributor;
 import kieker.common.record.IMonitoringRecord;
 
 public abstract class WorkloadGenerator<T extends IMonitoringRecord> implements IWorkloadGenerator {
 
+  private static final Logger LOGGER = LoggerFactory.getLogger(WorkloadGenerator.class);
+
+  private final ZooKeeper zooKeeper;
+
   private final KeySpace keySpace;
 
+  private final int threads;
+
   private final Period period;
 
   private final Duration duration;
@@ -42,7 +51,7 @@ public abstract class WorkloadGenerator<T extends IMonitoringRecord> implements
 
   /**
    * Start the workload generation. The generation terminates automatically after the specified
-   * {@code duration}.s
+   * {@code duration}.
    */
   @Override
   public void start() {
@@ -55,13 +64,17 @@ public abstract class WorkloadGenerator<T extends IMonitoringRecord> implements
   }
 
   public WorkloadGenerator(
+      final ZooKeeper zooKeeper,
       final KeySpace keySpace,
+      final int threads,
       final Period period,
       final Duration duration,
       final BeforeAction beforeAction,
       final MessageGenerator<T> generatorFunction,
       final Transport<T> transport) {
+    this.zooKeeper = zooKeeper;
     this.period = period;
+    this.threads = threads;
     this.keySpace = keySpace;
     this.duration = duration;
     this.beforeAction = beforeAction;
@@ -80,29 +93,27 @@ public abstract class WorkloadGenerator<T extends IMonitoringRecord> implements
     };
     this.transport = transport;
 
-    final int threads = 10; // env
     this.executor = Executors.newScheduledThreadPool(threads);
     final Random random = new Random();
 
-    final int periodMs = period.getDuration();
+    final int periodMs = period.getPeriod();
 
     final BiConsumer<WorkloadDefinition, Worker> workerAction = (declaration, worker) -> {
 
       final List<WorkloadEntity<T>> entities = this.workloadSelector.apply(declaration, worker);
 
-      System.out.println("Beginning of Experiment...");
-      System.out.println("Experiment is going to be executed for the specified duration...");
+      LOGGER.info("Beginning of Experiment...");
+      LOGGER.info("Experiment is going to be executed for the specified duration...");
       entities.forEach(entity -> {
         final OutputMessage<T> message = entity.generateMessage();
         final long initialDelay = random.nextInt(periodMs);
         this.executor.scheduleAtFixedRate(() -> this.transport.transport(message), initialDelay,
             periodMs, period.getTimeUnit());
-
       });
 
       try {
         this.executor.awaitTermination(duration.getDuration(), duration.getTimeUnit());
-        System.out.println("Terminating now...");
+        LOGGER.info("Terminating now...");
         this.stop();
       } catch (final InterruptedException e) {
         // TODO Auto-generated catch block
@@ -111,6 +122,6 @@ public abstract class WorkloadGenerator<T extends IMonitoringRecord> implements
     };
 
     this.workloadDistributor =
-        new WorkloadDistributor(this.keySpace, this.beforeAction, workerAction);
+        new WorkloadDistributor(this.zooKeeper, this.keySpace, this.beforeAction, workerAction);
   }
 }
diff --git a/workload-generator-common/src/main/java/common/generators/copy/IWorkloadGenerator.java b/workload-generator-common/src/main/java/common/generators/copy/IWorkloadGenerator.java
deleted file mode 100644
index 5d5c9170e6439068138b56eafd649fbcc0adf8a8..0000000000000000000000000000000000000000
--- a/workload-generator-common/src/main/java/common/generators/copy/IWorkloadGenerator.java
+++ /dev/null
@@ -1,18 +0,0 @@
-package common.generators.copy;
-
-/**
- * Base methods for workload generators.
- */
-public interface IWorkloadGenerator {
-
-  /**
-   * Start the workload generation.
-   */
-  void start();
-
-  /**
-   * Stop the workload generation.
-   */
-  void stop();
-
-}
diff --git a/workload-generator-common/src/main/java/common/generators/copy/KafkaWorkloadGenerator.java b/workload-generator-common/src/main/java/common/generators/copy/KafkaWorkloadGenerator.java
deleted file mode 100644
index 4e954a080986ae232a6d333a5db88ede31f5963e..0000000000000000000000000000000000000000
--- a/workload-generator-common/src/main/java/common/generators/copy/KafkaWorkloadGenerator.java
+++ /dev/null
@@ -1,51 +0,0 @@
-package common.generators.copy;
-
-import common.dimensions.Duration;
-import common.dimensions.KeySpace;
-import common.dimensions.Period;
-import common.functions.BeforeAction;
-import common.functions.MessageGenerator;
-import communication.kafka.KafkaRecordSender;
-import kieker.common.record.IMonitoringRecord;
-
-/**
- * Workload generator for generating load for the kafka messaging system.
- */
-public class KafkaWorkloadGenerator<T extends IMonitoringRecord> extends WorkloadGenerator<T> {
-
-  private final KafkaRecordSender<T> recordSender;
-
-  /**
-   * Create a new workload generator.
-   *
-   * @param keySpace the key space to generate the workload for.
-   * @param period the period how often a message is generated for each key specified in the
-   *        {@code keySpace}
-   * @param duration the duration how long the workload generator will emit messages.
-   * @param beforeAction the action which will be performed before the workload generator starts
-   *        generating messages. If {@code null}, no before action will be performed.
-   * @param generatorFunction the generator function. This function is executed, each time a message
-   *        is generated.
-   * @param recordSender the record sender which is used to send the generated messages to kafka.
-   */
-  public KafkaWorkloadGenerator(
-      final KeySpace keySpace,
-      final Period period,
-      final Duration duration,
-      final BeforeAction beforeAction,
-      final MessageGenerator<T> generatorFunction,
-      final KafkaRecordSender<T> recordSender) {
-    super(keySpace, period, duration, beforeAction, generatorFunction, o -> {
-      System.out.println(o.getKey());
-    });
-    this.recordSender = recordSender;
-  }
-
-
-  @Override
-  public void stop() {
-    // this.recordSender.terminate();
-
-    super.stop();
-  }
-}
diff --git a/workload-generator-common/src/main/java/common/generators/copy/KafkaWorkloadGeneratorBuilder.java b/workload-generator-common/src/main/java/common/generators/copy/KafkaWorkloadGeneratorBuilder.java
deleted file mode 100644
index c4fb96f2c000d00e1381bcbc675c9b15e31d3fa5..0000000000000000000000000000000000000000
--- a/workload-generator-common/src/main/java/common/generators/copy/KafkaWorkloadGeneratorBuilder.java
+++ /dev/null
@@ -1,132 +0,0 @@
-package common.generators.copy;
-
-import java.util.Objects;
-import common.dimensions.Duration;
-import common.dimensions.KeySpace;
-import common.dimensions.Period;
-import common.functions.BeforeAction;
-import common.functions.MessageGenerator;
-import communication.kafka.KafkaRecordSender;
-import kieker.common.record.IMonitoringRecord;
-
-public class KafkaWorkloadGeneratorBuilder<T extends IMonitoringRecord> {
-
-  private KeySpace keySpace;
-
-  private Period period;
-
-  private Duration duration;
-
-  private BeforeAction beforeAction;
-
-  private MessageGenerator<T> generatorFunction;
-
-  private KafkaRecordSender<T> kafkaRecordSender;
-
-  private KafkaWorkloadGeneratorBuilder() {
-
-  }
-
-  /**
-   * Get a builder for the {@link KafkaWorkloadGenerator}.
-   *
-   * @return the builder.
-   */
-  public static KafkaWorkloadGeneratorBuilder<IMonitoringRecord> builder() {
-    return new KafkaWorkloadGeneratorBuilder<>();
-  }
-
-  /**
-   * Set the before action for the {@link KafkaWorkloadGenerator}.
-   *
-   * @param beforeAction the {@link BeforeAction}.
-   * @return the builder.
-   */
-  public KafkaWorkloadGeneratorBuilder<T> setBeforeAction(final BeforeAction beforeAction) {
-    this.beforeAction = beforeAction;
-    return this;
-  }
-
-  /**
-   * Set the key space for the {@link KafkaWorkloadGenerator}.
-   *
-   * @param keySpace the {@link KeySpace}.
-   * @return the builder.
-   */
-  public KafkaWorkloadGeneratorBuilder<T> setKeySpace(final KeySpace keySpace) {
-    this.keySpace = keySpace;
-    return this;
-  }
-
-  /**
-   * Set the period for the {@link KafkaWorkloadGenerator}.
-   *
-   * @param period the {@link Period}
-   * @return the builder.
-   */
-  public KafkaWorkloadGeneratorBuilder<T> setPeriod(final Period period) {
-    this.period = period;
-    return this;
-  }
-
-  /**
-   * Set the durtion for the {@link KafkaWorkloadGenerator}.
-   *
-   * @param duration the {@link Duration}.
-   * @return the builder.
-   */
-  public KafkaWorkloadGeneratorBuilder<T> setDuration(final Duration duration) {
-    this.duration = duration;
-    return this;
-  }
-
-  /**
-   * Set the generator function for the {@link KafkaWorkloadGenerator}.
-   *
-   * @param generatorFunction the generator function.
-   * @return the builder.
-   */
-  public KafkaWorkloadGeneratorBuilder<T> setGeneratorFunction(
-      final MessageGenerator<T> generatorFunction) {
-    this.generatorFunction = generatorFunction;
-    return this;
-  }
-
-  /**
-   * Set the {@link KafkaRecordSender} for the {@link KafkaWorkloadGenerator}.
-   *
-   * @param kafkaRecordSender the record sender to use.
-   * @return the builder.
-   */
-  public KafkaWorkloadGeneratorBuilder<T> setKafkaRecordSender(
-      final KafkaRecordSender<T> kafkaRecordSender) {
-    this.kafkaRecordSender = kafkaRecordSender;
-    return this;
-  }
-
-  /**
-   * Build the actual {@link KafkaWorkloadGenerator}. The following parameters are must be
-   * specicified before this method is called:
-   * <ul>
-   * <li>key space</li>
-   * <li>period</li>
-   * <li>duration</li>
-   * <li>generator function</li>
-   * <li>kafka record sender</li>
-   * </ul>
-   *
-   * @return the built instance of the {@link KafkaWorkloadGenerator}.
-   */
-  public KafkaWorkloadGenerator<T> build() {
-    Objects.requireNonNull(this.keySpace, "Please specify the key space.");
-    Objects.requireNonNull(this.period, "Please specify the period.");
-    Objects.requireNonNull(this.duration, "Please specify the duration.");
-    final BeforeAction beforeAction = Objects.requireNonNullElse(this.beforeAction, () -> {
-    });
-    Objects.requireNonNull(this.generatorFunction, "Please specify the generator function.");
-    // Objects.requireNonNull(this.kafkaRecordSender, "Please specify the kafka record sender.");
-
-    return new KafkaWorkloadGenerator<>(this.keySpace, this.period, this.duration, beforeAction,
-        this.generatorFunction, this.kafkaRecordSender);
-  }
-}
diff --git a/workload-generator-common/src/main/java/common/generators/copy/WorkloadGenerator.java b/workload-generator-common/src/main/java/common/generators/copy/WorkloadGenerator.java
deleted file mode 100644
index 4e56eade9e26bbd3c5caf343a097b9371e153054..0000000000000000000000000000000000000000
--- a/workload-generator-common/src/main/java/common/generators/copy/WorkloadGenerator.java
+++ /dev/null
@@ -1,116 +0,0 @@
-package common.generators.copy;
-
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Random;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.function.BiConsumer;
-import java.util.function.BiFunction;
-import common.dimensions.Duration;
-import common.dimensions.KeySpace;
-import common.dimensions.Period;
-import common.functions.BeforeAction;
-import common.functions.MessageGenerator;
-import common.functions.Transport;
-import common.messages.OutputMessage;
-import common.misc.Worker;
-import common.misc.WorkloadDefinition;
-import common.misc.WorkloadEntity;
-import communication.zookeeper.WorkloadDistributor;
-import kieker.common.record.IMonitoringRecord;
-
-public abstract class WorkloadGenerator<T extends IMonitoringRecord> implements IWorkloadGenerator {
-
-  private final KeySpace keySpace;
-
-  private final Period period;
-
-  private final Duration duration;
-
-  private final BeforeAction beforeAction;
-
-  private final BiFunction<WorkloadDefinition, Worker, List<WorkloadEntity<T>>> workloadSelector;
-
-  private final MessageGenerator<T> generatorFunction;
-
-  private final Transport<T> transport;
-
-  private WorkloadDistributor workloadDistributor;
-
-  private final ScheduledExecutorService executor;
-
-  /**
-   * Start the workload generation. The generation terminates automatically after the specified
-   * {@code duration}.s
-   */
-  @Override
-  public void start() {
-    this.workloadDistributor.start();
-  }
-
-  @Override
-  public void stop() {
-    this.workloadDistributor.stop();
-  }
-
-  public WorkloadGenerator(
-      final KeySpace keySpace,
-      final Period period,
-      final Duration duration,
-      final BeforeAction beforeAction,
-      final MessageGenerator<T> generatorFunction,
-      final Transport<T> transport) {
-    this.period = period;
-    this.keySpace = keySpace;
-    this.duration = duration;
-    this.beforeAction = beforeAction;
-    this.generatorFunction = generatorFunction;
-    this.workloadSelector = (workloadDeclaration, worker) -> {
-      final List<WorkloadEntity<T>> workloadEntities = new LinkedList<>();
-
-      for (int i =
-          workloadDeclaration.getKeySpace().getMin() + worker.getId(); i <= workloadDeclaration
-              .getKeySpace().getMax(); i += workloadDeclaration.getNumberOfWorkers()) {
-        final String id = workloadDeclaration.getKeySpace().getPrefix() + i;
-        workloadEntities.add(new WorkloadEntity<>(id, this.generatorFunction));
-      }
-
-      return workloadEntities;
-    };
-    this.transport = transport;
-
-    final int threads = 10; // env
-    this.executor = Executors.newScheduledThreadPool(threads);
-    final Random random = new Random();
-
-    final int periodMs = period.getDuration();
-
-    final BiConsumer<WorkloadDefinition, Worker> workerAction = (declaration, worker) -> {
-
-      final List<WorkloadEntity<T>> entities = this.workloadSelector.apply(declaration, worker);
-
-      System.out.println("Beginning of Experiment...");
-      System.out.println("Experiment is going to be executed for the specified duration...");
-      entities.forEach(entity -> {
-        final OutputMessage<T> message = entity.generateMessage();
-        final long initialDelay = random.nextInt(periodMs);
-        this.executor.scheduleAtFixedRate(() -> this.transport.transport(message), initialDelay,
-            periodMs, period.getTimeUnit());
-
-      });
-
-      try {
-        this.executor.awaitTermination(duration.getDuration(), duration.getTimeUnit());
-        System.out.println("Terminating now...");
-        this.stop();
-      } catch (final InterruptedException e) {
-        // TODO Auto-generated catch block
-        e.printStackTrace();
-      }
-    };
-
-    this.workloadDistributor =
-        new WorkloadDistributor(this.keySpace, this.beforeAction, workerAction);
-  }
-}
diff --git a/workload-generator-common/src/main/java/common/messages/copy/OutputMessage.java b/workload-generator-common/src/main/java/common/messages/copy/OutputMessage.java
deleted file mode 100644
index 174ae8a126378d48ceed1495b0b341148b41bc73..0000000000000000000000000000000000000000
--- a/workload-generator-common/src/main/java/common/messages/copy/OutputMessage.java
+++ /dev/null
@@ -1,32 +0,0 @@
-package common.messages.copy;
-
-import kieker.common.record.IMonitoringRecord;
-
-/*
- * Wrapper class for messages within the messaging system.
- */
-public class OutputMessage<T extends IMonitoringRecord> {
-  private final String key;
-  private final T value;
-
-  /***
-   * Create a new Message.
-   * 
-   * @param key the key of the message.
-   * @param value the value of the message.
-   */
-  public OutputMessage(final String key, final T value) {
-    super();
-    this.key = key;
-    this.value = value;
-  }
-
-  public String getKey() {
-    return this.key;
-  }
-
-  public T getValue() {
-    return this.value;
-  }
-
-}
diff --git a/workload-generator-common/src/main/java/common/messages/copy2/OutputMessage.java b/workload-generator-common/src/main/java/common/messages/copy2/OutputMessage.java
deleted file mode 100644
index 87da409ec847d1be7a3a261952a56dd6c61c6949..0000000000000000000000000000000000000000
--- a/workload-generator-common/src/main/java/common/messages/copy2/OutputMessage.java
+++ /dev/null
@@ -1,32 +0,0 @@
-package common.messages.copy2;
-
-import kieker.common.record.IMonitoringRecord;
-
-/*
- * Wrapper class for messages within the messaging system.
- */
-public class OutputMessage<T extends IMonitoringRecord> {
-  private final String key;
-  private final T value;
-
-  /***
-   * Create a new Message.
-   * 
-   * @param key the key of the message.
-   * @param value the value of the message.
-   */
-  public OutputMessage(final String key, final T value) {
-    super();
-    this.key = key;
-    this.value = value;
-  }
-
-  public String getKey() {
-    return this.key;
-  }
-
-  public T getValue() {
-    return this.value;
-  }
-
-}
diff --git a/workload-generator-common/src/main/java/common/misc/ZooKeeper.java b/workload-generator-common/src/main/java/common/misc/ZooKeeper.java
new file mode 100644
index 0000000000000000000000000000000000000000..e1698e8c5cf7ad66382e200512020f9446b51ca1
--- /dev/null
+++ b/workload-generator-common/src/main/java/common/misc/ZooKeeper.java
@@ -0,0 +1,30 @@
+package common.misc;
+
+/*
+ * Wrapper for connection information for ZooKeeper.
+ */
+public class ZooKeeper {
+
+  private final String host;
+  private final int port;
+
+  /**
+   * Create a new representation of an ZooKeeper instance.
+   *
+   * @param host of zookeeper.
+   * @param port of zookeeper.
+   */
+  public ZooKeeper(final String host, final int port) {
+    super();
+    this.host = host;
+    this.port = port;
+  }
+
+  public String getHost() {
+    return this.host;
+  }
+
+  public int getPort() {
+    return this.port;
+  }
+}
diff --git a/workload-generator-common/src/main/java/common/misc/copy/Worker.java b/workload-generator-common/src/main/java/common/misc/copy/Worker.java
deleted file mode 100644
index 8077c0797dead702ab7d666aebc1fe4495484bc9..0000000000000000000000000000000000000000
--- a/workload-generator-common/src/main/java/common/misc/copy/Worker.java
+++ /dev/null
@@ -1,24 +0,0 @@
-package common.misc.copy;
-
-/*
- * Wrapper class for a worker.
- */
-public class Worker {
-
-  private final int id;
-
-  /**
-   * Create a new worker with an {@code id}
-   * 
-   * @param id the id of the worker.
-   */
-  public Worker(final int id) {
-    super();
-    this.id = id;
-  }
-
-  public int getId() {
-    return this.id;
-  }
-
-}
diff --git a/workload-generator-common/src/main/java/common/misc/copy/WorkloadDefinition.java b/workload-generator-common/src/main/java/common/misc/copy/WorkloadDefinition.java
deleted file mode 100644
index 672812a2f9f45d10a7891377960bb374a8ff502c..0000000000000000000000000000000000000000
--- a/workload-generator-common/src/main/java/common/misc/copy/WorkloadDefinition.java
+++ /dev/null
@@ -1,63 +0,0 @@
-package common.misc.copy;
-
-import common.dimensions.KeySpace;
-
-/*
- * The central class that contains all information that needs to be exchanged between the nodes for
- * distributed workload generation.
- */
-public class WorkloadDefinition {
-  private final KeySpace keySpace;
-  private final int numberOfWorkers;
-
-  /**
-   * Create a new workload definition.
-   *
-   * @param keySpace the key space to use.
-   * @param numberOfWorkers the number of workers participating in the workload generation.
-   */
-  public WorkloadDefinition(final KeySpace keySpace, final int numberOfWorkers) {
-
-    this.keySpace = keySpace;
-    this.numberOfWorkers = numberOfWorkers;
-  }
-
-  public KeySpace getKeySpace() {
-    return this.keySpace;
-  }
-
-  public int getNumberOfWorkers() {
-    return this.numberOfWorkers;
-  }
-
-  /**
-   * Simple method for encoding all information of the workload definition into one string.
-   *
-   * @return a string that encodes all information of the workload generation in a compact format.
-   *         The format is 'keySpace;keySpace.min;keySpace.max;numberOfWorkers'.
-   */
-  @Override
-  public String toString() {
-    return this.getKeySpace().getPrefix() + ";" + this.getKeySpace().getMin() + ";"
-        + this.getKeySpace().getMax() + ";" + this.getNumberOfWorkers();
-  }
-
-  /**
-   * Parse a workload generation from a previously encoded string with the format returned by
-   * {@link WorkloadDefinition#toString()}.
-   *
-   * @param workloadDefinitionString the workload definition string.
-   * @return the parsed workload definition.
-   */
-  public static WorkloadDefinition fromString(final String workloadDefinitionString) {
-    final String[] deserialized = workloadDefinitionString.split(";");
-
-    if (deserialized.length != 4) {
-      throw new IllegalArgumentException(
-          "Wrong workload definition string when trying to parse the workload generation.");
-    }
-
-    return new WorkloadDefinition(new KeySpace(deserialized[0], Integer.valueOf(deserialized[1]),
-        Integer.valueOf(deserialized[2])), Integer.valueOf(deserialized[3]));
-  }
-}
diff --git a/workload-generator-common/src/main/java/common/misc/copy/WorkloadEntity.java b/workload-generator-common/src/main/java/common/misc/copy/WorkloadEntity.java
deleted file mode 100644
index 6b44fe45dafb77e797dc9c7cee22ef59225f32f6..0000000000000000000000000000000000000000
--- a/workload-generator-common/src/main/java/common/misc/copy/WorkloadEntity.java
+++ /dev/null
@@ -1,19 +0,0 @@
-package common.misc.copy;
-
-import common.functions.MessageGenerator;
-import common.messages.OutputMessage;
-import kieker.common.record.IMonitoringRecord;
-
-public class WorkloadEntity<T extends IMonitoringRecord> {
-  private final String key;
-  private final MessageGenerator<T> generator;
-
-  public WorkloadEntity(final String key, final MessageGenerator<T> generator) {
-    this.key = key;
-    this.generator = generator;
-  }
-
-  public OutputMessage<T> generateMessage() {
-    return this.generator.generateMessage(this.key);
-  }
-}
diff --git a/workload-generator-common/src/main/java/communication/kafka/KafkaRecordSender.java b/workload-generator-common/src/main/java/communication/kafka/KafkaRecordSender.java
index 5af3fc0d5416093740584a3b69cfb1a3fbc24883..8a0fb026695d0cc998d04f3d612bced89ae73027 100644
--- a/workload-generator-common/src/main/java/communication/kafka/KafkaRecordSender.java
+++ b/workload-generator-common/src/main/java/communication/kafka/KafkaRecordSender.java
@@ -2,7 +2,6 @@ package communication.kafka;
 
 import java.util.Properties;
 import java.util.function.Function;
-import kieker.common.record.IMonitoringRecord;
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerRecord;
@@ -11,6 +10,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import common.functions.Transport;
 import common.messages.OutputMessage;
+import kieker.common.record.IMonitoringRecord;
 import titan.ccp.common.kieker.kafka.IMonitoringRecordSerde;
 
 
@@ -84,7 +84,8 @@ public class KafkaRecordSender<T extends IMonitoringRecord> implements Transport
   }
 
   @Override
-  public void transport(OutputMessage<T> message) {
+  public void transport(final OutputMessage<T> message) {
+    System.out.println(message.getKey());
     this.write(message.getValue());
   }
 
diff --git a/workload-generator-common/src/main/java/communication/kafka/copy/KafkaRecordSender.java b/workload-generator-common/src/main/java/communication/kafka/copy/KafkaRecordSender.java
deleted file mode 100644
index 168e34e0b3e2bc872b20fcd057471c68a0263925..0000000000000000000000000000000000000000
--- a/workload-generator-common/src/main/java/communication/kafka/copy/KafkaRecordSender.java
+++ /dev/null
@@ -1,91 +0,0 @@
-package communication.kafka.copy;
-
-import java.util.Properties;
-import java.util.function.Function;
-import kieker.common.record.IMonitoringRecord;
-import org.apache.kafka.clients.producer.KafkaProducer;
-import org.apache.kafka.clients.producer.Producer;
-import org.apache.kafka.clients.producer.ProducerRecord;
-import org.apache.kafka.common.serialization.StringSerializer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import common.functions.Transport;
-import common.messages.OutputMessage;
-import titan.ccp.common.kieker.kafka.IMonitoringRecordSerde;
-
-
-/**
- * Sends monitoring records to Kafka.
- *
- * @param <T> {@link IMonitoringRecord} to send
- */
-public class KafkaRecordSender<T extends IMonitoringRecord> implements Transport<T> {
-
-  private static final Logger LOGGER = LoggerFactory.getLogger(KafkaRecordSender.class);
-
-  private final String topic;
-
-  private final Function<T, String> keyAccessor;
-
-  private final Function<T, Long> timestampAccessor;
-
-  private final Producer<String, T> producer;
-
-  public KafkaRecordSender(final String bootstrapServers, final String topic) {
-    this(bootstrapServers, topic, x -> "", x -> null, new Properties());
-  }
-
-  public KafkaRecordSender(final String bootstrapServers, final String topic,
-      final Function<T, String> keyAccessor) {
-    this(bootstrapServers, topic, keyAccessor, x -> null, new Properties());
-  }
-
-  public KafkaRecordSender(final String bootstrapServers, final String topic,
-      final Function<T, String> keyAccessor, final Function<T, Long> timestampAccessor) {
-    this(bootstrapServers, topic, keyAccessor, timestampAccessor, new Properties());
-  }
-
-  /**
-   * Create a new {@link KafkaRecordSender}.
-   */
-  public KafkaRecordSender(final String bootstrapServers, final String topic,
-      final Function<T, String> keyAccessor, final Function<T, Long> timestampAccessor,
-      final Properties defaultProperties) {
-    this.topic = topic;
-    this.keyAccessor = keyAccessor;
-    this.timestampAccessor = timestampAccessor;
-
-    final Properties properties = new Properties();
-    properties.putAll(defaultProperties);
-    properties.put("bootstrap.servers", bootstrapServers);
-    // properties.put("acks", this.acknowledges);
-    // properties.put("batch.size", this.batchSize);
-    // properties.put("linger.ms", this.lingerMs);
-    // properties.put("buffer.memory", this.bufferMemory);
-
-    this.producer = new KafkaProducer<>(properties, new StringSerializer(),
-        IMonitoringRecordSerde.serializer());
-  }
-
-  /**
-   * Write the passed monitoring record to Kafka.
-   */
-  public void write(final T monitoringRecord) {
-    final ProducerRecord<String, T> record =
-        new ProducerRecord<>(this.topic, null, this.timestampAccessor.apply(monitoringRecord),
-            this.keyAccessor.apply(monitoringRecord), monitoringRecord);
-
-    LOGGER.debug("Send record to Kafka topic {}: {}", this.topic, record);
-    this.producer.send(record);
-  }
-
-  public void terminate() {
-    this.producer.close();
-  }
-
-  @Override
-  public void transport(OutputMessage<T> message) {
-    this.write(message.getValue());
-  }
-
-}
diff --git a/workload-generator-common/src/main/java/communication/zookeeper/WorkloadDistributor.java b/workload-generator-common/src/main/java/communication/zookeeper/WorkloadDistributor.java
index 8cbba005ebbfefa8805c4e28dbf5e70c8628281a..f677b8f712d59034d424d92e91c0700fdaef4345 100644
--- a/workload-generator-common/src/main/java/communication/zookeeper/WorkloadDistributor.java
+++ b/workload-generator-common/src/main/java/communication/zookeeper/WorkloadDistributor.java
@@ -7,32 +7,37 @@ import org.apache.curator.framework.CuratorFrameworkFactory;
 import org.apache.curator.framework.api.CuratorWatcher;
 import org.apache.curator.framework.recipes.atomic.AtomicValue;
 import org.apache.curator.framework.recipes.atomic.DistributedAtomicInteger;
-import org.apache.curator.retry.RetryNTimes;
+import org.apache.curator.retry.ExponentialBackoffRetry;
 import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.WatchedEvent;
 import org.apache.zookeeper.Watcher.Event.EventType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import common.dimensions.KeySpace;
 import common.functions.BeforeAction;
 import common.misc.Worker;
 import common.misc.WorkloadDefinition;
+import common.misc.ZooKeeper;
 
 /*
  * The central class responsible for distributing the workload through all workload generators.
  */
 public class WorkloadDistributor {
 
+  private static final Logger LOGGER = LoggerFactory.getLogger(WorkloadDistributor.class);
+
+  private static final String NAMESPACE = "workload-generation";
   private static final String COUNTER_PATH = "/counter";
   private static final String WORKLOAD_PATH = "/workload";
   private static final String WORKLOAD_DEFINITION_PATH = "/workload/definition";
 
   private final DistributedAtomicInteger counter;
-
   private final KeySpace keySpace;
   private final BeforeAction beforeAction;
   private final BiConsumer<WorkloadDefinition, Worker> workerAction;
 
-  private final CuratorFramework client =
-      CuratorFrameworkFactory.newClient("127.0.0.1:2181", new RetryNTimes(3, 1000));
+  private final ZooKeeper zooKeeper;
+  private final CuratorFramework client;
 
   /**
    * Create a new workload distributor.
@@ -41,13 +46,22 @@ public class WorkloadDistributor {
    * @param beforeAction the before action for the workload generation.
    * @param workerAction the action to perform by the workers.
    */
-  public WorkloadDistributor(final KeySpace keySpace, final BeforeAction beforeAction,
+  public WorkloadDistributor(
+      final ZooKeeper zooKeeper,
+      final KeySpace keySpace,
+      final BeforeAction beforeAction,
       final BiConsumer<WorkloadDefinition, Worker> workerAction) {
-
+    this.zooKeeper = zooKeeper;
     this.keySpace = keySpace;
     this.beforeAction = beforeAction;
     this.workerAction = workerAction;
 
+    this.client = CuratorFrameworkFactory.builder()
+        .namespace(NAMESPACE)
+        .connectString(this.zooKeeper.getHost() + ":" + this.zooKeeper.getPort())
+        .retryPolicy(new ExponentialBackoffRetry(2000, 5))
+        .build();
+
     this.client.start();
 
     try {
@@ -58,7 +72,8 @@ public class WorkloadDistributor {
     }
 
     this.counter =
-        new DistributedAtomicInteger(this.client, COUNTER_PATH, new RetryNTimes(3, 1000));
+        new DistributedAtomicInteger(this.client, COUNTER_PATH,
+            new ExponentialBackoffRetry(2000, 5));
   }
 
   /**
@@ -78,7 +93,7 @@ public class WorkloadDistributor {
       this.client.checkExists().creatingParentsIfNeeded().forPath(WORKLOAD_DEFINITION_PATH);
 
       if (worker.getId() == 0) {
-        System.out.println("is master with id " + worker.getId());
+        LOGGER.info("This instance is master with id {}", worker.getId());
 
         this.beforeAction.run();
 
@@ -89,7 +104,7 @@ public class WorkloadDistributor {
 
         final int numberOfWorkers = this.counter.get().postValue();
 
-        System.out.printf("Number of Workers: %d\n", numberOfWorkers);
+        LOGGER.info("Number of Workers: {}", numberOfWorkers);
 
         final WorkloadDefinition declaration =
             new WorkloadDefinition(this.keySpace, numberOfWorkers);
@@ -98,7 +113,7 @@ public class WorkloadDistributor {
             declaration.toString().getBytes(StandardCharsets.UTF_8));
 
       } else {
-        System.out.println("is worker with id " + worker.getId());
+        LOGGER.info("This instance is worker with id {}", worker.getId());
 
         this.client.getChildren().usingWatcher(watcher).forPath(WORKLOAD_PATH);
       }
@@ -143,7 +158,6 @@ public class WorkloadDistributor {
    */
   public void stop() {
     this.client.close();
-
   }
 
 }