Dev/feilkoe headers

.nais/ebms-async-dev.yaml

@@ -82,6 +82,8 @@ spec:
       value: "true"
     - name: EBMS_PAYLOAD_PRODUCER
       value: "true"
+    - name: EBMS_RETRY_QUEUE
+      value: "true"
     - name: VAULT_JDBC_URL
       value: jdbc:postgresql://b27dbvl033.preprod.local:5432/
     - name: CPA_REPO_URL

.nais/kafka/kafka-dev.yaml

@@ -57,3 +57,31 @@ spec:
     - access: read
       application: emottak-kafka-manager
       team: team-emottak
+
+---
+
+apiVersion: kafka.nais.io/v1
+kind: Topic
+metadata:
+  labels:
+    team: team-emottak
+  name: ebxml.retry
+  namespace: team-emottak
+spec:
+  pool: nav-dev
+  config:
+    cleanupPolicy: compact
+    maxMessageBytes: 1048588
+    minimumInSyncReplicas: 2
+    partitions: 1
+    replication: 3
+    retentionBytes: -1
+    retentionHours: 168
+    segmentHours: 168
+  acl:
+    - access: readwrite
+      application: ebms-async
+      team: team-emottak
+    - access: read
+      application: emottak-kafka-manager
+      team: team-emottak

ebms-async/src/main/kotlin/no/nav/emottak/ebms/async/App.kt

@@ -7,9 +7,14 @@ import arrow.continuations.SuspendApp
 import arrow.continuations.ktor.server
 import arrow.core.raise.result
 import arrow.fx.coroutines.resourceScope
+import io.ktor.http.HttpStatusCode
 import io.ktor.server.application.Application
 import io.ktor.server.auth.authenticate
 import io.ktor.server.netty.Netty
+import io.ktor.server.response.respondText
+import io.ktor.server.routing.Route
+import io.ktor.server.routing.Routing
+import io.ktor.server.routing.get
 import io.ktor.server.routing.routing
 import io.ktor.utils.io.CancellationException
 import io.micrometer.prometheusmetrics.PrometheusConfig
@@ -28,6 +33,8 @@ import no.nav.emottak.ebms.SendInClient
 import no.nav.emottak.ebms.SmtpTransportClient
 import no.nav.emottak.ebms.async.configuration.Config
 import no.nav.emottak.ebms.async.configuration.config
+import no.nav.emottak.ebms.async.kafka.consumer.failedMessageQueue
+import no.nav.emottak.ebms.async.kafka.consumer.getRecord
 import no.nav.emottak.ebms.async.kafka.consumer.startPayloadReceiver
 import no.nav.emottak.ebms.async.kafka.consumer.startSignalReceiver
 import no.nav.emottak.ebms.async.kafka.producer.EbmsMessageProducer
@@ -48,7 +55,9 @@ import no.nav.emottak.ebms.registerRootEndpoint
 import no.nav.emottak.ebms.scopedAuthHttpClient
 import no.nav.emottak.ebms.sendin.SendInService
 import no.nav.emottak.ebms.validation.DokumentValidator
+import no.nav.emottak.message.util.isProdEnv
 import org.slf4j.LoggerFactory
+import kotlin.uuid.ExperimentalUuidApi
 
 val log = LoggerFactory.getLogger("no.nav.emottak.ebms.async.App")
 
@@ -81,6 +90,15 @@ fun main() = SuspendApp {
         ebmsPayloadProducer = ebmsPayloadProducer
     )
 
