Send Kafka-key til behov-rivers som aktiveres i parallell (#797)

* Send Kafka-key til behov-rivers som aktiveres i parallell

* Fiks signaturkonflikt i Java

* Få med nytt meldingsfelt i testene

* Oppdater integrasjonstest

* Sjekk at svarkey blir sendt i tester

aareg/src/main/kotlin/no/nav/helsearbeidsgiver/inntektsmelding/aareg/HentArbeidsforholdRiver.kt

@@ -8,9 +8,11 @@ import no.nav.helsearbeidsgiver.felles.Key
 import no.nav.helsearbeidsgiver.felles.domene.Arbeidsforhold
 import no.nav.helsearbeidsgiver.felles.json.krev
 import no.nav.helsearbeidsgiver.felles.json.les
+import no.nav.helsearbeidsgiver.felles.json.lesOrNull
 import no.nav.helsearbeidsgiver.felles.json.toJson
 import no.nav.helsearbeidsgiver.felles.json.toMap
 import no.nav.helsearbeidsgiver.felles.metrics.Metrics
+import no.nav.helsearbeidsgiver.felles.rapidsrivers.KafkaKey
 import no.nav.helsearbeidsgiver.felles.rapidsrivers.model.Fail
 import no.nav.helsearbeidsgiver.felles.rapidsrivers.river.ObjectRiver
 import no.nav.helsearbeidsgiver.felles.utils.Log
@@ -26,6 +28,7 @@ data class HentArbeidsforholdMelding(
     val behovType: BehovType,
     val transaksjonId: UUID,
     val data: Map<Key, JsonElement>,
+    val svarKafkaKey: KafkaKey?,
     val fnr: Fnr,
 )
 
@@ -46,6 +49,7 @@ class HentArbeidsforholdRiver(
                 behovType = Key.BEHOV.krev(BehovType.HENT_ARBEIDSFORHOLD, BehovType.serializer(), json),
                 transaksjonId = Key.KONTEKST_ID.les(UuidSerializer, json),
                 data = data,
+                svarKafkaKey = Key.SVAR_KAFKA_KEY.lesOrNull(KafkaKey.serializer(), data),
                 fnr = Key.FNR.les(Fnr.serializer(), data),
             )
         }

aareg/src/test/kotlin/no/nav/helsearbeidsgiver/inntektsmelding/inntekt/HentArbeidsforholdRiverTest.kt

@@ -131,6 +131,7 @@ private object Mock {
                 mapOf(
                     Key.FNR to fnr.toJson(Fnr.serializer()),
                 ),
+            svarKafkaKey = null,
             fnr = fnr,
         )
     }

aktiveorgnrservice/src/main/kotlin/no/nav/helsearbeidsgiver/inntektsmelding/aktiveorgnrservice/AktiveOrgnrService.kt

@@ -14,6 +14,7 @@ import no.nav.helsearbeidsgiver.felles.domene.ResultJson
 import no.nav.helsearbeidsgiver.felles.json.les
 import no.nav.helsearbeidsgiver.felles.json.personMapSerializer
 import no.nav.helsearbeidsgiver.felles.json.toJson
+import no.nav.helsearbeidsgiver.felles.rapidsrivers.KafkaKey
 import no.nav.helsearbeidsgiver.felles.rapidsrivers.model.Fail
 import no.nav.helsearbeidsgiver.felles.rapidsrivers.publish
 import no.nav.helsearbeidsgiver.felles.rapidsrivers.redis.RedisStore
