Test 6

Consumer/pom.xml

@@ -14,6 +14,14 @@
     <name>Consumer</name>
     <description>Consumer</description>
 
+
+    <properties>
+        <java.version>17</java.version>
+        <spring-cloud.version>2022.0.0</spring-cloud.version>
+        <maven.compiler.source>17</maven.compiler.source>
+        <maven.compiler.target>17</maven.compiler.target>
+    </properties>
+
     <dependencies>
         <dependency>
             <groupId>com.kowalczyk.konrad</groupId>

KafkaStreams/pom.xml

@@ -14,12 +14,18 @@
     <name>KafkaStreams</name>
     <description>KafkaStreams</description>
 
+    <properties>
+        <java.version>17</java.version>
+        <spring-cloud.version>2022.0.0</spring-cloud.version>
+        <maven.compiler.source>17</maven.compiler.source>
+        <maven.compiler.target>17</maven.compiler.target>
+    </properties>
+
     <dependencies>
         <dependency>
             <groupId>com.kowalczyk.konrad</groupId>
             <artifactId>Utils</artifactId>
             <version>0.0.1-SNAPSHOT</version>
-            <scope>compile</scope>
         </dependency>
         <dependency>
             <groupId>org.apache.kafka</groupId>
@@ -29,5 +35,11 @@
             <groupId>org.springframework.cloud</groupId>
             <artifactId>spring-cloud-stream-binder-kafka-streams</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+            <version>1.18.24</version>
+        </dependency>
     </dependencies>
+
 </project>

KafkaStreams/src/main/java/com/kowalczyk/konrad/kafkastreams/DataCalc.java

@@ -0,0 +1,97 @@
+package com.kowalczyk.konrad.kafkastreams;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.kowalczyk.konrad.utils.DataModel;
+import lombok.Getter;
+import lombok.Setter;
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.Collections;
+
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class DataCalc {
+
+    @Setter
+    @Getter
+    private String date;
+    @Setter
+    @Getter
+    private double value;
+    @Getter
+    @Setter
+    private String positionCode;
+    @Getter
+    @Setter
+    private String unit;
+    @Getter
+    @Setter
+    private String averagingTime;
+    @Getter
+    @Setter
+    private String indicator;
+    @Getter
+    @Setter
+    private String stationCode;
+    @Setter
+    @Getter
+    private long timestampSend;
+    @Setter
+    @Getter
+    private long timestampConsumer;
+    @Setter
+    @Getter
+    private double averageValue;
+    @Setter
+    @Getter
+    private double sum;
+    @Setter
+    @Getter
+    private long count;
+
+
+    public DataCalc(String date, double value, String positionCode, String unit, String averagingTime, String indicator
+            , String stationCode, long timestampSend, long timestampConsumer, double averageValue) {
+        this.date = date;
+        this.value = value;
+        this.positionCode = positionCode;
+        this.unit = unit;
+        this.averagingTime = averagingTime;
+        this.indicator = indicator;
+        this.stationCode = stationCode;
+        this.timestampSend = timestampSend;
+        this.timestampConsumer = timestampConsumer;
+        this.averageValue = averageValue;
+    }
+
+    public DataCalc() {
+        this.date = StringUtils.EMPTY;
+        this.value = 0.0;
+        this.positionCode = StringUtils.EMPTY;
+        this.unit = StringUtils.EMPTY;
+        this.averagingTime = StringUtils.EMPTY;
+        this.indicator = StringUtils.EMPTY;
+        this.stationCode = StringUtils.EMPTY;
+        this.timestampSend = 0;
+        this.timestampConsumer = 0;
+        this.averageValue = 0.0;
+        this.sum = 0;
+        this.count = 0;
+    }
+
+    public DataCalc(DataModel dataModel) {
+        this.date = dataModel.getDate();
+        this.value = dataModel.getValue();
+        this.positionCode = dataModel.getPositionCode();
+        this.unit = dataModel.getUnit();
+        this.averagingTime = dataModel.getAveragingTime();
+        this.indicator = dataModel.getIndicator();
+        this.stationCode = dataModel.getStationCode();
+        this.timestampSend = dataModel.getTimestampSend();
+        this.timestampConsumer = 0;
+        this.averageValue = 0.0;
+        this.sum = 0;
+        this.count = 0;
+    }
+
+}

KafkaStreams/src/main/java/com/kowalczyk/konrad/kafkastreams/DataCalcDeserializer.java

