Endret til å benytte individuelle JDBC env vars

apps/kafka-key-generator/src/main/kotlin/no/nav/paw/kafkakeygenerator/Application.kt

@@ -21,7 +21,7 @@ import no.nav.paw.kafkakeygenerator.config.KAFKA_TOPOLOGY_CONFIG
 import no.nav.paw.kafkakeygenerator.config.KafkaTopologyConfig
 import no.nav.paw.kafkakeygenerator.config.PDL_CLIENT_CONFIG
 import no.nav.paw.kafkakeygenerator.config.PdlClientConfig
-import no.nav.paw.kafkakeygenerator.database.createDataSource
+import no.nav.paw.kafkakeygenerator.utils.createDataSource
 import no.nav.paw.kafkakeygenerator.merge.MergeDetector
 import no.nav.paw.kafkakeygenerator.plugin.configSerialization
 import no.nav.paw.kafkakeygenerator.plugin.configureAuthentication

apps/kafka-key-generator/src/main/kotlin/no/nav/paw/kafkakeygenerator/config/DatabaseConfig.kt

@@ -3,7 +3,13 @@ package no.nav.paw.kafkakeygenerator.config
 const val DATABASE_CONFIG = "database_config.toml"
 
 data class DatabaseConfig(
-    val jdbcUrl: String,
+    val host: String,
+    val port: Int,
+    val database: String,
+    val username: String,
+    val password: String,
     val driverClassName: String,
     val autoCommit: Boolean
-)
+) {
+    val jdbcUrl = "jdbc:postgresql://$host:$port/$database?user=$username&password=$password"
+}

apps/kafka-key-generator/src/main/kotlin/no/nav/paw/kafkakeygenerator/plugin/Kafka.kt

