KafkaProducerRecordProcessor

Tar hensyn til rekkefølge per topic meldinger blir skrevet til db.
Selve repository laget som sørger for å hente i riktig rekkefølge.

Author: fredrikpe

common/kafka/src/main/kotlin/no/nav/mulighetsrommet/kafka/KafkaProducerRecordProcessor.kt

@@ -0,0 +1,99 @@
+package no.nav.mulighetsrommet.kafka
+
+import kotlinx.coroutines.*
+import no.nav.common.kafka.producer.KafkaProducerClient
+import no.nav.common.kafka.util.KafkaUtils
+import org.apache.kafka.clients.producer.ProducerRecord
+import org.apache.kafka.clients.producer.RecordMetadata
+import org.slf4j.Logger
+import org.slf4j.LoggerFactory
+import java.util.*
+import java.util.concurrent.CountDownLatch
+import java.util.function.Consumer
+
+interface ProducerRecordRepository {
+    fun getRecords(): List<ProducerRecordDbo>
+    fun deleteRecords(ids: List<Long>)
+}
+
+class KafkaProducerRecordProcessor(
+    private val repository: ProducerRecordRepository,
+    private val producerClient: KafkaProducerClient<ByteArray, ByteArray>,
+) {
+    private val pollTimeout: Long = 3000
+
+    private val log: Logger = LoggerFactory.getLogger(this.javaClass)
+    private var job: Job? = null
+
+    fun start() {
+        if (job != null) return
+
+        job = CoroutineScope(Dispatchers.Default).launch {
+            while (isActive) {
+                recordHandler()
+            }
+        }.apply {
+            invokeOnCompletion {
+                log.info("Shutting down Kafka producer")
+                producerClient.close()
+            }
+        }
+    }
+
+    fun close() {
+        job?.cancel()
+        job = null
+    }
+
+    private suspend fun recordHandler() {
+        try {
+            val records = repository.getRecords()
+
+            if (records.isNotEmpty()) {
+                publishRecords(records)
+            } else {
+                delay(pollTimeout)
+            }
+        } catch (e: Exception) {
+            log.error("Failed to process kafka producer records", e)
+            delay(pollTimeout)
+        }
+    }
+
+    private fun publishRecords(records: List<ProducerRecordDbo>) {
+        val idsToDelete = Collections.synchronizedList(mutableListOf<Long>())
+
+        val latch = CountDownLatch(records.size)
+
+        records.forEach(
+            Consumer { record ->
+                producerClient.send(record.toProducerRecord()) { _: RecordMetadata?, exception: java.lang.Exception? ->
+                    latch.countDown()
+                    if (exception != null) {
+                        log.warn("Failed to send record to topic ${record.topic}", exception)
+                    } else {
+                        idsToDelete.add(record.id)
+                    }
+                }
+            },
+        )
+
+        producerClient.producer.flush()
+        latch.await()
+
+        repository.deleteRecords(idsToDelete)
+    }
+}
+
+fun ProducerRecordDbo.toProducerRecord(): ProducerRecord<ByteArray, ByteArray> {
+    val headers = KafkaUtils.jsonToHeaders(headersJson)
+
+    return ProducerRecord(
+        topic,
+        null,
+        null,
+        key,
+        value,
+        headers,
+    )
+}

common/kafka/src/main/kotlin/no/nav/mulighetsrommet/kafka/KafkaProducerRecordQueries.kt

@@ -0,0 +1,57 @@
+package no.nav.mulighetsrommet.kafka
+
+import kotliquery.Row
+import kotliquery.Session
+import kotliquery.queryOf
+import no.nav.common.kafka.util.KafkaUtils
+import org.apache.kafka.clients.producer.ProducerRecord
+import org.intellij.lang.annotations.Language
+
+class KafkaProducerRecordQueries(private val session: Session) {
+    fun insert(record: ProducerRecord<ByteArray, ByteArray?>) {
+        @Language("PostgreSQL")
+        val sql = """
+            insert into kafka_producer_record (topic, key, value, headers_json) values (?, ?, ?, ?)
+        """.trimIndent()
+
+        val query = queryOf(sql, record.topic(), record.key(), record.value(), KafkaUtils.headersToJson(record.headers()))
+
+        session.execute(query)
+    }
+
+    fun deleteRecords(ids: List<Long>) {
+        @Language("PostgreSQL")
+        val sql = """
+            delete from kafka_producer_record where id = any(?::bigint[])
+        """.trimIndent()
+
+        session.update(queryOf(sql, session.createArrayOf("bigint", ids)))
+    }
+
+    fun getRecords(): List<ProducerRecordDbo> {
+        @Language("PostgreSQL")
+        val sql = """
+            select distinct on (topic) * from kafka_producer_record order by topic, id
+        """.trimIndent()
+
+        val query = queryOf(sql)
+
+        return session.list(query) { it.toProducerRecordDbo() }
+    }
+}
+
+fun Row.toProducerRecordDbo() = ProducerRecordDbo(
+    long("id"),
+    string("topic"),
+    bytes("key"),
+    bytes("value"),
+    stringOrNull("headers_json"),
+)
+
+data class ProducerRecordDbo(
+    val id: Long? = 0,
+    val topic: String,
+    val key: ByteArray,
+    val value: ByteArray?,
+    val headersJson: String?,
+)