@@ -95,13 +96,16 @@ class AktiveOrgnrService(
         data: Map<Key, JsonElement>,
         steg0: Steg0,
     ) {
+        val svarKafkaKey = KafkaKey(steg0.sykmeldtFnr)
+
         rapid.publish(
             key = steg0.sykmeldtFnr,
             Key.EVENT_NAME to eventName.toJson(),
             Key.BEHOV to BehovType.ARBEIDSGIVERE.toJson(),
             Key.KONTEKST_ID to steg0.transaksjonId.toJson(),
             Key.DATA to
                 mapOf(
+                    Key.SVAR_KAFKA_KEY to svarKafkaKey.toJson(),
                     Key.ARBEIDSGIVER_FNR to steg0.avsenderFnr.toJson(),
                 ).toJson(),
         )
@@ -113,6 +117,7 @@ class AktiveOrgnrService(
             Key.KONTEKST_ID to steg0.transaksjonId.toJson(),
             Key.DATA to
                 mapOf(
+                    Key.SVAR_KAFKA_KEY to svarKafkaKey.toJson(),
                     Key.FNR to steg0.sykmeldtFnr.toJson(),
                 ).toJson(),
         )
@@ -124,6 +129,7 @@ class AktiveOrgnrService(
             Key.KONTEKST_ID to steg0.transaksjonId.toJson(),
             Key.DATA to
                 mapOf(
+                    Key.SVAR_KAFKA_KEY to svarKafkaKey.toJson(),
                     Key.FNR_LISTE to
                         setOf(
                             steg0.sykmeldtFnr,

aktiveorgnrservice/src/test/kotlin/no/nav/helsearbeidsgiver/inntektsmelding/aktiveorgnrservice/AktiveOrgnrServiceTest.kt

@@ -18,11 +18,14 @@ import no.nav.helsearbeidsgiver.felles.domene.Arbeidsgiver
 import no.nav.helsearbeidsgiver.felles.domene.PeriodeNullable
 import no.nav.helsearbeidsgiver.felles.domene.Person
 import no.nav.helsearbeidsgiver.felles.domene.ResultJson
+import no.nav.helsearbeidsgiver.felles.json.lesOrNull
 import no.nav.helsearbeidsgiver.felles.json.personMapSerializer
 import no.nav.helsearbeidsgiver.felles.json.toJson
+import no.nav.helsearbeidsgiver.felles.rapidsrivers.KafkaKey
 import no.nav.helsearbeidsgiver.felles.rapidsrivers.redis.RedisPrefix
 import no.nav.helsearbeidsgiver.felles.rapidsrivers.service.ServiceRiverStateful
 import no.nav.helsearbeidsgiver.felles.test.json.lesBehov
+import no.nav.helsearbeidsgiver.felles.test.json.lesData
 import no.nav.helsearbeidsgiver.felles.test.json.plusData
 import no.nav.helsearbeidsgiver.felles.test.mock.MockRedis
 import no.nav.helsearbeidsgiver.felles.test.mock.mockFail
@@ -62,9 +65,18 @@ class AktiveOrgnrServiceTest :
             )
 
             testRapid.inspektør.size shouldBeExactly 3
-            testRapid.message(0).lesBehov() shouldBe BehovType.ARBEIDSGIVERE
-            testRapid.message(1).lesBehov() shouldBe BehovType.HENT_ARBEIDSFORHOLD
-            testRapid.message(2).lesBehov() shouldBe BehovType.HENT_PERSONER
+            testRapid.message(0).also {
+                it.lesBehov() shouldBe BehovType.ARBEIDSGIVERE
+                Key.SVAR_KAFKA_KEY.lesOrNull(KafkaKey.serializer(), it.lesData()) shouldBe KafkaKey(Mock.sykmeldtFnr)
+            }
+            testRapid.message(1).also {
+                it.lesBehov() shouldBe BehovType.HENT_ARBEIDSFORHOLD
+                Key.SVAR_KAFKA_KEY.lesOrNull(KafkaKey.serializer(), it.lesData()) shouldBe KafkaKey(Mock.sykmeldtFnr)
+            }
+            testRapid.message(2).also {
+                it.lesBehov() shouldBe BehovType.HENT_PERSONER
+                Key.SVAR_KAFKA_KEY.lesOrNull(KafkaKey.serializer(), it.lesData()) shouldBe KafkaKey(Mock.sykmeldtFnr)
+            }
 
             testRapid.sendJson(
                 Mock.steg1Data(transaksjonId, orgnr),
@@ -180,11 +192,12 @@ class AktiveOrgnrServiceTest :
     })
 
 private object Mock {
+    val sykmeldtFnr = Fnr.genererGyldig()
+
     private const val SYKMELDT_NAVN = "Ole Idole"
     private const val AVSENDER_NAVN = "Ole Jacob Evenrud"
     private const val ORG_NAVN = "Mexican Standup A/S"
 
-    private val sykmeldtFnr = Fnr.genererGyldig()
     private val avsenderFnr = Fnr.genererGyldig()
 
     private val personer =

altinn/src/main/kotlin/no/nav/helsearbeidsgiver/inntektsmelding/altinn/AltinnRiver.kt

@@ -8,9 +8,11 @@ import no.nav.helsearbeidsgiver.felles.EventName
 import no.nav.helsearbeidsgiver.felles.Key
 import no.nav.helsearbeidsgiver.felles.json.krev
 import no.nav.helsearbeidsgiver.felles.json.les
+import no.nav.helsearbeidsgiver.felles.json.lesOrNull
 import no.nav.helsearbeidsgiver.felles.json.toJson
 import no.nav.helsearbeidsgiver.felles.json.toMap
 import no.nav.helsearbeidsgiver.felles.metrics.Metrics
+import no.nav.helsearbeidsgiver.felles.rapidsrivers.KafkaKey
 import no.nav.helsearbeidsgiver.felles.rapidsrivers.model.Fail
 import no.nav.helsearbeidsgiver.felles.rapidsrivers.river.ObjectRiver
 import no.nav.helsearbeidsgiver.felles.utils.Log
@@ -27,6 +29,7 @@ data class Melding(
     val behovType: BehovType,
     val transaksjonId: UUID,
     val data: Map<Key, JsonElement>,
+    val svarKafkaKey: KafkaKey?,
     val fnr: Fnr,
 )
 
@@ -47,6 +50,7 @@ class AltinnRiver(
                 behovType = Key.BEHOV.krev(BehovType.ARBEIDSGIVERE, BehovType.serializer(), json),
                 transaksjonId = Key.KONTEKST_ID.les(UuidSerializer, json),
                 data = data,
+                svarKafkaKey = Key.SVAR_KAFKA_KEY.lesOrNull(KafkaKey.serializer(), data),
                 fnr = Key.ARBEIDSGIVER_FNR.les(Fnr.serializer(), data),
             )
         }

altinn/src/test/kotlin/no/nav/helsearbeidsgiver/inntektsmelding/altinn/Mock.kt

@@ -24,6 +24,7 @@ object Mock {
                 mapOf(
                     Key.ARBEIDSGIVER_FNR to fnr.toJson(),
                 ),
+            svarKafkaKey = null,
             fnr = fnr,
         )
     }

berik-inntektsmelding-service/src/test/kotlin/no.nav.helsearbeidsgiver.inntektsmelding.berikinntektsmeldingservice/BerikInntektsmeldingServiceTest.kt

@@ -56,46 +56,42 @@ class BerikInntektsmeldingServiceTest :
         }
 
         test("nytt inntektsmeldingskjema berikes, lagres og sendes videre til journalføring") {
-            testRapid.sendJson(Mock.steg0(Mock.transaksjonId))
+            val transaksjonId = UUID.randomUUID()
+
+            testRapid.sendJson(Mock.steg0(transaksjonId))
 
             // Melding med forventet behov og data sendt for å hente forespørsel
             testRapid.inspektør.size shouldBeExactly 1
             testRapid.message(0).also {
                 it.lesBehov() shouldBe BehovType.HENT_TRENGER_IM
 
-                Key.FORESPOERSEL_ID.lesOrNull(
-                    serializer = UuidSerializer,
-                    melding = it.lesData(),
-                ) shouldBe Mock.skjema.forespoerselId
+                val data = it.lesData()
+                Key.FORESPOERSEL_ID.lesOrNull(UuidSerializer, data) shouldBe Mock.skjema.forespoerselId
             }
 
-            testRapid.sendJson(Mock.steg1(Mock.transaksjonId))
+            testRapid.sendJson(Mock.steg1(transaksjonId))
 
             // Melding med forventet behov og data sendt for å hente virksomhetsnavn
             testRapid.inspektør.size shouldBeExactly 2
             testRapid.message(1).also {
                 it.lesBehov() shouldBe BehovType.HENT_VIRKSOMHET_NAVN
 
-                Key.ORGNR_UNDERENHETER.lesOrNull(
-                    serializer = Orgnr.serializer().set(),
-                    melding = it.lesData(),
-                ) shouldNotBe null
+                val data = it.lesData()
+                Key.ORGNR_UNDERENHETER.lesOrNull(Orgnr.serializer().set(), data) shouldNotBe null
             }
 
-            testRapid.sendJson(Mock.steg2(Mock.transaksjonId))
+            testRapid.sendJson(Mock.steg2(transaksjonId))
 
             // Melding med forventet behov og data sendt for å hente personnavn
             testRapid.inspektør.size shouldBeExactly 3
             testRapid.message(2).also {
                 it.lesBehov() shouldBe BehovType.HENT_PERSONER
 
-                Key.FNR_LISTE.lesOrNull(
-                    serializer = Fnr.serializer().set(),
-                    melding = it.lesData(),
-                ) shouldNotBe null
+                val data = it.lesData()
+                Key.FNR_LISTE.lesOrNull(Fnr.serializer().set(), data) shouldNotBe null
             }
 
-            testRapid.sendJson(Mock.steg3(Mock.transaksjonId))
+            testRapid.sendJson(Mock.steg3(transaksjonId))
 
             // Melding med forventet behov og data sendt for å lagre inntektsmelding
             testRapid.inspektør.size shouldBeExactly 4
@@ -108,7 +104,7 @@ class BerikInntektsmeldingServiceTest :
                 Key.INNSENDING_ID.lesOrNull(Long.serializer(), data) shouldBe Mock.INNSENDING_ID
             }
 
