spring-kafka: improve kafka publisher listener selection strategy (#602)

Author: osoykan

gradle.properties

@@ -9,6 +9,6 @@ projectUrl=https://github.com/Trendyol/stove
 licenceUrl=https://github.com/Trendyol/stove/blob/master/LICENCE
 licence=Apache-2.0 license
 snapshot=1.0.0-SNAPSHOT
-version=0.14.0
+version=0.14.1
 
 

starters/spring/stove-spring-testing-e2e-kafka/src/main/kotlin/com/trendyol/stove/testing/e2e/kafka/KafkaSystem.kt

@@ -53,15 +53,27 @@ class KafkaSystem(
         it.setProducerListener(getInterceptor())
         it.setCloseTimeout(1.seconds.toJavaDuration())
       }
-      .firstOrNone {
-        it.producerFactory.configurationProperties[ProducerConfig.BOOTSTRAP_SERVERS_CONFIG] == exposedConfiguration.bootstrapServers
-      }
+      .firstOrNone { safeContains(it, exposedConfiguration) }
       .getOrElse {
         logger.warn("No KafkaTemplate found for the configured bootstrap servers, using a fallback KafkaTemplate")
         createFallbackTemplate(exposedConfiguration)
       }
   }
 
+  @Suppress("UNCHECKED_CAST")
+  private fun safeContains(
+    it: KafkaTemplate<Any, Any>,
+    exposedConfiguration: KafkaExposedConfiguration
+  ): Boolean = it.producerFactory.configurationProperties[ProducerConfig.BOOTSTRAP_SERVERS_CONFIG]
+    .toOption()
+    .map {
+      when (it) {
+        is String -> it
+        is List<*> -> (it as List<String>).joinToString(",")
+        else -> ""
+      }
+    }.isSome { it.contains(exposedConfiguration.bootstrapServers) }
+
   private fun createFallbackTemplate(exposedConfiguration: KafkaExposedConfiguration): KafkaTemplate<Any, Any> {
     val producerFactory = DefaultKafkaProducerFactory<Any, Any>(
       mapOf(