common/kafka/src/main/kotlin/no/nav/mulighetsrommet/kafka/KafkaProducerRepositoryImpl.kt

@@ -20,6 +20,7 @@ import no.nav.mulighetsrommet.api.utbetaling.db.UtbetalingQueries
 import no.nav.mulighetsrommet.api.veilederflate.VeilederJoyrideQueries
 import no.nav.mulighetsrommet.api.veilederflate.VeilederflateTiltakQueries
 import no.nav.mulighetsrommet.database.Database
+import no.nav.mulighetsrommet.kafka.KafkaProducerRecordQueries
 import no.nav.mulighetsrommet.notifications.NotificationQueries
 import no.nav.mulighetsrommet.utdanning.db.UtdanningQueries
 import javax.sql.DataSource
@@ -72,5 +73,6 @@ class QueryContext(val session: Session) {
         val totrinnskontroll = TotrinnskontrollQueries(session)
         val veilderTiltak = VeilederflateTiltakQueries(session)
         val veilederJoyride = VeilederJoyrideQueries(session)
+        val kafkaProducerRecord = KafkaProducerRecordQueries(session)
     }
 }

common/kafka/src/test/kotlin/no/nav/mulighetsrommet/kafka/KafkaProducerRepositoryTest.kt

@@ -79,6 +79,9 @@ import no.nav.mulighetsrommet.brreg.BrregClient
 import no.nav.mulighetsrommet.database.Database
 import no.nav.mulighetsrommet.database.DatabaseConfig
 import no.nav.mulighetsrommet.kafka.KafkaConsumerOrchestrator
+import no.nav.mulighetsrommet.kafka.KafkaProducerRecordProcessor
+import no.nav.mulighetsrommet.kafka.ProducerRecordDbo
+import no.nav.mulighetsrommet.kafka.ProducerRecordRepository
 import no.nav.mulighetsrommet.metrics.Metrikker
 import no.nav.mulighetsrommet.notifications.NotificationTask
 import no.nav.mulighetsrommet.oppgaver.OppgaverService
@@ -138,7 +141,6 @@ private fun kafka(appConfig: AppConfig) = module {
             .withMetrics(Metrikker.appMicrometerRegistry)
             .build()
     }
-
     single {
         ArenaMigreringTiltaksgjennomforingerV1KafkaProducer(
             get(),
@@ -180,6 +182,21 @@ private fun kafka(appConfig: AppConfig) = module {
             consumers = consumers,
         )
     }
+    single {
+        val db = get<ApiDatabase>()
+        val repository = object : ProducerRecordRepository {
+            override fun getRecords(): List<ProducerRecordDbo> {
+                return db.session { queries.kafkaProducerRecord.getRecords() }
+            }
+            override fun deleteRecords(ids: List<Long>) {
+                return db.session { queries.kafkaProducerRecord.deleteRecords(ids) }
+            }
+        }
+        KafkaProducerRecordProcessor(
+            repository = repository,
+            producerClient = get(),
+        )
+    }
 }
 
 private fun services(appConfig: AppConfig) = module {
@@ -367,7 +384,7 @@ private fun services(appConfig: AppConfig) = module {
     single { TilsagnService(appConfig.okonomi, get(), get()) }
     single { AltinnRettigheterService(get(), get()) }
     single { OppgaverService(get()) }
-    single { OkonomiBestillingService(appConfig.kafka.clients.okonomiBestilling, get(), get()) }
+    single { OkonomiBestillingService(appConfig.kafka.clients.okonomiBestilling, get()) }
     single { ArrangorFlateService(get(), get()) }
 }
 
@@ -403,7 +420,6 @@ private fun tasks(config: TaskConfig) = module {
         )
         val updateApentForPamelding = UpdateApentForPamelding(config.updateApentForPamelding, get(), get())
         val notificationTask: NotificationTask by inject()
-        val okonomi: OkonomiBestillingService by inject()
         val generateValidationReport: GenerateValidationReport by inject()
         val initialLoadGjennomforinger: InitialLoadGjennomforinger by inject()
         val initialLoadTiltakstyper: InitialLoadTiltakstyper by inject()
@@ -418,7 +434,6 @@ private fun tasks(config: TaskConfig) = module {
             .create(
                 db.getDatasource(),
                 notificationTask.task,
-                okonomi.task,
                 generateValidationReport.task,
                 initialLoadGjennomforinger.task,
                 initialLoadTiltakstyper.task,