@@ -0,0 +1,20 @@
+package com.kowalczyk.konrad.kafkastreams;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.kafka.common.serialization.Deserializer;
+
+public class DataCalcDeserializer implements Deserializer<DataCalc> {
+
+    private final static ObjectMapper objectMapper = new ObjectMapper();
+
+    @Override
+    public DataCalc deserialize(String topic, byte[] data) {
+        DataCalc object = null;
+        try {
+            object = objectMapper.readValue(data, DataCalc.class);
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+        return object;
+    }
+}

KafkaStreams/src/main/java/com/kowalczyk/konrad/kafkastreams/DataCalcSerde.java

@@ -0,0 +1,24 @@
+package com.kowalczyk.konrad.kafkastreams;
+
+import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.serialization.Serializer;
+
+public class DataCalcSerde implements Serde<DataCalc> {
+
+    @Override
+    public Serializer<DataCalc> serializer() {
+        return new DataCalcSerializer();
+    }
+
+    @Override
+    public Deserializer<DataCalc> deserializer() {
+        return new DataCalcDeserializer();
+    }
+
+    @Override
+    public void close() {
+    }
+
+}
+

KafkaStreams/src/main/java/com/kowalczyk/konrad/kafkastreams/DataCalcSerializer.java

@@ -0,0 +1,20 @@
+package com.kowalczyk.konrad.kafkastreams;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.kafka.common.serialization.Serializer;
+
+public class DataCalcSerializer implements Serializer<DataCalc> {
+
+    private final static ObjectMapper objectMapper = new ObjectMapper();
+
+    @Override
+    public byte[] serialize(String s, DataCalc dataCalc) {
+        byte[] result = null;
+        try {
+            result = objectMapper.writeValueAsBytes(dataCalc);
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+        return result;
+    }
+}

KafkaStreams/src/main/java/com/kowalczyk/konrad/kafkastreams/KafkaStreamsProcess.java

@@ -1,7 +1,11 @@
 package com.kowalczyk.konrad.kafkastreams;
 
 import com.kowalczyk.konrad.utils.DataModel;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.kstream.Grouped;
 import org.apache.kafka.streams.kstream.KStream;
+import org.apache.kafka.streams.kstream.Materialized;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
@@ -10,10 +14,38 @@
 @Configuration
 public class KafkaStreamsProcess {
 
+    public static final String START_DATE = "1/1/2019 01:00";
+
     @Bean
     public Function<KStream<String, DataModel>, KStream<String, DataModel>> process() {
-        return kStream -> kStream;
-    }
+        return kStream -> kStream.filter((key, value) -> value.getValue() > 0)
+                .map((key, value) -> new KeyValue<>(value.getId(), new DataCalc(value)))
+                .groupByKey(Grouped.with(Serdes.String(), new DataCalcSerde()))
+                .aggregate(DataCalc::new,
+                        (key, value, aggregate) -> {
+                            if (START_DATE.equals(value.getDate())) {
+                                aggregate = value;
+                            }
+                            aggregate.setSum(aggregate.getSum() + value.getValue());
+                            aggregate.setCount(aggregate.getCount() + 1);
+                            aggregate.setAverageValue(aggregate.getSum() / aggregate.getCount());
+                            return aggregate;
+                        }
+                        , Materialized.with(Serdes.String(), new DataCalcSerde()))
+                .toStream()
+                .map((key, value) -> new KeyValue<>(String.valueOf(value.getPositionCode())
+                        , new DataModel(value.getDate()
+                        , value.getValue()
+                        , value.getPositionCode()
+                        , value.getUnit()
+                        , value.getAveragingTime()
+                        , value.getIndicator()
+                        , value.getStationCode()
+                        , value.getTimestampSend()
+                        , 0
+                        , value.getAverageValue()
+                        , value.getCount())));
 
+    }
 
 }

Producer/pom.xml

@@ -14,6 +14,13 @@
     <packaging>jar</packaging>
     <description>Producer</description>
 
+    <properties>
+        <java.version>17</java.version>
+        <spring-cloud.version>2022.0.0</spring-cloud.version>
+        <maven.compiler.source>17</maven.compiler.source>
+        <maven.compiler.target>17</maven.compiler.target>
+    </properties>
+
     <dependencies>
         <dependency>
             <groupId>com.kowalczyk.konrad</groupId>

Producer/src/main/java/com/kowalczyk/konrad/producer/Producer.java

@@ -4,11 +4,9 @@
 import com.kowalczyk.konrad.utils.IoTSimulation;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
-
 import reactor.core.publisher.Flux;
 
 import java.time.Duration;
-import java.time.Instant;
 import java.util.function.Function;
 import java.util.function.Supplier;