-            testRapid.sendJson(Mock.steg4(Mock.transaksjonId))
+            testRapid.sendJson(Mock.steg4(transaksjonId))
 
             // Inntektsmelding sendt videre til journalføring med forventet data
             testRapid.inspektør.size shouldBeExactly 5
@@ -126,7 +122,7 @@ class BerikInntektsmeldingServiceTest :
         test("duplikat IM sendes _ikke_ videre til journalføring") {
             testRapid.sendJson(
                 Mock
-                    .steg4(Mock.transaksjonId)
+                    .steg4(UUID.randomUUID())
                     .plusData(Key.ER_DUPLIKAT_IM to true.toJson(Boolean.serializer())),
             )
 
@@ -150,72 +146,63 @@ class BerikInntektsmeldingServiceTest :
 private object Mock {
     const val INNSENDING_ID = 1L
 
-    val transaksjonId: UUID = UUID.randomUUID()
-    val forespoersel = mockForespoersel()
     val skjema = mockSkjemaInntektsmelding()
-    val inntektsmelding = mockInntektsmeldingV1()
 
-    val avsender =
+    private val forespoersel = mockForespoersel()
+
+    private val avsender =
         Person(
             fnr = Fnr.genererGyldig(),
             navn = "Skrue McDuck",
         )
 
-    val sykmeldt =
+    private val sykmeldt =
         Person(
             fnr = Fnr(forespoersel.fnr),
             navn = "Dolly Duck",
         )
 
-    val orgnrMedNavn = mapOf(Orgnr(forespoersel.orgnr) to "Lasses kasserollesjappe")
+    private val orgnrMedNavn = mapOf(Orgnr(forespoersel.orgnr) to "Lasses kasserollesjappe")
 
-    val personer =
+    private val personer =
         mapOf(
             avsender.fnr to avsender,
             sykmeldt.fnr to sykmeldt,
         )
 
-    val steg0data =
+    fun steg0(transaksjonId: UUID): Map<Key, JsonElement> =
         mapOf(
-            Key.ARBEIDSGIVER_FNR to avsender.fnr.toJson(),
-            Key.SKJEMA_INNTEKTSMELDING to skjema.toJson(SkjemaInntektsmelding.serializer()),
-            Key.INNSENDING_ID to INNSENDING_ID.toJson(Long.serializer()),
+            Key.EVENT_NAME to EventName.INNTEKTSMELDING_SKJEMA_LAGRET.toJson(),
+            Key.KONTEKST_ID to transaksjonId.toJson(),
+            Key.DATA to
+                mapOf(
+                    Key.ARBEIDSGIVER_FNR to avsender.fnr.toJson(),
+                    Key.SKJEMA_INNTEKTSMELDING to skjema.toJson(SkjemaInntektsmelding.serializer()),
+                    Key.INNSENDING_ID to INNSENDING_ID.toJson(Long.serializer()),
+                ).toJson(),
         )
 
-    val steg1data =
-        mapOf(
+    fun steg1(transaksjonId: UUID): Map<Key, JsonElement> =
+        steg0(transaksjonId).plusData(
             Key.FORESPOERSEL_SVAR to forespoersel.toJson(Forespoersel.serializer()),
         )
 
-    val steg2data =
-        mapOf(
+    fun steg2(transaksjonId: UUID): Map<Key, JsonElement> =
+        steg1(transaksjonId).plusData(
             Key.VIRKSOMHETER to orgnrMedNavn.toJson(orgMapSerializer),
         )
 
-    val steg3data =
-        mapOf(
+    fun steg3(transaksjonId: UUID): Map<Key, JsonElement> =
+        steg2(transaksjonId).plusData(
             Key.PERSONER to personer.toJson(personMapSerializer),
         )
 
-    val steg4data =
-        mapOf(
-            Key.ER_DUPLIKAT_IM to false.toJson(Boolean.serializer()),
-            Key.INNTEKTSMELDING to inntektsmelding.toJson(InntektsmeldingV1.serializer()),
-            Key.BESTEMMENDE_FRAVAERSDAG to 20.oktober.toJson(),
+    fun steg4(transaksjonId: UUID): Map<Key, JsonElement> =
+        steg3(transaksjonId).plusData(
+            mapOf(
+                Key.ER_DUPLIKAT_IM to false.toJson(Boolean.serializer()),
+                Key.INNTEKTSMELDING to mockInntektsmeldingV1().toJson(InntektsmeldingV1.serializer()),
+                Key.BESTEMMENDE_FRAVAERSDAG to 20.oktober.toJson(),
+            ),
         )
-
-    fun steg0(transaksjonId: UUID): Map<Key, JsonElement> =
-        mapOf(
-            Key.EVENT_NAME to EventName.INNTEKTSMELDING_SKJEMA_LAGRET.toJson(),
-            Key.KONTEKST_ID to transaksjonId.toJson(),
-            Key.DATA to steg0data.toJson(),
-        )
-
-    fun steg1(transaksjonId: UUID): Map<Key, JsonElement> = steg0(transaksjonId).plusData(steg1data)
-
-    fun steg2(transaksjonId: UUID): Map<Key, JsonElement> = steg1(transaksjonId).plusData(steg2data)
-
-    fun steg3(transaksjonId: UUID): Map<Key, JsonElement> = steg2(transaksjonId).plusData(steg3data)
-
-    fun steg4(transaksjonId: UUID): Map<Key, JsonElement> = steg3(transaksjonId).plusData(steg4data)
 }

brreg/src/main/kotlin/no/nav/helsearbeidsgiver/inntektsmelding/brreg/HentVirksomhetNavnRiver.kt

@@ -8,9 +8,11 @@ import no.nav.helsearbeidsgiver.felles.EventName
 import no.nav.helsearbeidsgiver.felles.Key
 import no.nav.helsearbeidsgiver.felles.json.krev
 import no.nav.helsearbeidsgiver.felles.json.les
+import no.nav.helsearbeidsgiver.felles.json.lesOrNull
 import no.nav.helsearbeidsgiver.felles.json.toJson
 import no.nav.helsearbeidsgiver.felles.json.toMap
 import no.nav.helsearbeidsgiver.felles.metrics.Metrics
+import no.nav.helsearbeidsgiver.felles.rapidsrivers.KafkaKey
 import no.nav.helsearbeidsgiver.felles.rapidsrivers.model.Fail
 import no.nav.helsearbeidsgiver.felles.rapidsrivers.river.ObjectRiver
 import no.nav.helsearbeidsgiver.felles.utils.Log
@@ -27,6 +29,7 @@ data class HentVirksomhetMelding(
     val behovType: BehovType,
     val transaksjonId: UUID,
     val data: Map<Key, JsonElement>,
+    val svarKafkaKey: KafkaKey?,
     val orgnr: Set<Orgnr>,
 )
 
@@ -48,6 +51,7 @@ class HentVirksomhetNavnRiver(
                 behovType = Key.BEHOV.krev(BehovType.HENT_VIRKSOMHET_NAVN, BehovType.serializer(), json),
                 transaksjonId = Key.KONTEKST_ID.les(UuidSerializer, json),
                 data = data,
+                svarKafkaKey = Key.SVAR_KAFKA_KEY.lesOrNull(KafkaKey.serializer(), data),
                 orgnr = Key.ORGNR_UNDERENHETER.les(Orgnr.serializer().set(), data),
             )
         }

brreg/src/test/kotlin/no/nav/helsearbeidsgiver/inntektsmelding/brreg/HentVirksomhetNavnRiverTest.kt

@@ -190,6 +190,7 @@ private object Mock {
                 mapOf(
                     Key.ORGNR_UNDERENHETER to orgnr.toJson(Orgnr.serializer()),
                 ),
+            svarKafkaKey = null,
             orgnr = orgnr,
         )
 

felles/src/main/kotlin/no/nav/helsearbeidsgiver/felles/Key.kt

@@ -43,6 +43,7 @@ enum class Key : IKey {
     SKAL_HA_PAAMINNELSE,
     SKJEMA_INNTEKTSMELDING,
     SPINN_INNTEKTSMELDING_ID,
+    SVAR_KAFKA_KEY,
     SYKMELDT,
     TILGANG,
     VEDTAKSPERIODE_ID_LISTE,