Dev/feilkoe headers #112

RettIProd · 2025-03-04T10:13:21Z

No description provided.

kfh

Noen kommentarer elles ser jo dette ut som en god start 👍

kfh · 2025-03-21T08:16:30Z

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

+): Route =
+    get("/api/retry/{$RETRY_LIMIT}") {
+        resourceScope {
+            CoroutineScope(Dispatchers.IO).launch {


Når du oppretter et CoroutineScope her på denne måten er det greit å være klar over at dette blir et helt frittstående scope som ikke er linket til resourceScope'et. Joda, det vil bli tatt ned når jvm'en shutter ned, men ikke på en clean måte.

Jeg snakket jo litt om dette for noen uker siden (i en delingstime) hvordan en kan linke frittstående scopes til omsluttende ressurs-scope så scopet i seg selv blir en ressurs / avhengighet på lik linje med alle andre ressurser og avhengigheter. Fordelen er selvsagt at scopet blir inkludert i livssyklusen til ressurs-scopet i SuspendApp og tatt ned og opprettet på en clean måte.

Her er linken til utility funksjonen som gjør dette for deg: https://github.com/navikt/smtp-transport/blob/main/src/main/kotlin/no/nav/emottak/util/ResourceUtil.kt

Sikkert noe vi burde putte inn i felles-biblioteket på et tidspunkt.

kfh · 2025-03-21T08:17:10Z

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

+
+fun Route.simulateError(): Route = get("/api/forceretry/{$KAFKA_OFFSET}") {
+    resourceScope {
+        CoroutineScope(Dispatchers.IO).launch {


Samme som over

kfh · 2025-03-21T08:18:18Z

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

+                if (config().kafkaErrorQueue.active) {
+                    failedMessageQueue.receive(
+                        payloadMessageProcessorProvider.invoke(),
+                        limit = (call.parameters[RETRY_LIMIT] as String).toInt()


Litt flisespikkeri, men syntes koden blir mer ryddig hvis du løfter call... ut til en egen val

Hvorfor synes du det?

Tipper fordi blir koden mer verbose. I prinsipp er jeg helt enig, men akkurat i dette eksemplet tenker jeg at det ikke gir så mye verdi å bruke mer tid på denne linjen

kfh · 2025-03-21T08:25:18Z

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

+    suspend fun send(record: ReceiverRecord<String, ByteArray>, key: String = record.key(), value: ByteArray = record.value()) {
+        record.addHeader(RETRY_AFTER, getNextRetryTime(record))
+        try {
+            val result = producer.send(ProducerRecord(kafkaErrorQueue.topic, null, key, value, record.headers())).get()


Jeg tenker at her bør du bruke KafkaPublisher fra kotlin-kafka-biblioteket så får du et publisherScope ut av boksen. Dette vil forenkle og gjøre koden mer tydelig.

kfh · 2025-03-21T08:26:08Z

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

+
+    fun getNextRetryTime(record: ReceiverRecord<String, ByteArray>): String {
+        return DateTime.now().plusMinutes(5)
+            .toString() // TODO create retry strategy


Ref retry strategi, kanskje du bør se litt på https://arrow-kt.io/learn/resilience/retry-and-repeat/

kfh · 2025-03-21T08:27:17Z

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

+            ByteArrayDeserializer()
+        )
+    ) {
+        partitionsFor(topic)


Det er mye som foregår her, hva med å splitte dette opp i et par mindre funksjoner for å tydeliggjøre hva som faktisk skjer

kfh · 2025-03-21T08:29:49Z

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

+    }
+}
+
+fun getReceiverRecord(consumerRecord: ConsumerRecord<String, ByteArray>?): ReceiverRecord<String, ByteArray>? {


Det blir veldig mye kode inne i denne funksjonen og det går utover lesbarheten. Kan du ikke trekke klassen ut ?

Skjønner hva du mener, om man bare er interessert i flyten er det ikke interessant eller nødvendig med støyen en plutselig klassedefinisjon skaper men jeg har en annen vinkling.
Klassen eksisterer kun i denne contexten og trenger ikke eksistere i annet scope, ved å flytte den "forsøpler" den lesbarheten til resten av koden "forøvrig". Ved å se den her vet man at den kun eksisterer her umiddelbart. Ved å se den frittstående vil man tro den kan brukes flere steder og har et "udefinert eksistensområde" i koden. Det kan være vel så kognitivt belastende for en leser.

kfh · 2025-03-21T08:30:31Z

ebms-async/src/test/kotlin/no/nav/emottak/ebms/async/kafka/KafkaIntegrationTest.kt

+import kotlin.io.path.Path
+import kotlin.io.path.exists
+
+class KafkaIntegrationTest {


Veldig tynn integrasjonstest. savner feil-scenarioer

ivanskodje · 2025-03-27T12:19:21Z

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

-                }
-            }
+        if (!config().kafkaErrorQueue.active) {
+            call.respondText(status = HttpStatusCode.ServiceUnavailable, text = "Retry not active.")


Selv om Retry i teksten her refererer til API-endepunktet, kan teksten lett misforstås. Kanskje det er bedre å tydeliggjøre at det er endepunktet som ikke er aktivt, i stedet for å bare kalle det "Retry"

Det er faktisk retry som er inaktivt, ikke endepunktet. ;)
Altså kafka-køen er ikke aktiv. Så man kan ikke få gjort noe retrys.

ivanskodje · 2025-03-27T12:21:35Z

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

+            keySerializer = StringSerializer(),
+            valueSerializer = ByteArraySerializer(),
+            acknowledgments = Acks.All,
+            properties = kafka.toProperties()
        )


Dette er en stor forbering! 👍 ⚡

ivanskodje · 2025-03-27T12:23:44Z

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

+        record: ReceiverRecord<String, ByteArray>,
+        key: String = record.key(),
+        value: ByteArray = record.value()
+    ) {
        record.addHeader(RETRY_AFTER, getNextRetryTime(record))
        try {


Nå som jeg tenker på det - hva sier du til at vi bruker Arrow her også, så vi slipper de litt kjipe try-catchene?

ivanskodje · 2025-03-27T12:25:22Z

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

+    suspend fun consumeRetryQueue( // TODO refine retry logic
+        payloadMessageProcessor: PayloadMessageProcessor,
+        limit: Int = 10 // TODO default limit to offset
+    ) {
+        // TODO DefaultKafkaReceiver is too constrainted so need own impl for custom logic
+        val consumer: Flow<ReceiverRecord<String, ByteArray>> =
+            errorTopicKafkaReceiver.receive(kafkaErrorQueue.topic)
+
        logger.debug("Reading from error queue")
        var counter = 0
-        consumerFlow.map { record ->
+        consumer.map { record ->
            counter++
            if (counter > limit) {
-                throw Exception("Error queue limit exceeded: $limit") // TODO fjern dette
+                logger.info("Kafka retryQueue Limit reached: $limit")
+                return@map
            }
            record.offset.acknowledge()
            record.retryCounter()
-            payloadMessageProcessor.process(record)
-            record.offset.acknowledge()
+            if (DateTime.parse(
+                    String(record.headers().lastHeader(RETRY_AFTER).value())
+                ).isAfter(DateTime.now())
+            ) {
+                payloadMessageProcessor.process(record)
+            } else {
+                logger.info("${record.key()} is not retryable yet.")
+                failedMessageQueue.sendToRetry(record)
+            }
+            record.offset.commit()
        }.collect()
    }


Siden vi vet at denne delen trenger mer arbeid, hva tenker du om å legge til en enkel unit test som kan hjelpe oss å oppdage hvis vi ubevisst ødelegger noe etter refaktorering? Det ville gjort det mye tryggere å rydde opp her, og samtidig gjøre terskelen til å rydde opp her lavere

ivanskodje · 2025-03-27T12:29:45Z

ebms-async/src/test/kotlin/no/nav/emottak/ebms/async/kafka/KafkaIntegrationTest.kt

@@ -26,10 +29,15 @@ class KafkaIntegrationTest {
    val kafkaConfig = config()

    fun noLocalKafkaEnv(): Boolean {


En liten forenkling du kan vurdere her: Hva med å kalle denne hasLocalKafkaEnv i stedet for noLocalKafkaEnv? Det gjør det litt enklere å lese, siden vi slipper å tenke i det negative. 😅

ivanskodje · 2025-03-27T12:30:48Z

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

-    suspend fun receive(payloadMessageProcessor: PayloadMessageProcessor, limit: Int = 10) { // TODO limit til offset
+    suspend fun consumeRetryQueue( // TODO refine retry logic
+        payloadMessageProcessor: PayloadMessageProcessor,
+        limit: Int = 10 // TODO default limit to offset


// TODO default limit to offset jeg forstår ikke denne kommentaren. Kan vi forbedre den eller ta den vekk hvis du ikke heller gjør det?

Tenkte limiten er siste offset på tidspunktet man starter processen. Men man kan ikke innhente siste offset vha kotlin-kafka sin DefaultKafkaReceiver. Det blir en større jobb isåfall som er litt tentativ. Så lagde TODO

RettIProd · 2025-03-27T14:42:39Z

Mer kan selvsagt gjøres men tenker det er nok foreløpig. Har testet det i dev vha. retry endepunktet
https://ebms-async-fss.intern.dev.nav.no/api/retry/5
Feilende meldinger vil også bli lagt på køen. De vil ikke bli prosessert automatisk ihvertfall foreløpig.

RettIProd added 19 commits March 4, 2025 10:58

Init commit

b3867e3

Resolve conflict

501c6d8

fix test setup

5e59172

Logging og active kø

0bcf452

Limit

29d6b2e

Endepunkt for å simulere ting på feilkø

f18b0bd

Klammeparantes

ba756d5

Get path params

d3af0a8

Getrecord config

0753c45

Bootstrap servers ikke satt i props

190a9d9

Debug

6617a97

Test

c050c35

merge conflict

64975e1

Integrasjonstest

db4d56c

Permissions for async

9cb3a70

Merge branch 'main' into dev/feilkoe-headers

2621350

Disable local tests

7d46ab0

Disable local tests

b37844d

Fix producer

d7f7163

RettIProd marked this pull request as ready for review March 20, 2025 15:04

RettIProd requested a review from a team as a code owner March 20, 2025 15:04

Disable test

a0a5e8f

kfh reviewed Mar 21, 2025

View reviewed changes

RettIProd added 3 commits March 26, 2025 16:44

Fix deprecated kafka setup

ba3fc33

Basic RetryQueue logic

1b80d83

Merge branch 'main' into dev/feilkoe-headers

73bf16d

ivanskodje reviewed Mar 27, 2025

View reviewed changes

ivanskodje requested a review from a team March 27, 2025 12:31

RettIProd added 3 commits March 27, 2025 13:32

Ta inn util endring

007260e

Fix retry time check

225fb97

Reason added to headers

dc2e0dd

RettIProd requested review from ivanskodje and kfh March 27, 2025 14:39

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Dev/feilkoe headers #112

Dev/feilkoe headers #112

RettIProd commented Mar 4, 2025

kfh left a comment

kfh Mar 21, 2025

kfh Mar 21, 2025

kfh Mar 21, 2025

RettIProd Mar 21, 2025

ivanskodje Mar 24, 2025

kfh Mar 21, 2025

kfh Mar 21, 2025

kfh Mar 21, 2025

kfh Mar 21, 2025

RettIProd Mar 21, 2025 •

edited

Loading

kfh Mar 21, 2025

ivanskodje Mar 27, 2025

RettIProd Mar 27, 2025 •

edited

Loading

ivanskodje Mar 27, 2025

ivanskodje Mar 27, 2025

ivanskodje Mar 27, 2025 •

edited

Loading

ivanskodje Mar 27, 2025

ivanskodje Mar 27, 2025

RettIProd Mar 27, 2025

RettIProd commented Mar 27, 2025

		@@ -26,10 +29,15 @@ class KafkaIntegrationTest {
		val kafkaConfig = config()

		fun noLocalKafkaEnv(): Boolean {

Dev/feilkoe headers #112

Are you sure you want to change the base?

Dev/feilkoe headers #112

Conversation

RettIProd commented Mar 4, 2025

kfh left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

RettIProd Mar 21, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

RettIProd Mar 27, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ivanskodje Mar 27, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

RettIProd commented Mar 27, 2025

RettIProd Mar 21, 2025 •

edited

Loading

RettIProd Mar 27, 2025 •

edited

Loading

ivanskodje Mar 27, 2025 •

edited

Loading