Opprettet felles modul for failhåndtering

lib/error-handling/build.gradle.kts

@@ -0,0 +1,16 @@
+plugins {
+    kotlin("jvm")
+}
+
+dependencies {
+    compileOnly(ktorServer.core)
+    compileOnly(orgApacheKafka.kafkaStreams)
+    compileOnly(loggingLibs.logbackClassic)
+
+    // Test
+    testImplementation(testLibs.bundles.withUnitTesting)
+}
+
+tasks.withType<Test>().configureEach {
+    useJUnitPlatform()
+}

lib/error-handling/src/main/kotlin/no/nav/paw/error/exception/ClientResponseException.kt

@@ -0,0 +1,10 @@
+package no.nav.paw.error.exception
+
+import io.ktor.http.HttpStatusCode
+
+open class ClientResponseException(
+    val status: HttpStatusCode,
+    override val code: String,
+    override val message: String,
+    override val cause: Throwable?
+) : ErrorCodeAwareException(code, message, cause)

lib/error-handling/src/main/kotlin/no/nav/paw/error/exception/ErrorCodeAwareException.kt

@@ -0,0 +1,7 @@
+package no.nav.paw.error.exception
+
+open class ErrorCodeAwareException(open val code: String, override val message: String, override val cause: Throwable?) :
+    Exception(message, cause) {
+
+    constructor(code: String, message: String) : this(code, message, null)
+}

lib/error-handling/src/main/kotlin/no/nav/paw/error/exception/ServerResponseException.kt

@@ -0,0 +1,10 @@
+package no.nav.paw.error.exception
+
+import io.ktor.http.HttpStatusCode
+
+open class ServerResponseException(
+    val status: HttpStatusCode,
+    override val code: String,
+    override val message: String,
+    override val cause: Throwable?
+) : ErrorCodeAwareException(code, message, cause)

lib/error-handling/src/main/kotlin/no/nav/paw/error/handler/HttpExceptionHandler.kt

@@ -0,0 +1,85 @@
+package no.nav.paw.error.handler
+
+import io.ktor.server.application.ApplicationCall
+import io.ktor.server.plugins.BadRequestException
+import io.ktor.server.plugins.ContentTransformationException
+import io.ktor.server.request.RequestAlreadyConsumedException
+import io.ktor.server.request.uri
+import io.ktor.server.response.respond
+import no.nav.paw.error.exception.ClientResponseException
+import no.nav.paw.error.exception.ServerResponseException
+import no.nav.paw.error.model.build400Error
+import no.nav.paw.error.model.build500Error
+import no.nav.paw.error.model.buildError
+import org.slf4j.Logger
+import org.slf4j.LoggerFactory
+
+private const val ERROR_TYPE_PREFIX = "PAW_"
+private val logger: Logger = LoggerFactory.getLogger("paw.application.error.http")
+
+suspend fun <T : Throwable> ApplicationCall.handleException(throwable: T) {
+    when (throwable) {
+        is ContentTransformationException -> {
+            val error = build400Error(
+                "${ERROR_TYPE_PREFIX}KUNNE_IKKE_TOLKE_INNHOLD",
+                "Kunne ikke tolke innhold i kall",
+                this.request.uri
+            )
+            logger.debug(error.detail, throwable)
+            this.respond(error.status, error)
+        }
+
+        is ClientResponseException -> {
+            val error = buildError(
+                "${ERROR_TYPE_PREFIX}${throwable.code}",
+                throwable.message,
+                throwable.status,
+                this.request.uri
+            )
+            logger.warn(error.detail, throwable)
+            this.respond(error.status, error)
+        }
+
+        is ServerResponseException -> {
+            val error = buildError(
+                "${ERROR_TYPE_PREFIX}${throwable.code}",
+                throwable.message,
+                throwable.status,
+                this.request.uri
+            )
+            logger.error(error.detail, throwable)
+            this.respond(error.status, error)
+        }
+
+        is BadRequestException -> {
+            val error =
+                build400Error(
+                    "${ERROR_TYPE_PREFIX}ULOVLIG_FORESPOERSEL",
+                    "Kunne ikke tolke innhold i forespørsel",
+                    this.request.uri
+                )
+            logger.error(error.detail, throwable)
+            this.respond(error.status, error)
+        }
+
+        is RequestAlreadyConsumedException -> {
+            val error = build500Error(
+                "${ERROR_TYPE_PREFIX}FORESPOERSEL_ALLEREDE_MOTTATT",
+                "Forespørsel er allerede mottatt. Dette er en kodefeil",
+                this.request.uri
+            )
+            logger.error(error.detail, throwable)
+            this.respond(error.status, error)
+        }
+
+        else -> {
+            val error = build500Error(
+                "${ERROR_TYPE_PREFIX}UKJENT_FEIL",
+                "Forespørsel feilet med ukjent feil",
+                this.request.uri
+            )
+            logger.error(error.detail, throwable)
+            this.respond(error.status, error)
+        }
+    }
+}

