-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Some sort of handling for partial transfers
- Loading branch information
1 parent
6d5a7dd
commit 3c7f6ef
Showing
12 changed files
with
254 additions
and
61 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
58 changes: 58 additions & 0 deletions
58
.../no/nav/pensjon/opptjening/omsorgsopptjening/start/innlesning/config/KafkaErrorHandler.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,58 @@ | ||
| package no.nav.pensjon.opptjening.omsorgsopptjening.start.innlesning.config | ||
|
|
||
| import no.nav.pensjon.opptjening.omsorgsopptjening.start.innlesning.InnlesingRepository | ||
| import no.nav.pensjon.opptjening.omsorgsopptjening.start.innlesning.barnetrygd.InvalidateInnlesingException | ||
| import no.nav.pensjon.opptjening.omsorgsopptjening.start.innlesning.barnetrygd.UkjentKafkaMeldingException | ||
| import org.apache.kafka.clients.consumer.ConsumerRecord | ||
| import org.slf4j.Logger | ||
| import org.slf4j.LoggerFactory | ||
| import org.springframework.kafka.listener.DefaultErrorHandler | ||
| import org.springframework.kafka.listener.RetryListener | ||
| import org.springframework.stereotype.Component | ||
| import org.springframework.util.backoff.FixedBackOff | ||
| import java.util.UUID | ||
|
|
||
| @Component | ||
| class KafkaErrorHandler( | ||
| private val innlesingRepository: InnlesingRepository | ||
| ) : DefaultErrorHandler(FixedBackOff(3000, 3)) { | ||
| init { | ||
| this.setRetryListeners(InnlesingInvalidatingRetryListener(innlesingRepository)) | ||
| this.addNotRetryableExceptions(InvalidateInnlesingException::class.java) | ||
| this.addNotRetryableExceptions(UkjentKafkaMeldingException::class.java) | ||
| } | ||
| } | ||
|
|
||
| class InnlesingInvalidatingRetryListener( | ||
| private val innlesingRepository: InnlesingRepository | ||
| ) : RetryListener { | ||
|
|
||
| private val invalidated: MutableList<UUID> = mutableListOf() | ||
|
|
||
| companion object { | ||
| private val log: Logger = LoggerFactory.getLogger(this::class.java) | ||
| } | ||
|
|
||
| override fun failedDelivery(record: ConsumerRecord<*, *>, ex: Exception, deliveryAttempt: Int) {} | ||
|
|
||
| override fun recovered(record: ConsumerRecord<*, *>, ex: java.lang.Exception) { | ||
| log.error("Processing and retries failed for record: $record, ex: $ex") | ||
| ex.cause?.also { throwable -> | ||
| when (throwable) { | ||
| is InvalidateInnlesingException -> { | ||
| if (!invalidated.contains(throwable.innlesingId)) { | ||
| log.info("Invalidating innlesing with id: ${throwable.innlesingId} due to all records not being processed successfully.") | ||
| innlesingRepository.invalider(throwable.innlesingId) | ||
| .also { invalidated.add(throwable.innlesingId) } | ||
| log.info("Invalidated id: ${throwable.innlesingId}") | ||
| } | ||
| } | ||
|
|
||
| else -> { | ||
| //NOOP | ||
| } | ||
| } | ||
| } | ||
| super.recovered(record, ex) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,7 @@ | ||
| alter table barnetrygdmottaker | ||
| drop constraint barnetrygdmottaker_innlesing_id_fkey, | ||
| add foreign key (innlesing_id) references innlesing(id) on delete cascade; | ||
|
|
||
| alter table barnetrygdmottaker_status | ||
| drop constraint barnetrygdmottaker_status_id_fkey, | ||
| add foreign key (id) references barnetrygdmottaker(id) on delete cascade; |
42 changes: 42 additions & 0 deletions
42
...nav/pensjon/opptjening/omsorgsopptjening/start/innlesning/InnlesingCascadingDeleteTest.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,42 @@ | ||
| package no.nav.pensjon.opptjening.omsorgsopptjening.start.innlesning | ||
|
|
||
| import no.nav.pensjon.opptjening.omsorgsopptjening.felles.CorrelationId | ||
| import no.nav.pensjon.opptjening.omsorgsopptjening.felles.InnlesingId | ||
| import no.nav.pensjon.opptjening.omsorgsopptjening.start.innlesning.barnetrygd.Barnetrygdmottaker | ||
| import no.nav.pensjon.opptjening.omsorgsopptjening.start.innlesning.barnetrygd.BarnetrygdmottakerRepository | ||
| import no.nav.pensjon.opptjening.omsorgsopptjening.start.innlesning.barnetrygd.SpringContextTest | ||
| import org.springframework.beans.factory.annotation.Autowired | ||
| import kotlin.test.Test | ||
| import kotlin.test.assertEquals | ||
| import kotlin.test.assertNull | ||
|
|
||
| class InnlesingCascadingDeleteTest : SpringContextTest.NoKafka() { | ||
| @Autowired | ||
| private lateinit var innlesingRepository: InnlesingRepository | ||
|
|
||
| @Autowired | ||
| private lateinit var barnetrygdmottakerRepository: BarnetrygdmottakerRepository | ||
|
|
||
| @Test | ||
| fun `invalidering av innlesing sletter alle barnetrygdmottakere knyttet til innlesingen`() { | ||
| val a = Innlesing( | ||
| id = InnlesingId.generate(), | ||
| år = 2023 | ||
| ) | ||
|
|
||
| val b = Barnetrygdmottaker( | ||
| ident = "12345678910", | ||
| correlationId = CorrelationId.generate(), | ||
| innlesingId = a.id | ||
| ) | ||
|
|
||
| val aa = innlesingRepository.bestilt(a) | ||
| val bb = barnetrygdmottakerRepository.save(b) | ||
|
|
||
| assertEquals(aa, innlesingRepository.finn(a.id.toString())) | ||
| assertEquals(bb, barnetrygdmottakerRepository.find(bb.id!!)) | ||
|
|
||
| innlesingRepository.invalider(aa.id.toUUID()) | ||
| assertNull(barnetrygdmottakerRepository.find(bb.id!!)) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.