clean warnings, removed duplicated / unused code

Author: flyaruu

floodplain-debezium/src/main/kotlin/io/floodplain/debezium/postgres/DebeziumSource.kt

@@ -100,7 +100,7 @@ fun createDebeziumChangeFlow(name: String, taskClass: String, hostname: String,
 private fun runDebeziumServer(props: Properties): Flow<ChangeRecord> {
     val engineKillSwitch = EngineKillSwitch()
     val totalTimeInSend = AtomicLong(0L)
-    return callbackFlow<ChangeRecord> {
+    return callbackFlow {
         val engine = DebeziumEngine.create(Json::class.java)
             .using(props)
             .notifying { record: ChangeEvent<String, String> ->

floodplain-debezium/src/test/java/SimpleMySQL.java

@@ -54,9 +54,7 @@ public static void main(String[] args) throws IOException, InterruptedException
 // Create the engine with this configuration ...
         try (DebeziumEngine<ChangeEvent<String, String>> engine = DebeziumEngine.create(Json.class)
                 .using(props)
-                .notifying(record -> {
-                    System.out.println(record);
-                }).build()
+                .notifying(System.out::println).build()
         ) {
             Executors.newSingleThreadExecutor().execute(engine);
 

floodplain-dsl/src/main/kotlin/io/floodplain/kotlindsl/LocalRuntime.kt

@@ -38,13 +38,13 @@ import kotlinx.coroutines.Job
 import kotlinx.coroutines.channels.awaitClose
 import kotlinx.coroutines.channels.sendBlocking
 import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.asFlow
-import kotlinx.coroutines.flow.broadcastIn
+import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.callbackFlow
 import kotlinx.coroutines.flow.catch
 import kotlinx.coroutines.flow.collect
 import kotlinx.coroutines.flow.filter
 import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.shareIn
 import kotlinx.coroutines.isActive
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.newSingleThreadContext
@@ -197,6 +197,7 @@ class LocalDriverContext(
     private fun <T> Flow<T>.handleErrors(): Flow<T> =
         catch { e -> logger.error("Error in flow $e") }
 
+    @OptIn(kotlinx.coroutines.ObsoleteCoroutinesApi::class)
     override fun connectSourceAndSink(): List<Job> {
         val outputJob = GlobalScope.launch(newSingleThreadContext("TopologySource"), CoroutineStart.UNDISPATCHED) {
             val outputFlows = outputFlows(this)
@@ -266,8 +267,8 @@ class LocalDriverContext(
                 val result = if (value == null) null else mapper.convertValue(parsed, object : TypeReference<Map<String, Any>>() {})
                 Triple(topic, key, result)
             }
-            .broadcastIn(context)
-            .asFlow()
+            .shareIn(context,SharingStarted.Lazily)
+            // .asFlow()
             .handleErrors()
         return topics.map { topic ->
             val flow = sourceFlow.filter { (incomingTopic, _, _) ->

floodplain-dsl/src/test/kotlin/io/floodplain/test/InstantiatedContainer.kt

@@ -9,6 +9,7 @@ dependencies {
     implementation(project(":streams"))
     implementation(project(":floodplain-stream-topology"))
     implementation(project(":floodplain-dsl"))
+    implementation(project(":floodplain-test"))
     implementation("io.confluent:kafka-connect-elasticsearch:5.5.0")
     implementation(FloodplainDeps.kotlinCoroutines)
     testImplementation("org.jetbrains.kotlin:kotlin-test")

floodplain-elasticsearch/build.gradle

@@ -23,10 +23,10 @@ import com.fasterxml.jackson.databind.ObjectMapper
 import io.floodplain.kotlindsl.message.empty
 import io.floodplain.kotlindsl.source
 import io.floodplain.kotlindsl.stream
+import io.floodplain.test.InstantiatedContainer
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.withTimeout
 import org.junit.Test
-import org.testcontainers.containers.GenericContainer
 import org.testcontainers.shaded.com.fasterxml.jackson.databind.node.NullNode
 import java.net.URI
 import java.net.http.HttpClient
@@ -42,25 +42,12 @@ class TestElasticSearch {
 
     private val objectMapper = ObjectMapper()
 
-    class KGenericContainer(imageName: String) : GenericContainer<KGenericContainer>(imageName)
-    var address: String? = "localhost"
-    var port: Int? = 0
-    var container: GenericContainer<*>? = null
-
-    init {
-        container = KGenericContainer("docker.elastic.co/elasticsearch/elasticsearch-oss:7.7.0")
-            .apply {
-                withExposedPorts(9200)
-                    .withEnv("discovery.type", "single-node")
-            }
-        container?.start()
-        address = container?.getHost()
-        port = container?.getFirstMappedPort()
-    }
+    private val container = InstantiatedContainer("docker.elastic.co/elasticsearch/elasticsearch-oss:7.7.0",
+        9200, mapOf("discovery.type" to "single-node"))
 
     @Test
     fun testElasticInsert() {
-        val uri = "http://$address:$port"
+        val uri = "http://${container.host}:${container.exposedPort}"
         stream {
             source("sometopic") {
                 val config = elasticSearchConfig("elasticName", uri)
@@ -116,7 +103,9 @@ class TestElasticSearch {
     }
 
     private fun queryUUIDHits(query: String): Int {
-        val node = queryUUID("http://$address:$port", "q=$query")
+        val uri = "http://${container.host}:${container.exposedPort}"
+
+        val node = queryUUID(uri, "q=$query")
         logger.info("Query uri: $node")
         val error = node.get("error")
         if (error == null || error is NullNode) {

floodplain-elasticsearch/src/test/kotlin/io/floodplain/elasticsearch/TestElasticSearch.kt

@@ -19,6 +19,7 @@
 package io.floodplain.test
 
 import org.testcontainers.containers.GenericContainer
+import org.testcontainers.utility.DockerImageName
 
 val useIntegraton: Boolean by lazy {
     System.getenv("NO_INTEGRATION") == null
@@ -29,7 +30,7 @@ val useIntegraton: Boolean by lazy {
  */
 class InstantiatedContainer(image: String, port: Int, env: Map<String, String> = emptyMap()) {
 
-    class KGenericContainer(imageName: String) : GenericContainer<KGenericContainer>(imageName)
+    class KGenericContainer(imageName: String) : GenericContainer<KGenericContainer>(DockerImageName.parse(imageName))
     var container: KGenericContainer?
     var host: String
     var exposedPort: Int = -1

floodplain-test/src/main/kotlin/io/floodplain/test/InstantiatedContainer.kt

@@ -201,6 +201,7 @@ public ImmutableMessage rename(String columnName, String newName) {
         return this;
     }
 
+    @SuppressWarnings("unchecked")
     @Override
     public ImmutableMessage with(String key, Object value, ValueType type) {