Legg på mer logging og try/catch

Author: JMLindseth

dp-inntekt-api/src/main/kotlin/no/nav/dagpenger/inntekt/Application.kt

@@ -49,6 +49,7 @@ fun main() {
         val subsumsjonBruktDataConsumer =
             KafkaSubsumsjonBruktDataConsumer(config, postgresInntektStore)
                 .apply {
+                    LOGGER.info("Prøver å starte lytting på kafka")
                     listen()
                 }.also {
                     Runtime.getRuntime().addShutdownHook(

dp-inntekt-api/src/main/kotlin/no/nav/dagpenger/inntekt/subsumsjonbrukt/KafkaSubsumsjonBruktDataConsumer.kt

@@ -47,58 +47,63 @@ internal class KafkaSubsumsjonBruktDataConsumer(
         launch(coroutineContext) {
             logger.info { "Starting ${config.application.id}" }
 
-            KafkaConsumer<String, String>(
-                consumerConfig(
-                    groupId = config.application.id,
-                    bootstrapServerUrl = config.application.brokers,
-                    credential = config.application.credential,
-                ).also {
-                    it[ConsumerConfig.AUTO_OFFSET_RESET_CONFIG] = "earliest"
-                    it[ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG] = "false"
-                    it[ConsumerConfig.MAX_POLL_RECORDS_CONFIG] = 10
-                },
-            ).use { consumer ->
-                try {
-                    consumer.subscribe(listOf(config.inntektBruktDataTopic))
-                    while (job.isActive) {
-                        val records = consumer.poll(Duration.ofMillis(100))
-                        val ids =
-                            records
-                                .asSequence()
-                                .map { record -> record.value() }
-                                .map { jacksonObjectMapper.readTree(it) }
-                                .filter { packet ->
-                                    packet.has("@event_name") &&
-                                        packet.has("aktorId") &&
-                                        packet.has("inntektsId") &&
-                                        packet.has("kontekst") &&
-                                        packet.get("@event_name").asText() == "brukt_inntekt"
-                                }.map { packet -> InntektId(packet.get("inntektsId").asText()) }
-                                .toList()
+            try {
+                KafkaConsumer<String, String>(
+                    consumerConfig(
+                        groupId = config.application.id,
+                        bootstrapServerUrl = config.application.brokers,
+                        credential = config.application.credential,
+                    ).also {
+                        it[ConsumerConfig.AUTO_OFFSET_RESET_CONFIG] = "earliest"
+                        it[ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG] = "false"
+                        it[ConsumerConfig.MAX_POLL_RECORDS_CONFIG] = 10
+                    },
+                ).use { consumer ->
+                    try {
+                        consumer.subscribe(listOf(config.inntektBruktDataTopic))
+                        while (job.isActive) {
+                            val records = consumer.poll(Duration.ofMillis(100))
+                            val ids =
+                                records
+                                    .asSequence()
+                                    .map { record -> record.value() }
+                                    .map { jacksonObjectMapper.readTree(it) }
+                                    .filter { packet ->
+                                        packet.has("@event_name") &&
+                                                packet.has("aktorId") &&
+                                                packet.has("inntektsId") &&
+                                                packet.has("kontekst") &&
+                                                packet.get("@event_name").asText() == "brukt_inntekt"
+                                    }.map { packet -> InntektId(packet.get("inntektsId").asText()) }
+                                    .toList()
 
-                        try {
-                            ids.forEach { id ->
-                                if (inntektStore.markerInntektBrukt(id) == 1) {
-                                    logger.info("Marked inntekt with id $id as used")
+                            try {
+                                ids.forEach { id ->
+                                    if (inntektStore.markerInntektBrukt(id) == 1) {
+                                        logger.info("Marked inntekt with id $id as used")
+                                    }
                                 }
+                                if (ids.isNotEmpty()) {
+                                    consumer.commitSync()
+                                }
+                            } catch (e: CommitFailedException) {
+                                logger.warn("Kafka threw a commit fail exception, looping back", e)
                             }
-                            if (ids.isNotEmpty()) {
-                                consumer.commitSync()
-                            }
-                        } catch (e: CommitFailedException) {
-                            logger.warn("Kafka threw a commit fail exception, looping back", e)
                         }
-                    }
-                } catch (e: Exception) {
-                    logger.error(
-                        """
+                    } catch (e: Exception) {
+                        logger.error(
+                            """
                         "Unexpected exception while consuming messages. 
                          Stopping consumer, grace period ${grace.duration.seconds / 60} minutes"
                         """.trimIndent(),
-                        e,
-                    )
-                    stop()
+                            e,
+                        )
+                        stop()
+                    }
                 }
+            } catch (e: Exception) {
+                logger.error("Fikk en feil ved oppsett av KafkaConsumer", e)
+                stop()
             }
         }
     }