lib/error-handling/src/main/kotlin/no/nav/paw/error/handler/KafkaExceptionHandler.kt

@@ -0,0 +1,13 @@
+package no.nav.paw.error.handler
+
+import org.apache.kafka.streams.KafkaStreams
+import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler
+import org.slf4j.Logger
+import org.slf4j.LoggerFactory
+
+private val logger: Logger = LoggerFactory.getLogger("paw.application.error.kafka")
+
+fun KafkaStreams.withApplicationTerminatingExceptionHandler() = StreamsUncaughtExceptionHandler { throwable ->
+    logger.error("Kafka Streams opplevde en uventet feil", throwable)
+    StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_APPLICATION
+}

lib/error-handling/src/main/kotlin/no/nav/paw/error/model/ProblemDetails.kt

@@ -0,0 +1,39 @@
+package no.nav.paw.error.model
+
+import io.ktor.http.HttpStatusCode
+
+/**
+ * Object som inneholder detaljer om en oppstått feilsituasjon, basert på RFC 7807.
+ * @see <a href="https://datatracker.ietf.org/doc/html/rfc7807">IETF RFC 7807</a>
+ */
+data class ProblemDetails(
+    val type: String,
+    val title: String,
+    val status: HttpStatusCode,
+    val detail: String,
+    val instance: String
+) {
+    constructor(
+        title: String,
+        status: HttpStatusCode,
+        detail: String,
+        instance: String
+    ) : this("about:blank", title, status, detail, instance)
+}
+
+fun build400Error(type: String, detail: String, instance: String) =
+    buildError(type, detail, HttpStatusCode.BadRequest, instance)
+
+fun build403Error(type: String, detail: String, instance: String) =
+    buildError(type, detail, HttpStatusCode.Forbidden, instance)
+
+fun build500Error(type: String, detail: String, instance: String) =
+    buildError(type, detail, HttpStatusCode.InternalServerError, instance)
+
+fun buildError(type: String, detail: String, status: HttpStatusCode, instance: String) = ProblemDetails(
+    type = type,
+    title = status.description,
+    status = status,
+    detail = detail,
+    instance = instance
+)

lib/error-handling/src/main/kotlin/no/nav/paw/health/listener/KafkaHealthIndicatorListener.kt

@@ -0,0 +1,43 @@
+package no.nav.paw.health.listener
+
+import no.nav.paw.health.model.HealthIndicator
+import org.apache.kafka.streams.KafkaStreams
+import org.slf4j.LoggerFactory
+
+private val logger = LoggerFactory.getLogger("paw.application.health.kafka")
+
+fun KafkaStreams.withHealthIndicatorStateListener(
+    livenessHealthIndicator: HealthIndicator,
+    readinessHealthIndicator: HealthIndicator
+) = KafkaStreams.StateListener { newState, previousState ->
+    when (newState) {
+        KafkaStreams.State.RUNNING -> {
+            readinessHealthIndicator.setHealthy()
+        }
+
+        KafkaStreams.State.REBALANCING -> {
+            readinessHealthIndicator.setHealthy()
+        }
+
+        KafkaStreams.State.PENDING_ERROR -> {
+            readinessHealthIndicator.setUnhealthy()
+        }
+
+        KafkaStreams.State.PENDING_SHUTDOWN -> {
+            readinessHealthIndicator.setUnhealthy()
+        }
+
+        KafkaStreams.State.ERROR -> {
+            readinessHealthIndicator.setUnhealthy()
+            livenessHealthIndicator.setUnhealthy()
+        }
+
+        else -> {
+            readinessHealthIndicator.setUnknown()
+        }
+    }
+
+    logger.debug("Kafka Streams state endret seg ${previousState.name} -> ${newState.name}")
+    logger.info("Kafka Streams liveness er ${livenessHealthIndicator.getStatus().value}")
+    logger.info("Kafka Streams readiness er ${readinessHealthIndicator.getStatus().value}")
+}

lib/error-handling/src/main/kotlin/no/nav/paw/health/model/Health.kt

@@ -0,0 +1,37 @@
+package no.nav.paw.health.model
+
+import java.util.concurrent.atomic.AtomicReference
+
+enum class HealthStatus(val value: String) {
+    UNKNOWN("UNKNOWN"),
+    HEALTHY("HEALTHY"),
+    UNHEALTHY("UNHEALTHY"),
+}
+
+interface HealthIndicator {
+    fun setUnknown()
+    fun setHealthy()
+    fun setUnhealthy()
+    fun getStatus(): HealthStatus
+}
+
+class StandardHealthIndicator(initialStatus: HealthStatus) : HealthIndicator {
+
+    private val status = AtomicReference(initialStatus)
+
+    override fun setUnknown() {
+        status.set(HealthStatus.UNKNOWN)
+    }
+
+    override fun setHealthy() {
+        status.set(HealthStatus.HEALTHY)
+    }
+
+    override fun setUnhealthy() {
+        status.set(HealthStatus.UNHEALTHY)
+    }
+
+    override fun getStatus(): HealthStatus {
+        return status.get()
+    }
+}

