Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Use unique consumer groups in KafkaDslKotlinTests

Browse Source
This commit is contained in:
Artem Bilan
2022-10-18 13:20:38 -04:00
parent 0951738ab3
commit beef2e68c7
1 changed files with 10 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/kotlin/KafkaDslKotlinTests.kt
View File

@@ -228,6 +228,8 @@ class KafkaDslKotlinTests {
.configureListenerContainer {
it.ackMode(ContainerProperties.AckMode.MANUAL)
.id("topic1ListenerContainer")
.groupId("kotlin_topic1_group")
}
.errorChannel(errorChannel())
.retryTemplate(RetryTemplate())
@@ -248,7 +250,10 @@ class KafkaDslKotlinTests {
consumerFactory(),
KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC2
)
.configureListenerContainer { it.ackMode(ContainerProperties.AckMode.MANUAL) }
.configureListenerContainer {
it.ackMode(ContainerProperties.AckMode.MANUAL)
.groupId("kotlin_topic2_group")
}
.errorChannel(errorChannel())
.retryTemplate(RetryTemplate())
.filterInRetry(true)) {
@@ -302,7 +307,10 @@ class KafkaDslKotlinTests {
@Bean
fun sourceFlow() =
integrationFlow(Kafka.inboundChannelAdapter(consumerFactory(), ConsumerProperties(TEST_TOPIC3)),
integrationFlow(Kafka.inboundChannelAdapter(consumerFactory(),
ConsumerProperties(TEST_TOPIC3).also {
it.setGroupId("kotlinMessageSourceGroup")
}),
{ poller(Pollers.fixedDelay(100)) }) {
handle { m ->
this@ContextConfiguration.fromSource = m.payload