Manager - Kafka Consumer for Observations (#97)

* Manager - Hilltop Crawler Changes

* Shuffle files around

* martin/aggregate water use (#95)

* add water use daily aggregation sql view

* optimise sql query performance

* Adding Error handling to Observations Consumer

---------

Co-authored-by: Martin Peak <[email protected]>

packages/Manager/Dockerfile

@@ -9,7 +9,7 @@ WORKDIR application
 COPY --from=builder application/dependencies/ ./
 COPY --from=builder application/spring-boot-loader/ ./
 COPY --from=builder application/application/ ./
-ENTRYPOINT ["java", "-XX:MaxRAMPercentage=85", "org.springframework.boot.loader.launch.JarLauncher"]
+ENTRYPOINT ["java", "-XX:MaxRAMPercentage=80", "org.springframework.boot.loader.launch.JarLauncher"]
 
 EXPOSE 8080
 HEALTHCHECK --interval=5s --retries=5 --start-period=60s CMD curl --fail http://localhost:8080/actuator/health || exit 1

packages/Manager/build.gradle.kts

@@ -40,6 +40,7 @@ dependencies {
   implementation("org.springframework.boot:spring-boot-starter-jdbc")
   implementation("org.springframework.boot:spring-boot-starter-jooq")
   implementation("org.springframework.kafka:spring-kafka")
+  implementation("org.apache.kafka:kafka-streams")
   implementation("io.micrometer:micrometer-tracing-bridge-brave")
   implementation("org.jetbrains.kotlin:kotlin-reflect")
   implementation("org.jetbrains.kotlin:kotlin-stdlib-jdk8")

packages/Manager/src/main/kotlin/nz/govt/eop/Application.kt

@@ -15,7 +15,7 @@ import org.springframework.web.client.RestTemplate
 @EnableKafka
 @EnableCaching
 @SpringBootApplication
-class Application() {
+class Application {
   @Bean fun restTemplate(): RestTemplate = RestTemplateBuilder().build()
 }
 

packages/Manager/src/main/kotlin/nz/govt/eop/consumers/observations/ObservationMessages.kt

@@ -0,0 +1,75 @@
+package nz.govt.eop.consumers.observations
+
+import com.fasterxml.jackson.annotation.JsonSubTypes
+import com.fasterxml.jackson.annotation.JsonTypeInfo
+import java.math.BigDecimal
+import java.time.OffsetDateTime
+import java.time.YearMonth
+
+enum class ObservationMessageType {
+  SITE_DETAILS,
+  OBSERVATION_DATA,
+}
+
+@JsonTypeInfo(
+    use = JsonTypeInfo.Id.NAME, include = JsonTypeInfo.As.EXISTING_PROPERTY, property = "type")
+@JsonSubTypes(
+    JsonSubTypes.Type(value = SiteMessageKey::class, name = "SITE_DETAILS"),
+    JsonSubTypes.Type(value = ObservationDataMessageKey::class, name = "OBSERVATION_DATA"),
+)
+abstract class ObservationMessageKey(val type: ObservationMessageType) {
+  abstract val councilId: Int
+  abstract val siteName: String
+}
+
+data class SiteMessageKey(
+    override val councilId: Int,
+    override val siteName: String,
+) : ObservationMessageKey(ObservationMessageType.SITE_DETAILS)
+
+data class ObservationDataMessageKey(
+    override val councilId: Int,
+    override val siteName: String,
+    val measurementName: String,
+    val yearMonth: YearMonth
+) : ObservationMessageKey(ObservationMessageType.OBSERVATION_DATA)
+
+@JsonTypeInfo(
+    use = JsonTypeInfo.Id.NAME, include = JsonTypeInfo.As.EXISTING_PROPERTY, property = "type")
+@JsonSubTypes(
+    JsonSubTypes.Type(value = SiteDetailsMessage::class, name = "SITE_DETAILS"),
+    JsonSubTypes.Type(value = ObservationDataMessage::class, name = "OBSERVATION_DATA"),
+)
+abstract class ObservationMessage(val type: ObservationMessageType) {
+  abstract val councilId: Int
+  abstract val siteName: String
+
+  abstract fun toKey(): ObservationMessageKey
+}
+
+data class SiteDetailsMessage(
+    override val councilId: Int,
+    override val siteName: String,
+    val location: Location?
+) : ObservationMessage(ObservationMessageType.SITE_DETAILS) {
+  override fun toKey() =
+      SiteMessageKey(
+          councilId,
+          siteName,
+      )
+}
+
+data class Location(val easting: Int, val northing: Int)
+
+data class ObservationDataMessage(
+    override val councilId: Int,
+    override val siteName: String,
+    val measurementName: String,
+    val observations: List<Observation>
+) : ObservationMessage(ObservationMessageType.OBSERVATION_DATA) {
+  override fun toKey() =
+      ObservationDataMessageKey(
+          councilId, siteName, measurementName, YearMonth.from(observations.first().observedAt))
+}
+
+data class Observation(val observedAt: OffsetDateTime, val value: BigDecimal)