Skip to content

Commit ce10dc1

Browse files
vitaxavitaxa
authored
TD-396: aws kafka config (#4)
1 parent 7c6f92c commit ce10dc1

File tree

13 files changed

+111
-205
lines changed

13 files changed

+111
-205
lines changed

‎.codecov.yml‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,3 +4,7 @@ coverage:
44
default:
55
target: auto
66
threshold: 5%
7+
patch:
8+
default:
9+
enabled: no
10+
if_not_found: success

‎.editorconfig‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,2 @@
1+
[*.{kt,kts}]
2+
disabled_rules=no-wildcard-imports

‎pom.xml‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
<parent>
77
<groupId>dev.vality</groupId>
88
<artifactId>service-parent-pom</artifactId>
9-
<version>2.0.0-BETA-8</version>
9+
<version>2.0.0-BETA-11</version>
1010
</parent>
1111

1212
<artifactId>vortigon</artifactId>
@@ -164,6 +164,10 @@
164164
<groupId>io.micrometer</groupId>
165165
<artifactId>micrometer-registry-prometheus</artifactId>
166166
</dependency>
167+
<dependency>
168+
<groupId>software.amazon.msk</groupId>
169+
<artifactId>aws-msk-iam-auth</artifactId>
170+
</dependency>
167171

168172
<!--test-->
169173
<dependency>

‎src/main/kotlin/dev/vality/vortigon/config/KafkaConfig.kt‎

Lines changed: 28 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -1,40 +1,47 @@
11
package dev.vality.vortigon.config
22

3-
import com.rbkmoney.kafka.common.exception.handler.SeekToCurrentWithSleepBatchErrorHandler
43
import dev.vality.machinegun.eventsink.MachineEvent
54
import dev.vality.mg.event.sink.service.ConsumerGroupIdService
6-
import dev.vality.vortigon.config.properties.KafkaProperties
75
import dev.vality.vortigon.serializer.MachineEventDeserializer
86
import lombok.RequiredArgsConstructor
9-
import mu.KotlinLogging
107
import org.apache.kafka.clients.consumer.ConsumerConfig
118
import org.apache.kafka.clients.consumer.OffsetResetStrategy
129
import org.apache.kafka.common.serialization.Deserializer
1310
import org.apache.kafka.common.serialization.StringDeserializer
11+
import org.springframework.beans.factory.annotation.Value
12+
import org.springframework.boot.autoconfigure.kafka.KafkaProperties
1413
import org.springframework.context.annotation.Bean
1514
import org.springframework.context.annotation.Configuration
1615
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory
1716
import org.springframework.kafka.core.DefaultKafkaConsumerFactory
1817
import org.springframework.kafka.listener.ContainerProperties
19-
import java.io.File
20-
21-
private val log = KotlinLogging.logger {}
18+
import org.springframework.kafka.listener.DefaultErrorHandler
19+
import org.springframework.util.backoff.ExponentialBackOff
2220

2321
@Configuration
2422
@RequiredArgsConstructor
2523
class KafkaConfig(
2624
private val kafkaProperties: KafkaProperties,
2725
private val consumerGroupIdService: ConsumerGroupIdService,
26+
@Value("\${kafka.consumer.concurrency}")
27+
private val concurrencyListenerCount: Int,
28+
@Value("\${kafka.max.poll.records}")
29+
private val maxPollRecords: String,
30+
@Value("\${kafka.error-handler.backoff.initial-interval}")
31+
private val initialInterval: Long,
32+
@Value("\${kafka.error-handler.backoff.max-interval}")
33+
private val maxInterval: Long
2834
) {
35+
2936
@Bean
3037
fun partyListenerContainerFactory(): ConcurrentKafkaListenerContainerFactory<String, MachineEvent> {
3138
val factory = ConcurrentKafkaListenerContainerFactory<String, MachineEvent>()
3239
val consumerGroup: String = consumerGroupIdService.generateGroupId(PARTY_CONSUMER_GROUP_NAME)
33-
initDefaultListenerProperties<MachineEvent>(
40+
initDefaultListenerProperties(
3441
factory,
3542
consumerGroup,
3643
MachineEventDeserializer(),
37-
kafkaProperties.maxPollRecords
44+
maxPollRecords
3845
)
3946
return factory
4047
}
@@ -49,8 +56,12 @@ class KafkaConfig(
4956
consumerGroup, deserializer, maxPollRecords
5057
)
5158
factory.consumerFactory = consumerFactory
52-
factory.setConcurrency(kafkaProperties.consumer.concurrency.toInt())
53-
factory.setBatchErrorHandler(SeekToCurrentWithSleepBatchErrorHandler())
59+
factory.setConcurrency(concurrencyListenerCount)
60+
val exponentialBackOff = ExponentialBackOff().apply {
61+
maxInterval = this@KafkaConfig.maxInterval
62+
initialInterval = this@KafkaConfig.initialInterval
63+
}
64+
factory.setCommonErrorHandler(DefaultErrorHandler(exponentialBackOff))
5465
factory.isBatchListener = true
5566
factory.containerProperties.ackMode = ContainerProperties.AckMode.MANUAL
5667
}
@@ -60,7 +71,7 @@ class KafkaConfig(
6071
deserializer: Deserializer<T>,
6172
maxPollRecords: String,
6273
): DefaultKafkaConsumerFactory<String, T> {
63-
val props: MutableMap<String, Any> = createDefaultProperties(consumerGroup)
74+
val props: MutableMap<String, Any> = createDefaultConsumerProperties(consumerGroup)
6475
props[ConsumerConfig.MAX_POLL_RECORDS_CONFIG] = maxPollRecords
6576
return DefaultKafkaConsumerFactory(
6677
props,
@@ -69,33 +80,13 @@ class KafkaConfig(
6980
)
7081
}
7182

72-
private fun createDefaultProperties(value: String): MutableMap<String, Any> {
73-
return HashMap<String, Any>().apply {
74-
put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.bootstrapServers)
75-
put(ConsumerConfig.GROUP_ID_CONFIG, value)
76-
put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, OffsetResetStrategy.EARLIEST.name.lowercase())
77-
put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false)
78-
put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, kafkaProperties.maxSessionTimeoutMs)
79-
put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, kafkaProperties.maxPollIntervalMs)
80-
putAll(createSslConfig())
81-
}
82-
}
83+
private fun createDefaultConsumerProperties(value: String): MutableMap<String, Any> {
84+
val properties = kafkaProperties.buildConsumerProperties()
85+
properties[ConsumerConfig.GROUP_ID_CONFIG] = value
86+
properties[ConsumerConfig.AUTO_OFFSET_RESET_CONFIG] = OffsetResetStrategy.EARLIEST.name.lowercase()
87+
properties[ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG] = false
8388

84-
private fun createSslConfig(): Map<String, Any> {
85-
val kafkaSslProperties = kafkaProperties.ssl
86-
log.info("Kafka ssl isEnabled: {}", kafkaSslProperties.enabled)
87-
val configProps = HashMap<String, Any>()
88-
if (kafkaSslProperties.enabled) {
89-
configProps["security.protocol"] = "SSL"
90-
configProps["ssl.truststore.location"] = File(kafkaSslProperties.trustStoreLocation).absolutePath
91-
configProps["ssl.truststore.password"] = kafkaSslProperties.trustStorePassword
92-
configProps["ssl.keystore.type"] = "PKCS12"
93-
configProps["ssl.truststore.type"] = "PKCS12"
94-
configProps["ssl.keystore.location"] = File(kafkaSslProperties.keyStoreLocation).absolutePath
95-
configProps["ssl.keystore.password"] = kafkaSslProperties.keyStorePassword
96-
configProps["ssl.key.password"] = kafkaSslProperties.keyPassword
97-
}
98-
return configProps
89+
return properties
9990
}
10091

10192
companion object {

‎src/main/kotlin/dev/vality/vortigon/config/properties/KafkaProperties.kt‎

Lines changed: 0 additions & 58 deletions
This file was deleted.

‎src/main/kotlin/dev/vality/vortigon/handler/party/PartyEventHandleManager.kt‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -4,9 +4,9 @@ import dev.vality.damsel.payment_processing.PartyChange
44
import dev.vality.damsel.payment_processing.PartyEventData
55
import dev.vality.machinegun.eventsink.MachineEvent
66
import dev.vality.sink.common.parser.impl.MachineEventParser
7-
import dev.vality.vortigon.config.properties.KafkaProperties
87
import dev.vality.vortigon.handler.ChangeHandler
98
import mu.KotlinLogging
9+
import org.springframework.beans.factory.annotation.Value
1010
import org.springframework.kafka.support.Acknowledgment
1111
import org.springframework.stereotype.Component
1212
import org.springframework.transaction.annotation.Transactional
@@ -17,7 +17,8 @@ private val log = KotlinLogging.logger {}
1717
class PartyEventHandleManager(
1818
private val eventParser: MachineEventParser<PartyEventData>,
1919
private val partyHandlers: List<ChangeHandler<PartyChange, MachineEvent>>,
20-
private val kafkaProperties: KafkaProperties,
20+
@Value("\${kafka.consumer.throttling-timeout-ms}")
21+
private val throttlingTimeoutMs: Int,
2122
) {
2223

2324
@Transactional
@@ -31,7 +32,7 @@ class PartyEventHandleManager(
3132
ack.acknowledge()
3233
} catch (e: Exception) {
3334
log.error(e) { "Exception during PartyListener process" }
34-
Thread.sleep(kafkaProperties.consumer.throttlingTimeoutMs.toLong())
35+
Thread.sleep(throttlingTimeoutMs.toLong())
3536
throw e
3637
}
3738
}

‎src/main/resources/application.yml‎

Lines changed: 11 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -44,26 +44,19 @@ spring:
4444
schemas: ps
4545
url: '@db.url@'
4646
table: flyway_schema_history
47+
kafka:
48+
bootstrap-servers: localhost:9092
49+
consumer:
50+
properties:
51+
session.timeout.ms: 60000
52+
max.poll.interval.ms: 60000
4753

4854
info:
4955
version: '@project.version@'
5056
stage: dev
5157

5258
kafka:
53-
bootstrap-servers: kenny-kafka1.bst1.rbkmoney.net:9092
54-
retry-attempts: 3
55-
max-poll-records: 100
56-
max-poll-interval-ms: 60000
57-
max-session-timeout-ms: 6000
58-
ssl:
59-
enabled: false
60-
trust-store-location: "test"
61-
trust-store-password: "test"
62-
key-store-location: "test"
63-
key-store-password: "test"
64-
key-password: "test"
65-
key-store-type: PKCS12
66-
trust-store-type: PKCS12
59+
max.poll.records: 100
6760
consumer:
6861
enabled: true
6962
prefix: vortigon
@@ -72,7 +65,10 @@ kafka:
7265
topic:
7366
party:
7467
initial: mg-events-party
75-
max-poll-records: 50
68+
error-handler:
69+
backoff:
70+
initial-interval: 1000
71+
max-interval: 600000
7672

7773
repository:
7874
url: http://dominant:8022/v1/domain/repository_client

‎src/test/kotlin/dev/vality/vortigon/AbstractKafkaIntegrationTest.kt‎ renamed to ‎src/test/kotlin/dev/vality/vortigon/AbstractIntegrationTest.kt‎

Lines changed: 3 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -22,16 +22,15 @@ import org.springframework.context.ApplicationContextInitializer
2222
import org.springframework.context.ConfigurableApplicationContext
2323
import org.springframework.test.annotation.DirtiesContext
2424
import org.springframework.test.context.ContextConfiguration
25-
import org.testcontainers.containers.KafkaContainer
2625
import java.time.Duration
2726
import java.time.Instant
28-
import java.util.Properties
27+
import java.util.*
2928

3029
private val log = KotlinLogging.logger {}
3130

3231
@DirtiesContext
33-
@ContextConfiguration(initializers = [AbstractKafkaIntegrationTest.Initializer::class])
34-
abstract class AbstractKafkaIntegrationTest : PostgresAbstractTest() {
32+
@ContextConfiguration(initializers = [AbstractIntegrationTest.Initializer::class])
33+
abstract class AbstractIntegrationTest : ContainerConfiguration() {
3534

3635
protected fun createMachineEvent(partyChange: PartyChange, sourceId: String, sequenceId: Long): MachineEvent {
3736
val message = MachineEvent()
@@ -102,11 +101,4 @@ abstract class AbstractKafkaIntegrationTest : PostgresAbstractTest() {
102101
return KafkaConsumer<String, T>(props)
103102
}
104103
}
105-
106-
companion object {
107-
private val kafka: KafkaContainer by lazy {
108-
KafkaContainer(KAFKA_DOCKER_VERSION).withEmbeddedZookeeper()
109-
}
110-
private const val KAFKA_DOCKER_VERSION = "5.0.1"
111-
}
112104
}

‎src/test/kotlin/dev/vality/vortigon/ContainerConfiguration.kt‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,45 @@
1+
package dev.vality.vortigon
2+
3+
import org.junit.jupiter.api.BeforeAll
4+
import org.springframework.boot.test.context.SpringBootTest
5+
import org.springframework.test.context.DynamicPropertyRegistry
6+
import org.springframework.test.context.DynamicPropertySource
7+
import org.testcontainers.containers.KafkaContainer
8+
import org.testcontainers.containers.PostgreSQLContainer
9+
import org.testcontainers.utility.DockerImageName
10+
11+
@SpringBootTest
12+
class ContainerConfiguration {
13+
companion object {
14+
@BeforeAll
15+
@JvmStatic
16+
fun beforeAll() {
17+
postgresql.start()
18+
kafka.start()
19+
}
20+
21+
@JvmStatic
22+
val postgresql: PostgreSQLContainer<Nothing> = PostgreSQLContainer<Nothing>("postgres:14-alpine").apply {
23+
withDatabaseName("postgresql")
24+
withUsername("user")
25+
withPassword("password")
26+
}
27+
28+
@JvmStatic
29+
val kafka: KafkaContainer = KafkaContainer(
30+
DockerImageName.parse("confluentinc/cp-kafka").withTag("7.1.1")
31+
).withEmbeddedZookeeper()
32+
33+
@JvmStatic
34+
@DynamicPropertySource
35+
fun properties(registry: DynamicPropertyRegistry) {
36+
registry.add("spring.kafka.bootstrap-servers") { kafka.bootstrapServers }
37+
registry.add("spring.datasource.url", postgresql::getJdbcUrl)
38+
registry.add("spring.datasource.password", postgresql::getPassword)
39+
registry.add("spring.datasource.username", postgresql::getUsername)
40+
registry.add("spring.flyway.url", postgresql::getJdbcUrl)
41+
registry.add("spring.flyway.user", postgresql::getUsername)
42+
registry.add("spring.flyway.password", postgresql::getPassword)
43+
}
44+
}
45+
}

0 commit comments

Comments
 (0)