+    val payloadMessageProcessorProvider = payloadMessageProcessorProvider(
+        ebmsMessageDetailsRepository = ebmsMessageDetailsRepository,
+        dokumentValidator = dokumentValidator,
+        processingService = processingService,
+        ebmsSignalProducer = ebmsSignalProducer,
+        smtpTransportClient = smtpTransportClient,
+        payloadMessageResponder = payloadMessageResponder
+    )
+
     result {
         resourceScope {
             launchSignalReceiver(
@@ -90,20 +108,16 @@ fun main() = SuspendApp {
             )
             launchPayloadReceiver(
                 config = config,
-                dokumentValidator = dokumentValidator,
-                processingService = processingService,
-                ebmsSignalProducer = ebmsSignalProducer,
-                smtpTransportClient = smtpTransportClient,
-                payloadMessageResponder = payloadMessageResponder,
-                ebmsMessageDetailsRepository = ebmsMessageDetailsRepository
+                payloadMessageProcessorProvider = payloadMessageProcessorProvider
             )
 
             server(
                 Netty,
                 port = 8080,
                 module = {
                     ebmsProviderModule(
-                        payloadRepository = payloadRepository
+                        payloadRepository = payloadRepository,
+                        payloadProcessorProvider = payloadMessageProcessorProvider
                     )
                 }
             ).also { it.engineConfig.maxChunkSize = 100000 }
@@ -119,26 +133,36 @@ fun main() = SuspendApp {
         }
 }
 
-private fun CoroutineScope.launchPayloadReceiver(
-    config: Config,
+fun payloadMessageProcessorProvider(
+    ebmsMessageDetailsRepository: EbmsMessageDetailsRepository,
     dokumentValidator: DokumentValidator,
     processingService: ProcessingService,
     ebmsSignalProducer: EbmsMessageProducer,
     smtpTransportClient: SmtpTransportClient,
-    payloadMessageResponder: PayloadMessageResponder,
-    ebmsMessageDetailsRepository: EbmsMessageDetailsRepository
+    payloadMessageResponder: PayloadMessageResponder
+
+): () -> PayloadMessageProcessor = {
+    PayloadMessageProcessor(
+        ebmsMessageDetailsRepository = ebmsMessageDetailsRepository,
+        validator = dokumentValidator,
+        processingService = processingService,
+        ebmsSignalProducer = ebmsSignalProducer,
+        smtpTransportClient = smtpTransportClient,
+        payloadMessageResponder
+    )
+}
+
+private fun CoroutineScope.launchPayloadReceiver(
+    config: Config,
+    payloadMessageProcessorProvider: () -> PayloadMessageProcessor
 ) {
     if (config.kafkaPayloadReceiver.active) {
         launch(Dispatchers.IO) {
-            val payloadMessageProcessor = PayloadMessageProcessor(
-                validator = dokumentValidator,
-                processingService = processingService,
-                ebmsSignalProducer = ebmsSignalProducer,
-                smtpTransportClient = smtpTransportClient,
-                ebmsMessageDetailsRepository = ebmsMessageDetailsRepository,
-                payloadMessageResponder = payloadMessageResponder
+            startPayloadReceiver(
+                config.kafkaPayloadReceiver.topic,
+                config.kafka,
+                payloadMessageProcessorProvider.invoke()
             )
-            startPayloadReceiver(config.kafkaPayloadReceiver.topic, config.kafka, payloadMessageProcessor)
         }
     }
 }
@@ -160,7 +184,8 @@ private fun CoroutineScope.launchSignalReceiver(
 }
 
 fun Application.ebmsProviderModule(
-    payloadRepository: PayloadRepository
+    payloadRepository: PayloadRepository,
+    payloadProcessorProvider: () -> PayloadMessageProcessor
 ) {
     val appMicrometerRegistry = PrometheusMeterRegistry(PrometheusConfig.DEFAULT)
 
@@ -173,9 +198,53 @@ fun Application.ebmsProviderModule(
         registerHealthEndpoints()
         registerPrometheusEndpoint(appMicrometerRegistry)
         registerNavCheckStatus()
-
+        if (!isProdEnv()) {
+            simulateError()
+        }
+        retryErrors(payloadProcessorProvider)
         authenticate(AZURE_AD_AUTH) {
             getPayloads(payloadRepository)
         }
     }
 }
+
+const val RETRY_LIMIT = "retryLimit"
+
+@OptIn(ExperimentalUuidApi::class)
+fun Routing.retryErrors(
+    payloadMessageProcessorProvider: () -> PayloadMessageProcessor
+): Route =
+    get("/api/retry/{$RETRY_LIMIT}") {
+        if (!config().kafkaErrorQueue.active) {
+            call.respondText(status = HttpStatusCode.ServiceUnavailable, text = "Retry not active.")
+            return@get
+        }
+        failedMessageQueue.consumeRetryQueue(
+            payloadMessageProcessorProvider.invoke(),
+            limit = (call.parameters[RETRY_LIMIT])?.toInt() ?: 10
+        )
+        call.respondText(
+            status = HttpStatusCode.OK,
+            text = "Retry processing started with limit ${call.parameters[RETRY_LIMIT] ?: "default"}"
+        )
+    }
+
+const val KAFKA_OFFSET = "offset"
+
+fun Route.simulateError(): Route = get("/api/forceretry/{$KAFKA_OFFSET}") {
+    CoroutineScope(Dispatchers.IO).launch() {
+        if (config().kafkaErrorQueue.active) {
+            val record = getRecord(
+                config()
+                    .kafkaPayloadReceiver.topic,
+                config().kafka
+                    .copy(groupId = "ebms-provider-retry"),
+                (call.parameters[KAFKA_OFFSET])?.toLong() ?: 0
+            )
+            failedMessageQueue.sendToRetry(
+                record = record ?: throw Exception("No Record found. Offset: ${call.parameters[KAFKA_OFFSET]}"),
+                reason = "Simulated Error"
+            )
+        }
+    }
+}

ebms-async/src/main/kotlin/no/nav/emottak/ebms/async/configuration/Config.kt

@@ -1,6 +1,7 @@
 package no.nav.emottak.ebms.async.configuration
 
 import com.sksamuel.hoplite.Masked
+import org.apache.kafka.clients.CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG
 import org.apache.kafka.clients.CommonClientConfigs.SECURITY_PROTOCOL_CONFIG
 import org.apache.kafka.common.config.SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG
 import org.apache.kafka.common.config.SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG
@@ -15,7 +16,8 @@ data class Config(
     val kafkaSignalReceiver: KafkaSignalReceiver,
     val kafkaSignalProducer: KafkaSignalProducer,
     val kafkaPayloadReceiver: KafkaPayloadReceiver,
-    val kafkaPayloadProducer: KafkaPayloadProducer
+    val kafkaPayloadProducer: KafkaPayloadProducer,
+    val kafkaErrorQueue: KafkaErrorQueue
 )
 
 @JvmInline
@@ -53,6 +55,11 @@ data class KafkaPayloadProducer(
     val topic: String
 )
 
+data class KafkaErrorQueue(
+    val active: Boolean,
+    val topic: String
+)
+
 data class Kafka(
     val bootstrapServers: String,
     val securityProtocol: SecurityProtocol,
@@ -62,11 +69,22 @@ data class Kafka(
     val truststoreType: TruststoreType,
     val truststoreLocation: TruststoreLocation,
     val truststorePassword: Masked,
-    val groupId: String
+    val groupId: String,
+    val properties: Properties = Properties().apply {
+        put(BOOTSTRAP_SERVERS_CONFIG, bootstrapServers)
+        put(SECURITY_PROTOCOL_CONFIG, securityProtocol.value)
+        put(SSL_KEYSTORE_TYPE_CONFIG, keystoreType.value)
+        put(SSL_KEYSTORE_LOCATION_CONFIG, keystoreLocation.value)
+        put(SSL_KEYSTORE_PASSWORD_CONFIG, keystorePassword.value)
+        put(SSL_TRUSTSTORE_TYPE_CONFIG, truststoreType.value)
+        put(SSL_TRUSTSTORE_LOCATION_CONFIG, truststoreLocation.value)
+        put(SSL_TRUSTSTORE_PASSWORD_CONFIG, truststorePassword.value)
+    }
 )
 
 fun Kafka.toProperties() = Properties()
     .apply {
+        put(BOOTSTRAP_SERVERS_CONFIG, bootstrapServers)
         put(SECURITY_PROTOCOL_CONFIG, securityProtocol.value)
         put(SSL_KEYSTORE_TYPE_CONFIG, keystoreType.value)
         put(SSL_KEYSTORE_LOCATION_CONFIG, keystoreLocation.value)

ebms-async/src/main/kotlin/no/nav/emottak/ebms/async/kafka/consumer/EbmsPayloadReceiver.kt

@@ -29,7 +29,7 @@ suspend fun startPayloadReceiver(topic: String, kafka: Kafka, payloadMessageProc
     KafkaReceiver(receiverSettings)
         .receive(topic)
         .map { record ->
-            payloadMessageProcessor.process(record.key(), record.value())
+            payloadMessageProcessor.process(record)
             record.offset.acknowledge()
         }.collect()
 }