From beef2e68c7e389abdfa9ff5b473d3abc084a005d Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 18 Oct 2022 13:20:38 -0400 Subject: [PATCH] Use unique consumer groups in KafkaDslKotlinTests --- .../kafka/dsl/kotlin/KafkaDslKotlinTests.kt | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/kotlin/KafkaDslKotlinTests.kt b/spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/kotlin/KafkaDslKotlinTests.kt index 52a07a08bd..c71613dbe7 100644 --- a/spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/kotlin/KafkaDslKotlinTests.kt +++ b/spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/kotlin/KafkaDslKotlinTests.kt @@ -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