Improve code quality

theodolite-benchmarks/flink-commons/src/main/java/rocks/theodolite/benchmarks/commons/flink/AbstractFlinkService.java

@@ -60,7 +60,7 @@ public abstract class AbstractFlinkService {
 
   protected void configureCheckpointing() {
     final boolean checkpointing = this.config.getBoolean(ConfigurationKeys.CHECKPOINTING, true);
-    final int commitIntervalMs = this.config.getInt(ConfigurationKeys.COMMIT_INTERVAL_MS);
+    final int commitIntervalMs = this.config.getInt(ConfigurationKeys.CHECKPOINTING_INTERVAL_MS);
     LOGGER.info("Set parallelism to: {}.", checkpointing);
     if (checkpointing) {
       this.env.enableCheckpointing(commitIntervalMs);

theodolite-benchmarks/flink-commons/src/main/java/rocks/theodolite/benchmarks/commons/flink/ConfigurationKeys.java

@@ -23,7 +23,9 @@ public final class ConfigurationKeys {
 
   public static final String WINDOW_GRACE_MS = "window.grace.ms";
 
-  public static final String COMMIT_INTERVAL_MS = "commit.interval.ms";
+  public static final String CHECKPOINTING = "checkpointing";
+
+  public static final String CHECKPOINTING_INTERVAL_MS = "commit.interval.ms";
 
   public static final String FLINK_STATE_BACKEND = "flink.state.backend";
 
@@ -34,8 +36,6 @@ public final class ConfigurationKeys {
 
   public static final String DEBUG = "debug";
 
-  public static final String CHECKPOINTING = "checkpointing";
-
   public static final String PARALLELISM = "parallelism";
 
   private ConfigurationKeys() {}

theodolite-benchmarks/flink-commons/src/main/java/rocks/theodolite/benchmarks/commons/flink/serialization/StatsSerializer.java

@@ -8,7 +8,8 @@ import com.google.common.math.Stats;
 import java.io.Serializable;
 
 /**
- * Custom Kryo {@link Serializer} for efficient transmission between Flink instances.
+ * Custom Kryo {@link Serializer}for {@link Stats} objects for efficient transmission between Flink
+ * instances.
  */
 public class StatsSerializer extends Serializer<Stats> implements Serializable {
 

theodolite-benchmarks/hazelcastjet-commons/src/main/java/rocks/theodolite/benchmarks/commons/hazelcastjet/HazelcastJetService.java

@@ -9,9 +9,8 @@ import org.slf4j.LoggerFactory;
 import titan.ccp.common.configuration.ServiceConfigurations;
 
 /**
- * Abstract HazelcastJetService.
- * Holds common fields and logic shared for all hazelcast jet services.
- * Set common settings and initiates a hazelcast jet instance.
+ * Abstract HazelcastJetService. Holds common fields and logic shared for all hazelcast jet
+ * services. Set common settings and initiates a hazelcast jet instance.
  */
 public abstract class HazelcastJetService {
 
@@ -33,39 +32,38 @@ public abstract class HazelcastJetService {
 
 
   /**
-   * Instantiate a new abstract service.
-   * Retrieves needed fields using ServiceConfiguration and build a new jet instance.
+   * Instantiate a new abstract service. Retrieves needed fields using ServiceConfiguration and
+   * build a new jet instance.
    */
   public HazelcastJetService(final Logger logger) {
-    this.jobName = config.getProperty(ConfigurationKeys.APPLICATION_NAME).toString();
+    this.jobName = this.config.getProperty(ConfigurationKeys.APPLICATION_NAME).toString();
 
-    this.kafkaBootstrapServer = config.getProperty(
+    this.kafkaBootstrapServer = this.config.getProperty(
         ConfigurationKeys.KAFKA_BOOTSTRAP_SERVERS).toString();
-    this.schemaRegistryUrl = config.getProperty(ConfigurationKeys.SCHEMA_REGISTRY_URL).toString();
+    this.schemaRegistryUrl =
+        this.config.getProperty(ConfigurationKeys.SCHEMA_REGISTRY_URL).toString();
     this.propsBuilder =
-        new KafkaPropertiesBuilder(kafkaBootstrapServer, schemaRegistryUrl, jobName);
+        new KafkaPropertiesBuilder(this.kafkaBootstrapServer, this.schemaRegistryUrl, this.jobName);
 
-    this.kafkaInputTopic = config.getProperty(ConfigurationKeys.KAFKA_INPUT_TOPIC).toString();
+    this.kafkaInputTopic = this.config.getProperty(ConfigurationKeys.KAFKA_INPUT_TOPIC).toString();
 
     final JetInstanceBuilder jetInstance = new JetInstanceBuilder()
-        .setConfigFromEnv(logger, kafkaBootstrapServer, HZ_KUBERNETES_SERVICE_DNS_KEY);
+        .setConfigFromEnv(logger, this.kafkaBootstrapServer, HZ_KUBERNETES_SERVICE_DNS_KEY);
     this.jetInstance = jetInstance.build();
   }
 
 
   /**
-   * Constructs and starts the pipeline.
-   * First initiates a pipeline,
-   * Second register the corresponding serializers,
-   * Third set the job name,
-   * Lastly, add the job to the hazelcast instance.
+   * Constructs and starts the pipeline. First, initiates a pipeline. Second, register the
+   * corresponding serializers. Third, set the job name. Lastly, add the job to the Hazelcast
+   * instance.
    */
   public void run() {
     try {
-      final Pipeline pipeline  = pipelineFactory.buildPipeline();
-      registerSerializer();
-      jobConfig.setName(config.getString("name"));
-      this.jetInstance.newJobIfAbsent(pipeline, jobConfig).join();
+      final Pipeline pipeline = this.pipelineFactory.buildPipeline();
+      this.registerSerializer();
+      this.jobConfig.setName(this.config.getString("name"));
+      this.jetInstance.newJobIfAbsent(pipeline, this.jobConfig).join();
     } catch (final Exception e) { // NOPMD
       LOGGER.error("ABORT MISSION!:", e);
     }

theodolite-benchmarks/hazelcastjet-commons/src/main/java/rocks/theodolite/benchmarks/commons/hazelcastjet/JetInstanceBuilder.java

@@ -42,17 +42,18 @@ public class JetInstanceBuilder {
   }
 
   /**
-   * Builds and returns a JetInstance. If a config is set, the JetInstance will contain the set
-   * config.
+   * Builds and returns a {@link JetInstance}. If a config is set, the {@link JetInstance} will
+   * contain the set config.
    *
    * @return JetInstance
    */
   public JetInstance build() {
     final JetInstance jet = Jet.newJetInstance();
-    if (this.config == null) {
+    final Config localConfig = this.config;
+    if (localConfig == null) {
       return jet;
     } else {
-      jet.getConfig().setHazelcastConfig(this.config);
+      jet.getConfig().setHazelcastConfig(localConfig);
       return jet;
     }