-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #16 from navikt/feature/tc-273-adjust-kafka-for-arena
Juster KafkaFactory mot topics fra Arena
- Loading branch information
Showing
15 changed files
with
105 additions
and
187 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,3 +1,3 @@ | ||
| FROM navikt/java:11 | ||
| LABEL org.opencontainers.image.source="https://github.com/navikt/mulighetsrommet" | ||
| COPY /build/libs/no.nav.mulighetsrommet.api.jar app.jar | ||
| COPY /build/libs/no.nav.mulighetsrommet.api-all.jar app.jar |
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
154 changes: 54 additions & 100 deletions
154
backend/src/main/kotlin/no/nav/mulighetsrommet/api/kafka/KafkaFactory.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 |
|---|---|---|
| @@ -1,120 +1,74 @@ | ||
| package no.nav.mulighetsrommet.api.kafka | ||
|
|
||
| import kotlinx.coroutines.delay | ||
| import com.typesafe.config.ConfigFactory | ||
| import io.ktor.config.HoconApplicationConfig | ||
| import no.nav.common.kafka.consumer.KafkaConsumerClient | ||
| import no.nav.common.kafka.consumer.util.KafkaConsumerClientBuilder | ||
| import no.nav.common.kafka.consumer.util.deserializer.Deserializers.stringDeserializer | ||
| import no.nav.common.kafka.util.KafkaPropertiesBuilder | ||
| import no.nav.common.kafka.util.KafkaPropertiesPreset | ||
| import no.nav.mulighetsrommet.api.database.DatabaseFactory | ||
| import no.nav.mulighetsrommet.api.domain.TiltaksgjennomforingTable | ||
| import no.nav.mulighetsrommet.api.domain.TiltaksvariantTable | ||
| import org.apache.kafka.clients.consumer.Consumer | ||
| import org.apache.kafka.clients.consumer.KafkaConsumer | ||
| import org.apache.kafka.common.serialization.StringDeserializer | ||
| import org.apache.kafka.streams.StreamsBuilder | ||
| import org.apache.kafka.streams.Topology | ||
| import org.jetbrains.exposed.dao.id.IntIdTable | ||
| import org.jetbrains.exposed.sql.insertAndGetId | ||
| import org.jetbrains.exposed.sql.selectAll | ||
| import java.time.Duration | ||
| import java.time.LocalDateTime | ||
| import java.util.UUID | ||
| import kotlin.random.Random | ||
| import org.apache.kafka.clients.consumer.ConsumerRecord | ||
| import org.apache.kafka.common.serialization.ByteArrayDeserializer | ||
| import org.slf4j.LoggerFactory | ||
| import java.util.Properties | ||
| import java.util.function.Consumer | ||
|
|
||
| class KafkaFactory(private val db: DatabaseFactory) { | ||
|
|
||
| private val streamsConfiguration = KafkaStreamConfig() | ||
| // private val kafkaStreams: KafkaStreams | ||
| // private val topology: Topology | ||
| // private val adminClient: AdminClient | ||
| private val logger = LoggerFactory.getLogger(KafkaFactory::class.java) | ||
| private val appConfig = HoconApplicationConfig(ConfigFactory.load()) | ||
| private val consumerClient: KafkaConsumerClient | ||
|
|
||
| init { | ||
| // topology = buildStream() | ||
| // kafkaStreams = KafkaStreams(topology, streamsConfiguration) | ||
| // adminClient = AdminClient.create(streamsConfiguration) | ||
| // kafkaStreams.cleanUp() | ||
| // kafkaStreams.start() | ||
| } | ||
| logger.debug("Initializing KafkaFactory.") | ||
|
|
||
| private fun buildStream(): Topology { | ||
| val builder = StreamsBuilder() | ||
| builder.stream<String, String>(KafkaTopics.Tiltaksgjennomforing.topic) | ||
| return builder.build() | ||
| } | ||
| val consumerProperties = configureProperties() | ||
| val topics = configureTopics() | ||
|
|
||
| // fun shutdown() { | ||
| // kafkaStreams.close() | ||
| // } | ||
| // | ||
| // fun isAlive(): Boolean { | ||
| // return kafkaStreams.state().isRunningOrRebalancing | ||
| // } | ||
| consumerClient = KafkaConsumerClientBuilder.builder() | ||
| .withProperties(consumerProperties) | ||
| .withTopicConfigs(topics) | ||
| .build() | ||
|
|
||
| private fun createConsumer(): Consumer<String, String> { | ||
| val props = streamsConfiguration | ||
| props["key.deserializer"] = StringDeserializer::class.java | ||
| props["value.deserializer"] = StringDeserializer::class.java | ||
| return KafkaConsumer(props) | ||
| } | ||
| consumerClient.start() | ||
|
|
||
| fun consumeArenaEvents() { | ||
| val consumer = createConsumer() | ||
| consumer.subscribe(listOf(KafkaTopics.Tiltaksgjennomforing.topic)) | ||
| while (true) { | ||
| val records = consumer.poll(Duration.ofSeconds(1)) | ||
| if (!records.isEmpty) { | ||
| println("Consumed ${records.count()} records") | ||
| records.iterator().forEach { | ||
| val message = it.value() | ||
| println("Message: $message") | ||
| } | ||
| } | ||
| } | ||
| logger.debug("Consumer client started. Done with initializing KafkaFactory.") | ||
| } | ||
|
|
||
| // Denne er kun for å ha en måte å simulere at events kommer inn fra Arena via Kafka. | ||
| // TODO: Fjern denne når bestilling av Arena er på plass. | ||
| suspend fun consumeTiltaksgjennomforingEventsFromArena() { | ||
| delay(Duration.ofMinutes(2).toMillis()) | ||
| while (true) { | ||
| val uuid = UUID.randomUUID() | ||
| val tiltaksnr = Random.nextInt(0, 999999) | ||
|
|
||
| val arenaEvent = ArenaEvent( | ||
| "Tiltaksgjennomføring ($uuid)", | ||
| "Beskrivelse", | ||
| tiltaksnr, | ||
| LocalDateTime.now(), | ||
| LocalDateTime.now().plusYears(2) | ||
| ) | ||
| fun stopClient() { | ||
| consumerClient.stop() | ||
| } | ||
|
|
||
| val tiltaksgjennomforingId = db.dbQuery { | ||
| TiltaksgjennomforingTable.insertAndGetId { | ||
| it[tittel] = arenaEvent.tittel | ||
| it[tiltaksvariantId] = getRandomId(TiltaksvariantTable) | ||
| it[tiltaksnummer] = arenaEvent.tiltaksnummer | ||
| it[beskrivelse] = arenaEvent.beskrivelse | ||
| it[fraDato] = arenaEvent.fraDato | ||
| it[tilDato] = arenaEvent.tilDato | ||
| } | ||
| } | ||
| println("Opprettet tiltaksgjennomforing med id $tiltaksgjennomforingId") | ||
| delay(Duration.ofHours(2).toMillis()) | ||
| private fun configureProperties(): Properties { | ||
| val consumerGroupId = "mulighetsrommet-api-consumer.v1" | ||
| return if (appConfig.property("ktor.localDevelopment").getString() == "true") { | ||
| KafkaPropertiesBuilder.consumerBuilder() | ||
| .withBrokerUrl("localhost:9092") | ||
| .withBaseProperties() | ||
| .withConsumerGroupId(consumerGroupId) | ||
| .withDeserializers(ByteArrayDeserializer::class.java, ByteArrayDeserializer::class.java) | ||
| .build() | ||
| } else { | ||
| KafkaPropertiesPreset.aivenDefaultConsumerProperties(consumerGroupId) | ||
| } | ||
| } | ||
|
|
||
| data class ArenaEvent( | ||
| val tittel: String, | ||
| val beskrivelse: String, | ||
| val tiltaksnummer: Int, // Ikke unikt, kan kolidere | ||
| val fraDato: LocalDateTime, | ||
| val tilDato: LocalDateTime | ||
| ) | ||
| } | ||
| private fun configureTopics(): List<KafkaConsumerClientBuilder.TopicConfig<String, String>> { | ||
| return KafkaTopics.values().map { it -> | ||
| KafkaConsumerClientBuilder.TopicConfig<String, String>() | ||
| .withLogging() | ||
| .withConsumerConfig( | ||
| it.topic, | ||
| stringDeserializer(), | ||
| stringDeserializer(), | ||
| Consumer<ConsumerRecord<String, String>> { logTopicContent(it) } | ||
| ) | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Only for testing | ||
| */ | ||
| fun <T : IntIdTable> getRandomId(table: T): Int { | ||
| return table | ||
| .slice(table.id) | ||
| .selectAll() | ||
| .map { it[table.id].value } | ||
| .random() | ||
| // Temporary print out until we actually implement something with the events. | ||
| private fun logTopicContent(consumerRecord: ConsumerRecord<String, String>) { | ||
| logger.debug("Topic: ${consumerRecord.topic()} - Value: ${consumerRecord.value()}") | ||
| } | ||
| } |
22 changes: 0 additions & 22 deletions
22
backend/src/main/kotlin/no/nav/mulighetsrommet/api/kafka/KafkaStreamConfig.kt
This file was deleted.
Oops, something went wrong.
Oops, something went wrong.