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

Make replyContainer as bean in Kafka tests

Browse Source 
It looks like `replyContainer()` is not registered as a bean,
so its lifecycle is somehow out of application context control

* Mark `replyContainer()` method as a `@Bean` in the `KafkaDslTests` and `KafkaDslKotlinTests`
to see if this fixes flaky state of the test suite
* Upgrade to Kotlin `1.7.20` and fix `KafkaDslKotlinTests` according to its requirements
This commit is contained in:
Artem Bilan
2022-10-06 16:01:08 -04:00
parent f24fbd992b
commit daaa30e67f
3 changed files with 7 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.gradle
View File

@@ -1,5 +1,5 @@
buildscript { buildscript {
ext.kotlinVersion = '1.7.10' ext.kotlinVersion = '1.7.20'
ext.isCI = System.getenv('GITHUB_ACTION') || System.getenv('bamboo_buildKey') ext.isCI = System.getenv('GITHUB_ACTION') || System.getenv('bamboo_buildKey')
repositories { repositories {
mavenCentral() mavenCentral()

3
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java
View File

@@ -445,7 +445,8 @@ public class KafkaDslTests {
.get(); .get();
} }
private GenericMessageListenerContainer<Integer, String> replyContainer() { @Bean
public GenericMessageListenerContainer<Integer, String> replyContainer() {
ContainerProperties containerProperties = new ContainerProperties(TEST_TOPIC5); ContainerProperties containerProperties = new ContainerProperties(TEST_TOPIC5);
containerProperties.setGroupId("outGate"); containerProperties.setGroupId("outGate");
containerProperties.setConsumerRebalanceListener(new ConsumerRebalanceListener() { containerProperties.setConsumerRebalanceListener(new ConsumerRebalanceListener() {

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

@@ -190,7 +190,7 @@ class KafkaDslKotlinTests {
@Test @Test
fun testGateways() { fun testGateways() {
assertThat(this.config.replyContainerLatch.await(30, TimeUnit.SECONDS)) assertThat(this.config.replyContainerLatch.await(30, TimeUnit.SECONDS)).isTrue()
assertThat(this.gate.exchange(TEST_TOPIC4, "foo")).isEqualTo("FOO") assertThat(this.gate.exchange(TEST_TOPIC4, "foo")).isEqualTo("FOO")
} }
@@ -292,7 +292,7 @@ class KafkaDslKotlinTests {
private fun kafkaMessageHandler(producerFactory: ProducerFactory<Int, String>, topic: String) = private fun kafkaMessageHandler(producerFactory: ProducerFactory<Int, String>, topic: String) =
Kafka.outboundChannelAdapter(producerFactory) Kafka.outboundChannelAdapter(producerFactory)
.messageKey { it.headers[IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER] } .messageKey<Any> { it.headers[IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER] }
.headerMapper(mapper()) .headerMapper(mapper())
.sync(true) .sync(true)
.partitionId<Any> { 0 } .partitionId<Any> { 0 }
@@ -326,7 +326,8 @@ class KafkaDslKotlinTests {
) )
} }
private fun replyContainer(): GenericMessageListenerContainer<Int, String> { @Bean
fun replyContainer(): GenericMessageListenerContainer<Int, String> {
val containerProperties = ContainerProperties(TEST_TOPIC5) val containerProperties = ContainerProperties(TEST_TOPIC5)
containerProperties.setGroupId("kotlinOutGate") containerProperties.setGroupId("kotlinOutGate")
containerProperties.setConsumerRebalanceListener(object : ConsumerRebalanceListener { containerProperties.setConsumerRebalanceListener(object : ConsumerRebalanceListener {