@@ -7,8 +7,8 @@ import org.apache.kafka.clients.consumer.ConsumerRecords
 import org.apache.kafka.clients.consumer.KafkaConsumer
 
 fun <K, V> Application.configureKafka(
-    consumeFunction: ((Sequence<ConsumerRecords<K, V>>) -> Unit),
-    successFunction: ((Unit) -> Unit)? = null,
+    consumeFunction: ((ConsumerRecords<K, V>) -> Unit),
+    successFunction: ((ConsumerRecords<K, V>) -> Unit)? = null,
     errorFunction: ((throwable: Throwable) -> Unit),
     kafkaConsumer: KafkaConsumer<K, V>,
     kafkaTopics: List<String>

apps/kafka-key-generator/src/main/kotlin/no/nav/paw/kafkakeygenerator/plugin/custom/KafkaConsumerPlugin.kt

@@ -12,7 +12,6 @@ import io.ktor.util.KtorDsl
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.launch
-import no.nav.paw.config.kafka.asSequence
 import no.nav.paw.kafkakeygenerator.listener.NoopConsumerRebalanceListener
 import no.nav.paw.kafkakeygenerator.utils.buildApplicationLogger
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener
@@ -21,12 +20,13 @@ import org.apache.kafka.clients.consumer.KafkaConsumer
 import java.time.Duration
 import java.util.concurrent.atomic.AtomicBoolean
 
+private val logger = buildApplicationLogger
 val KafkaConsumerReady: EventDefinition<Application> = EventDefinition()
 
 @KtorDsl
-class KafkaConsumerPluginConfig<K, V> {
-    var consumeFunction: ((Sequence<ConsumerRecords<K, V>>) -> Unit)? = null
-    var successFunction: ((Unit) -> Unit)? = null
+class KafkaConsumerPluginConfig<K, V, R> {
+    var consumeFunction: ((ConsumerRecords<K, V>) -> Unit)? = null
+    var successFunction: ((ConsumerRecords<K, V>) -> Unit)? = null
     var errorFunction: ((throwable: Throwable) -> Unit)? = null
     var kafkaConsumer: KafkaConsumer<K, V>? = null
     var kafkaTopics: Collection<String>? = null
@@ -40,15 +40,26 @@ class KafkaConsumerPluginConfig<K, V> {
     }
 }
 
-fun <K, V> kafkaConsumerPlugin(): ApplicationPlugin<KafkaConsumerPluginConfig<K, V>> =
+private fun <K, V> KafkaConsumer<K, V>.defaultSuccessFunction(records: ConsumerRecords<K, V>) {
+    if (!records.isEmpty) {
+        logger.debug("Kafka Consumer success. {} records processed", records.count())
+        this.commitSync()
+    }
+}
+
+private fun defaultErrorFunction(throwable: Throwable) {
+    logger.error("Kafka Consumer failed", throwable)
+    throw throwable
+}
+
+fun <K, V> kafkaConsumerPlugin(): ApplicationPlugin<KafkaConsumerPluginConfig<K, V, Unit>> =
     createApplicationPlugin(KafkaConsumerPluginConfig.PLUGIN_NAME, ::KafkaConsumerPluginConfig) {
         application.log.info("Oppretter {}", KafkaConsumerPluginConfig.PLUGIN_NAME)
-        val logger = buildApplicationLogger
-        val consumeFunction = requireNotNull(pluginConfig.consumeFunction) { "ConsumeFunction er null" }
-        val successFunction = pluginConfig.successFunction ?: { logger.debug("Kafka Consumer poll fullførte") }
-        val errorFunction = pluginConfig.errorFunction ?: { logger.error("Kafka Consumer poll feilet") }
-        val kafkaConsumer = requireNotNull(pluginConfig.kafkaConsumer) { "KafkaConsumer er null" }
         val kafkaTopics = requireNotNull(pluginConfig.kafkaTopics) { "KafkaTopics er null" }
+        val kafkaConsumer = requireNotNull(pluginConfig.kafkaConsumer) { "KafkaConsumer er null" }
+        val consumeFunction = requireNotNull(pluginConfig.consumeFunction) { "ConsumeFunction er null" }
+        val successFunction = pluginConfig.successFunction ?: kafkaConsumer::defaultSuccessFunction
+        val errorFunction = pluginConfig.errorFunction ?: ::defaultErrorFunction
         val pollTimeout = pluginConfig.pollTimeout ?: Duration.ofMillis(100)
         val closeTimeout = pluginConfig.closeTimeout ?: Duration.ofSeconds(1)
         val rebalanceListener = pluginConfig.rebalanceListener ?: NoopConsumerRebalanceListener()
@@ -72,15 +83,15 @@ fun <K, V> kafkaConsumerPlugin(): ApplicationPlugin<KafkaConsumerPluginConfig<K,
         on(MonitoringEvent(KafkaConsumerReady)) { application ->
             consumeJob = application.launch(Dispatchers.IO) {
                 logger.info("Kafka Consumer starter")
-                kafkaConsumer
-                    .asSequence(
-                        stop = shutdownFlag,
-                        pollTimeout = pollTimeout,
-                        closeTimeout = closeTimeout
-                    )
-                    .runCatching(consumeFunction)
-                    .mapCatching { kafkaConsumer.commitSync() }
-                    .fold(onSuccess = successFunction, onFailure = errorFunction)
+                while (!shutdownFlag.get()) {
+                    try {
+                        val records = kafkaConsumer.poll(pollTimeout)
+                        consumeFunction(records)
+                        successFunction(records)
+                    } catch (throwable: Throwable) {
+                        errorFunction(throwable)
+                    }
+                }
                 logger.info("Kafka Consumer avsluttet")
             }
         }

apps/kafka-key-generator/src/main/kotlin/no/nav/paw/kafkakeygenerator/service/KafkaConsumerService.kt

@@ -14,7 +14,6 @@ import no.nav.paw.kafkakeygenerator.repository.IdentitetRepository
 import no.nav.paw.kafkakeygenerator.repository.KafkaKeysAuditRepository
 import no.nav.paw.kafkakeygenerator.utils.buildErrorLogger
 import no.nav.paw.kafkakeygenerator.utils.buildLogger
-import no.nav.paw.kafkakeygenerator.utils.countKafkaFailed
 import no.nav.paw.kafkakeygenerator.utils.countKafkaIgnored
 import no.nav.paw.kafkakeygenerator.utils.countKafkaInserted
 import no.nav.paw.kafkakeygenerator.utils.countKafkaProcessed
@@ -44,30 +43,38 @@ class KafkaConsumerService(
         .addReadinessIndicator(ReadinessHealthIndicator(HealthStatus.HEALTHY))
 
     @WithSpan
-    fun handleRecords(sequence: Sequence<ConsumerRecords<Long, Hendelse>>) {
-        sequence.forEach { records ->
-            records
-                .map { it.value() }
-                .onEach {
-                    meterRegistry.countKafkaReceived()
-                    if (it is IdentitetsnummerSammenslaatt) {
-                        logger.debug("Prosesserer hendelse av type {}", it.hendelseType)
-                        meterRegistry.countKafkaProcessed()
-                    } else {
-                        logger.debug("Ignorerer hendelse av type {}", it.hendelseType)
-                        meterRegistry.countKafkaIgnored()
-                    }
-                }
-                .filterIsInstance<IdentitetsnummerSammenslaatt>()
-                .forEach { hendelse ->
-                    logger.info("Mottok hendelse om sammenslåing av Identitetsnummer")
-                    val identitetsnummer = hendelse.alleIdentitetsnummer
-                        .map { Identitetsnummer(it) } + Identitetsnummer(hendelse.identitetsnummer)
-                    val fraArbeidssoekerId = ArbeidssoekerId(hendelse.id)
-                    val tilArbeidssoekerId = ArbeidssoekerId(hendelse.flyttetTilArbeidssoekerId)
-                    updateIdentiteter(HashSet(identitetsnummer), fraArbeidssoekerId, tilArbeidssoekerId)
+    fun handleRecords(
+        records: ConsumerRecords<Long, Hendelse>
+    ) {
+        records
+            .onEach { record ->
+                logger.debug(
+                    "Mottok melding på topic: {}, partition: {}, offset {}",
+                    record.topic(),
+                    record.partition(),
+                    record.offset()
+                )
+            }
+            .map { it.value() }
+            .onEach { event ->
+                meterRegistry.countKafkaReceived()
+                if (event is IdentitetsnummerSammenslaatt) {
+                    logger.debug("Prosesserer hendelse av type {}", event.hendelseType)
+                    meterRegistry.countKafkaProcessed()
+                } else {
+                    logger.debug("Ignorerer hendelse av type {}", event.hendelseType)
+                    meterRegistry.countKafkaIgnored()
                 }
-        }
+            }
+            .filterIsInstance<IdentitetsnummerSammenslaatt>()
+            .forEach { event ->
+                logger.info("Mottok hendelse om sammenslåing av Identitetsnummer")
+                val identitetsnummer = event.alleIdentitetsnummer
+                    .map { Identitetsnummer(it) } + Identitetsnummer(event.identitetsnummer)
+                val fraArbeidssoekerId = ArbeidssoekerId(event.id)
+                val tilArbeidssoekerId = ArbeidssoekerId(event.flyttetTilArbeidssoekerId)
+                updateIdentiteter(HashSet(identitetsnummer), fraArbeidssoekerId, tilArbeidssoekerId)
+            }
     }
 
     private fun updateIdentiteter(
@@ -76,19 +83,6 @@ class KafkaConsumerService(
         tilArbeidssoekerId: ArbeidssoekerId
     ) {
         transaction(database) {
-            identitetRepository.find(fraArbeidssoekerId).let {
-                if (it == null) {
-                    meterRegistry.countKafkaFailed()
-                    throw IllegalStateException("ArbeidssøkerId ikke funnet")
-                }
-            }
-            identitetRepository.find(tilArbeidssoekerId).let {
-                if (it == null) {
-                    meterRegistry.countKafkaFailed()
-                    throw IllegalStateException("ArbeidssøkerId ikke funnet")
-                }
-            }
-
             identitetsnummerSet.forEach { identitetsnummer ->
                 val kafkaKey = identitetRepository.find(identitetsnummer)
                 if (kafkaKey != null) {

apps/kafka-key-generator/src/main/kotlin/no/nav/paw/kafkakeygenerator/database/DataSource.kt → apps/kafka-key-generator/src/main/kotlin/no/nav/paw/kafkakeygenerator/utils/DataSourceFactory.kt

@@ -1,4 +1,4 @@
-package no.nav.paw.kafkakeygenerator.database
+package no.nav.paw.kafkakeygenerator.utils
 
 import com.zaxxer.hikari.HikariConfig
 import com.zaxxer.hikari.HikariDataSource

apps/kafka-key-generator/src/main/resources/local/database_config.toml

@@ -1,3 +1,7 @@
-jdbcUrl = "jdbc:postgresql://localhost:5432/pawkafkakeys?user=admin&password=admin"
+host = "localhost"
+port = 5432
+database = "pawkafkakeys"
+username = "admin"
+password = "admin"
 driverClassName = "org.postgresql.Driver"
 autoCommit = false

apps/kafka-key-generator/src/main/resources/logback.xml

@@ -22,7 +22,7 @@
             </root>
             <logger name="org.apache.kafka.clients.consumer.ConsumerConfig" level="WARN" />
             <logger name="org.apache.kafka.clients.producer.ProducerConfig" level="WARN" />
-            <logger name="no.nav.paw.kafkakeygenerator" level="DEBUG" />
+            <logger name="no.nav.paw" level="DEBUG" />
         </else>
     </if>
 </configuration>

apps/kafka-key-generator/src/main/resources/nais/database_config.toml

@@ -1,3 +1,7 @@
-jdbcUrl = "${NAIS_DATABASE_PAW_KAFKA_KEY_GENERATOR_PAWKAFKAKEYS_URL}"
+host = "${NAIS_DATABASE_PAW_KAFKA_KEY_GENERATOR_PAWKAFKAKEYS_HOST}"
+port = "${NAIS_DATABASE_PAW_KAFKA_KEY_GENERATOR_PAWKAFKAKEYS_PORT}"
+database = "${NAIS_DATABASE_PAW_KAFKA_KEY_GENERATOR_PAWKAFKAKEYS_DATABASE}"
+username = "${NAIS_DATABASE_PAW_KAFKA_KEY_GENERATOR_PAWKAFKAKEYS_USERNAME}"
+password = "${NAIS_DATABASE_PAW_KAFKA_KEY_GENERATOR_PAWKAFKAKEYS_PASSWORD}"
 driverClassName = "org.postgresql.Driver"
 autoCommit = false

apps/kafka-key-generator/src/test/kotlin/no/nav/paw/kafkakeygenerator/service/KafkaConsumerServiceTest.kt

@@ -12,7 +12,7 @@ import no.nav.paw.kafkakeygenerator.repository.IdentitetRepository
 import no.nav.paw.kafkakeygenerator.repository.KafkaKeysAuditRepository
 import no.nav.paw.kafkakeygenerator.repository.KafkaKeysRepository
 import no.nav.paw.kafkakeygenerator.test.TestData
-import no.nav.paw.kafkakeygenerator.test.asConsumerRecordsSequence
+import no.nav.paw.kafkakeygenerator.test.asConsumerRecords
 import no.nav.paw.kafkakeygenerator.test.initTestDatabase
 import no.nav.paw.kafkakeygenerator.vo.ArbeidssoekerId
 import no.nav.paw.kafkakeygenerator.vo.Failure
@@ -60,7 +60,7 @@ class KafkaConsumerServiceTest : FreeSpec({
             TestData.getIdentitetsnummerOpphoert(identitetsnummer, arbeidssoekerId)
         )
 
-        kafkaConsumerService.handleRecords(hendelser.asConsumerRecordsSequence())
+        kafkaConsumerService.handleRecords(hendelser.asConsumerRecords())
 
         val keyResult = kafkaKeysRepository.hent(identitetsnummer)
         val auditResult = kafkaKeysAuditRepository.find(identitetsnummer)
@@ -80,7 +80,7 @@ class KafkaConsumerServiceTest : FreeSpec({
         )
 
         shouldThrow<IllegalStateException> {
-            kafkaConsumerService.handleRecords(hendelser.asConsumerRecordsSequence())
+            kafkaConsumerService.handleRecords(hendelser.asConsumerRecords())
         }
 
         val keyResult = kafkaKeysRepository.hent(identitetsnummer)
@@ -110,7 +110,7 @@ class KafkaConsumerServiceTest : FreeSpec({
                     )
                 )
 
-                kafkaConsumerService.handleRecords(hendelser.asConsumerRecordsSequence())
+                kafkaConsumerService.handleRecords(hendelser.asConsumerRecords())
 
                 val keyResult1 = kafkaKeysRepository.hent(identitetsnummer1)
                 val keyResult2 = kafkaKeysRepository.hent(identitetsnummer2)

apps/kafka-key-generator/src/test/kotlin/no/nav/paw/kafkakeygenerator/test/TestData.kt

@@ -143,7 +143,11 @@ fun MockRequestHandleScope.genererResponse(it: HttpRequestData): HttpResponseDat
     )
 }
 
-fun List<Hendelse>.asConsumerRecordsSequence(): Sequence<ConsumerRecords<Long, Hendelse>> =
+fun List<Hendelse>.asConsumerRecords(): ConsumerRecords<Long, Hendelse> =
+    this.map { TestData.getConsumerRecord(nextLong(), it) }
+        .let { TestData.getConsumerRecords(it) }
+
+fun List<Hendelse>.asConsumerSequence(): Sequence<ConsumerRecords<Long, Hendelse>> =
     this.map { TestData.getConsumerRecord(nextLong(), it) }
         .let { TestData.getConsumerRecords(it) }
         .let { sequenceOf(it) }

apps/kafka-key-generator/src/test/kotlin/no/nav/paw/kafkakeygenerator/test/TestDatabase.kt

@@ -1,15 +1,19 @@
 package no.nav.paw.kafkakeygenerator.test
 
 import no.nav.paw.kafkakeygenerator.config.DatabaseConfig
-import no.nav.paw.kafkakeygenerator.database.createDataSource
+import no.nav.paw.kafkakeygenerator.utils.createDataSource
 import org.testcontainers.containers.PostgreSQLContainer
 import org.testcontainers.containers.wait.strategy.Wait
 import javax.sql.DataSource
 
 fun initTestDatabase(): DataSource {
     val config = postgreSQLContainer().let {
         DatabaseConfig(
-            jdbcUrl = "jdbc:postgresql://${it.host}:${it.firstMappedPort}/${it.databaseName}?user=${it.username}&password=${it.password}",
+            host = it.host,
+            port = it.firstMappedPort,
+            database = it.databaseName,
+            username = it.username,
+            password = it.password,
             driverClassName = "org.postgresql.Driver",
             autoCommit = false
         )