Skip to content

Commit 2ba0020

Browse files
geiralundgeiralund
committed
Bli kvitt avhengighet til dagpenger-streams bibliotek.
1 parent c23df5b commit 2ba0020

File tree

4 files changed

+85
-27
lines changed

4 files changed

+85
-27
lines changed

‎dp-inntekt-api/build.gradle.kts‎

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -28,9 +28,6 @@ val log4j2Version = "2.24.3"
2828
dependencies {
2929

3030
implementation(project(":dp-inntekt-kontrakter"))
31-
32-
implementation("com.github.navikt:dagpenger-streams:20230831.f3d785")
33-
3431
implementation(libs.bundles.ktor.server)
3532
implementation("io.ktor:ktor-server-netty:${libs.versions.ktor.get()}")
3633
implementation("io.ktor:ktor-server-default-headers:${libs.versions.ktor.get()}")

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

Lines changed: 14 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -11,10 +11,9 @@ import com.natpryce.konfig.stringType
1111
import kotlinx.coroutines.runBlocking
1212
import no.nav.dagpenger.oauth2.CachedOauth2Client
1313
import no.nav.dagpenger.oauth2.OAuth2Config
14-
import no.nav.dagpenger.streams.Credential
15-
import no.nav.dagpenger.streams.KafkaAivenCredentials
14+
import org.apache.kafka.common.security.auth.SecurityProtocol
1615

17-
internal object Config {
16+
object Config {
1817
private val localProperties =
1918
ConfigurationMap(
2019
mapOf(
@@ -105,6 +104,17 @@ internal object Config {
105104
azureAdTokenSupplier(config[Key("pdl.api.scope", stringType)])
106105
}
107106

107+
data class KafkaAivenCredentials(
108+
val securityProtocolConfig: String = SecurityProtocol.SSL.name,
109+
val sslEndpointIdentificationAlgorithmConfig: String = "",
110+
val sslTruststoreTypeConfig: String = "jks",
111+
val sslKeystoreTypeConfig: String = "PKCS12",
112+
val sslTruststoreLocationConfig: String = "/var/run/secrets/nais.io/kafka/client.truststore.jks",
113+
val sslTruststorePasswordConfig: String = config[Key("KAFKA_CREDSTORE_PASSWORD", stringType)],
114+
val sslKeystoreLocationConfig: String = "/var/run/secrets/nais.io/kafka/client.keystore.p12",
115+
val sslKeystorePasswordConfig: String = sslTruststorePasswordConfig,
116+
)
117+
108118
val inntektsKomponentTokenProvider by lazy {
109119
azureAdTokenSupplier(config[Key("inntektskomponenten.api.scope", stringType)])
110120
}
@@ -135,7 +145,7 @@ data class InntektApiConfig(
135145
val id: String,
136146
val brokers: String,
137147
val profile: Profile,
138-
val credential: Credential?,
148+
val credential: Config.KafkaAivenCredentials?,
139149
val httpPort: Int,
140150
val hentinntektListeUrl: String,
141151
val enhetsregisteretUrl: String,

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

Lines changed: 61 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -6,20 +6,23 @@ import kotlinx.coroutines.Job
66
import kotlinx.coroutines.SupervisorJob
77
import kotlinx.coroutines.launch
88
import mu.KotlinLogging
9-
import no.nav.dagpenger.events.Packet
9+
import no.nav.dagpenger.inntekt.Config
1010
import no.nav.dagpenger.inntekt.HealthCheck
1111
import no.nav.dagpenger.inntekt.HealthStatus
1212
import no.nav.dagpenger.inntekt.InntektApiConfig
1313
import no.nav.dagpenger.inntekt.db.InntektId
1414
import no.nav.dagpenger.inntekt.db.InntektStore
15-
import no.nav.dagpenger.plain.consumerConfig
16-
import no.nav.dagpenger.streams.PacketDeserializer
15+
import no.nav.dagpenger.inntekt.serder.jacksonObjectMapper
16+
import org.apache.kafka.clients.CommonClientConfigs
1717
import org.apache.kafka.clients.consumer.CommitFailedException
1818
import org.apache.kafka.clients.consumer.ConsumerConfig
1919
import org.apache.kafka.clients.consumer.KafkaConsumer
20+
import org.apache.kafka.common.config.SslConfigs
21+
import org.apache.kafka.common.serialization.StringDeserializer
2022
import java.time.Duration
2123
import java.time.ZoneOffset
2224
import java.time.ZonedDateTime
25+
import java.util.Properties
2326
import kotlin.coroutines.CoroutineContext
2427

2528
internal class KafkaSubsumsjonBruktDataConsumer(
@@ -43,13 +46,12 @@ internal class KafkaSubsumsjonBruktDataConsumer(
4346
launch(coroutineContext) {
4447
logger.info { "Starting ${config.application.id}" }
4548

46-
KafkaConsumer<String, Packet>(
49+
KafkaConsumer<String, String>(
4750
consumerConfig(
4851
groupId = config.application.id,
4952
bootstrapServerUrl = config.application.brokers,
5053
credential = config.application.credential,
5154
).also {
52-
it[ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG] = PacketDeserializer::class.java
5355
it[ConsumerConfig.AUTO_OFFSET_RESET_CONFIG] = "earliest"
5456
it[ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG] = "false"
5557
it[ConsumerConfig.MAX_POLL_RECORDS_CONFIG] = 10
@@ -62,16 +64,15 @@ internal class KafkaSubsumsjonBruktDataConsumer(
6264
val ids =
6365
records.asSequence()
6466
.map { record -> record.value() }
67+
.map { jacksonObjectMapper.readTree(it) }
6568
.filter { packet ->
66-
packet.hasFields(
67-
"@event_name",
68-
"aktorId",
69-
"inntektsId",
70-
"kontekst",
71-
) &&
72-
packet.getStringValue("@event_name") == "brukt_inntekt"
69+
packet.has("@event_name") &&
70+
packet.has("aktorId") &&
71+
packet.has("inntektsId") &&
72+
packet.has("kontekst") &&
73+
packet.get("@event_name").asText() == "brukt_inntekt"
7374
}
74-
.map { packet -> InntektId(packet.getStringValue("inntektsId")) }
75+
.map { packet -> InntektId(packet.get("inntektsId").asText()) }
7576
.toList()
7677

7778
try {
@@ -126,4 +127,51 @@ internal class KafkaSubsumsjonBruktDataConsumer(
126127

127128
fun expired() = ZonedDateTime.now(ZoneOffset.UTC).isAfter(expires)
128129
}
130+
131+
companion object {
132+
private val defaultConsumerConfig =
133+
Properties().apply {
134+
put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer::class.java.name)
135+
put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer::class.java.name)
136+
}
137+
138+
internal fun commonConfig(
139+
bootstrapServers: String,
140+
credential: Config.KafkaAivenCredentials? = null,
141+
): Properties {
142+
return Properties().apply {
143+
put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers)
144+
credential?.let { creds ->
145+
putAll(
146+
Properties().apply {
147+
put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, creds.securityProtocolConfig)
148+
put(
149+
SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG,
150+
creds.sslEndpointIdentificationAlgorithmConfig,
151+
)
152+
put(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, creds.sslTruststoreTypeConfig)
153+
put(SslConfigs.SSL_KEYSTORE_TYPE_CONFIG, creds.sslKeystoreTypeConfig)
154+
put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, creds.sslTruststoreLocationConfig)
155+
put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, creds.sslTruststorePasswordConfig)
156+
put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, creds.sslKeystoreLocationConfig)
157+
put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, creds.sslKeystorePasswordConfig)
158+
},
159+
)
160+
}
161+
}
162+
}
163+
}
164+
165+
private fun consumerConfig(
166+
groupId: String,
167+
bootstrapServerUrl: String,
168+
credential: Config.KafkaAivenCredentials? = null,
169+
properties: Properties = defaultConsumerConfig,
170+
): Properties {
171+
return Properties().apply {
172+
putAll(properties)
173+
putAll(commonConfig(bootstrapServerUrl, credential))
174+
put(ConsumerConfig.GROUP_ID_CONFIG, groupId)
175+
}
176+
}
129177
}

‎dp-inntekt-api/src/test/kotlin/no/nav/dagpenger/inntekt/subsumsjonbrukt/KafkaSubsumsjonBruktDataConsumerTest.kt‎

Lines changed: 10 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -13,27 +13,29 @@ import no.nav.dagpenger.inntekt.HealthStatus
1313
import no.nav.dagpenger.inntekt.db.InntektId
1414
import no.nav.dagpenger.inntekt.db.InntektStore
1515
import no.nav.dagpenger.inntekt.serder.jacksonObjectMapper
16-
import no.nav.dagpenger.plain.producerConfig
16+
import no.nav.dagpenger.inntekt.subsumsjonbrukt.KafkaSubsumsjonBruktDataConsumer.Companion.commonConfig
1717
import org.apache.kafka.clients.producer.KafkaProducer
1818
import org.apache.kafka.clients.producer.ProducerConfig
1919
import org.apache.kafka.clients.producer.ProducerRecord
2020
import org.apache.kafka.common.serialization.StringSerializer
21+
import org.apache.logging.log4j.CloseableThreadContext.putAll
2122
import org.junit.jupiter.api.Test
22-
import org.testcontainers.containers.KafkaContainer
2323
import org.testcontainers.containers.wait.strategy.HostPortWaitStrategy
24+
import org.testcontainers.kafka.ConfluentKafkaContainer
2425
import org.testcontainers.utility.DockerImageName
2526
import java.sql.SQLTransientConnectionException
2627
import java.time.Duration
2728
import java.time.ZoneOffset
2829
import java.time.ZonedDateTime
30+
import java.util.Properties
2931
import java.util.concurrent.TimeUnit
3032

3133
private val LOGGER = KotlinLogging.logger { }
3234

3335
internal class KafkaSubsumsjonBruktDataConsumerTest {
3436
private object Kafka {
3537
val instance by lazy {
36-
KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka").withTag("7.5.1")).apply {
38+
ConfluentKafkaContainer(DockerImageName.parse("confluentinc/cp-kafka").withTag("7.5.1")).apply {
3739
this.waitingFor(HostPortWaitStrategy())
3840
this.start()
3941
}
@@ -64,11 +66,12 @@ internal class KafkaSubsumsjonBruktDataConsumerTest {
6466
)
6567
private val producer by lazy {
6668
KafkaProducer<String, String>(
67-
producerConfig(
68-
clientId = "test",
69-
bootstrapServers = Kafka.instance.bootstrapServers,
70-
).also {
69+
Properties().apply {
70+
putAll(commonConfig(Kafka.instance.bootstrapServers, null))
71+
put(ProducerConfig.CLIENT_ID_CONFIG, "test")
72+
}.also {
7173
it[ProducerConfig.ACKS_CONFIG] = "all"
74+
it[ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG] = StringSerializer::class.java.name
7275
it[ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG] = StringSerializer::class.java.name
7376
},
7477
)

0 commit comments

Comments
 (0)