Clean up and fix output

uc3-application/src/main/java/uc3/application/HistoryService.java

@@ -9,8 +9,8 @@ import titan.ccp.common.configuration.Configurations;
 import uc3.streamprocessing.KafkaStreamsBuilder;
 
 /**
- * A microservice that manages the history and, therefore, stores and aggregates
- * incoming measurements.
+ * A microservice that manages the history and, therefore, stores and aggregates incoming
+ * measurements.
  *
  */
 public class HistoryService {
@@ -18,14 +18,11 @@ public class HistoryService {
   private final Configuration config = Configurations.create();
 
   private final CompletableFuture<Void> stopEvent = new CompletableFuture<>();
-	final int KAFKA_WINDOW_DURATION_MINUTES = Integer
+  private final int windowDurationMinutes = Integer
       .parseInt(Objects.requireNonNullElse(System.getenv("KAFKA_WINDOW_DURATION_MINUTES"), "60"));
 
   /**
    * Start the service.
-	 *
-	 * @return {@link CompletableFuture} which is completed when the service is
-	 *         successfully started.
    */
   public void run() {
     this.createKafkaStreamsApplication();
@@ -40,10 +37,11 @@ public class HistoryService {
         .bootstrapServers(this.config.getString(ConfigurationKeys.KAFKA_BOOTSTRAP_SERVERS))
         .inputTopic(this.config.getString(ConfigurationKeys.KAFKA_INPUT_TOPIC))
         .outputTopic(this.config.getString(ConfigurationKeys.KAFKA_OUTPUT_TOPIC))
-				.windowDuration(Duration.ofMinutes(this.KAFKA_WINDOW_DURATION_MINUTES))
+        .windowDuration(Duration.ofMinutes(this.windowDurationMinutes))
         .numThreads(this.config.getInt(ConfigurationKeys.NUM_THREADS))
         .commitIntervalMs(this.config.getInt(ConfigurationKeys.COMMIT_INTERVAL_MS))
-				.cacheMaxBytesBuffering(this.config.getInt(ConfigurationKeys.CACHE_MAX_BYTES_BUFFERING)).build();
+        .cacheMaxBytesBuffering(this.config.getInt(ConfigurationKeys.CACHE_MAX_BYTES_BUFFERING))
+        .build();
     this.stopEvent.thenRun(kafkaStreams::close);
     kafkaStreams.start();
   }

uc3-application/src/main/java/uc3/streamprocessing/TopologyBuilder.java

@@ -48,7 +48,8 @@ public class TopologyBuilder {
         .stream(this.inputTopic,
             Consumed.with(Serdes.String(),
                 IMonitoringRecordSerde.serde(new ActivePowerRecordFactory())))
-        .groupByKey().windowedBy(TimeWindows.of(this.duration))
+        .groupByKey()
+        .windowedBy(TimeWindows.of(this.duration))
         // .aggregate(
         // () -> 0.0,
         // (key, activePowerRecord, agg) -> agg + activePowerRecord.getValueInW(),
@@ -61,7 +62,7 @@ public class TopologyBuilder {
                 GenericSerde.from(Stats::toByteArray, Stats::fromByteArray)))
         .toStream()
         .map((k, s) -> KeyValue.pair(k.key(), s.toString()))
-        .peek((k, v) -> System.out.printf("key %s, value %f \n", k, v))
+        .peek((k, v) -> System.out.println(k + ": " + v))
         .to(this.outputTopic, Produced.with(Serdes.String(), Serdes.String()));
 
     return this.builder.build();