lib/error-handling/src/main/kotlin/no/nav/paw/health/route/HealthRoutes.kt

@@ -0,0 +1,43 @@
+package no.nav.paw.health.route
+
+import io.ktor.http.ContentType
+import io.ktor.http.HttpStatusCode
+import io.ktor.server.application.call
+import io.ktor.server.response.respondText
+import io.ktor.server.routing.Route
+import io.ktor.server.routing.get
+import no.nav.paw.health.model.HealthStatus
+import no.nav.paw.health.service.HealthIndicatorService
+
+fun Route.healthRoutes(
+    healthIndicatorService: HealthIndicatorService,
+) {
+
+    get("/internal/isAlive") {
+        when (val status = healthIndicatorService.getLivenessStatus()) {
+            HealthStatus.HEALTHY -> call.respondText(
+                ContentType.Text.Plain,
+                HttpStatusCode.OK
+            ) { status.value }
+
+            else -> call.respondText(
+                ContentType.Text.Plain,
+                HttpStatusCode.ServiceUnavailable
+            ) { status.value }
+        }
+    }
+
+    get("/internal/isReady") {
+        when (val status = healthIndicatorService.getReadinessStatus()) {
+            HealthStatus.HEALTHY -> call.respondText(
+                ContentType.Text.Plain,
+                HttpStatusCode.OK
+            ) { status.value }
+
+            else -> call.respondText(
+                ContentType.Text.Plain,
+                HttpStatusCode.ServiceUnavailable
+            ) { status.value }
+        }
+    }
+}

lib/error-handling/src/main/kotlin/no/nav/paw/health/service/HealthIndicatorService.kt

@@ -0,0 +1,43 @@
+package no.nav.paw.health.service
+
+import no.nav.paw.health.model.HealthIndicator
+import no.nav.paw.health.model.HealthStatus
+import no.nav.paw.health.model.StandardHealthIndicator
+
+class HealthIndicatorService {
+
+    private val readinessIndicators = mutableListOf<HealthIndicator>()
+    private val livenessIndicators = mutableListOf<HealthIndicator>()
+
+    fun addReadinessIndicator(): HealthIndicator {
+        val healthIndicator = StandardHealthIndicator(HealthStatus.UNKNOWN)
+        readinessIndicators.add(healthIndicator)
+        return healthIndicator
+    }
+
+    fun addLivenessIndicator(): HealthIndicator {
+        val healthIndicator = StandardHealthIndicator(HealthStatus.HEALTHY)
+        livenessIndicators.add(healthIndicator)
+        return healthIndicator
+    }
+
+    fun getReadinessStatus(): HealthStatus {
+        return if (readinessIndicators.all { it.getStatus() == HealthStatus.HEALTHY }) {
+            HealthStatus.HEALTHY
+        } else if (readinessIndicators.any { it.getStatus() == HealthStatus.UNHEALTHY }) {
+            HealthStatus.UNHEALTHY
+        } else {
+            HealthStatus.UNKNOWN
+        }
+    }
+
+    fun getLivenessStatus(): HealthStatus {
+        return if (livenessIndicators.all { it.getStatus() == HealthStatus.HEALTHY }) {
+            HealthStatus.HEALTHY
+        } else if (livenessIndicators.any { it.getStatus() == HealthStatus.UNHEALTHY }) {
+            HealthStatus.UNHEALTHY
+        } else {
+            HealthStatus.UNKNOWN
+        }
+    }
+}

lib/kafka-streams/build.gradle.kts

@@ -2,17 +2,14 @@ plugins {
     kotlin("jvm")
 }
 
-val koTestVersion = "5.7.2"
-
 dependencies {
     api(project(":lib:kafka"))
-    implementation("org.apache.kafka:kafka-clients:3.6.0")
-    implementation("org.apache.kafka:kafka-streams:3.6.0")
-    implementation("io.confluent:kafka-streams-avro-serde:7.4.0")
+    implementation(orgApacheKafka.kafkaClients)
+    implementation(orgApacheKafka.kafkaStreams)
+    implementation(apacheAvro.kafkaStreamsAvroSerde)
 
     // Test
-    testImplementation("io.kotest:kotest-runner-junit5:$koTestVersion")
-    testImplementation("io.kotest:kotest-assertions-core:$koTestVersion")
+    testImplementation(testLibs.bundles.withUnitTesting)
 }
 
 tasks.withType<Test>().configureEach {