add support for embedded sinks

Author: flyaruu

buildSrc/src/main/kotlin/Dependencies.kt

@@ -30,6 +30,7 @@ object FloodplainDeps {
     const val debeziumMySQL       = "io.debezium:debezium-connector-mysql:$debezium_version"
     const val debeziumEmbedded    = "io.debezium:debezium-embedded:$debezium_version"
     const val debeziumServerCore  = "io.debezium:debezium-server-core:$debezium_version"
+    const val debeziumTestContainers = "io.debezium:debezium-testing-testcontainers:$debezium_version"
     const val cdiApi              = "jakarta.enterprise:jakarta.enterprise.cdi-api:2.0.2"
     const val microProfileConfig  = "org.eclipse.microprofile.config:microprofile-config-api:1.3"
     const val jUnit               = "junit:junit:4.13.1"

floodplain-dsl/build.gradle.kts

@@ -21,6 +21,7 @@ dependencies {
     implementation(FloodplainDeps.argParser)
     testCompile(FloodplainDeps.jUnit)
     testCompile(FloodplainDeps.testContainer)
+    implementation(FloodplainDeps.kafkaConnectRuntime)
     implementation("org.jetbrains.kotlin:kotlin-stdlib-jdk8")
     implementation(FloodplainDeps.kotlinCoroutines)
     testImplementation("org.jetbrains.kotlin:kotlin-test")

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

@@ -43,8 +43,16 @@ import io.floodplain.replication.api.ReplicationMessage
 import io.floodplain.streams.api.Topic
 import io.floodplain.streams.api.TopologyContext
 import io.floodplain.streams.remotejoin.TopologyConstructor
+import org.apache.kafka.connect.sink.SinkConnector
+import org.apache.kafka.connect.sink.SinkRecord
+import org.apache.kafka.connect.sink.SinkTask
+import java.lang.RuntimeException
 import java.time.Duration
 import java.util.Optional
+import java.util.concurrent.atomic.AtomicLong
+import kotlin.system.measureTimeMillis
+
+private val logger = mu.KotlinLogging.logger {}
 
 /**
  * Super (wrapper) class for all components (source, transformer or sink)
@@ -78,10 +86,28 @@ interface SourceConfig : Config {
 
 interface SinkConfig : Config {
     fun sinkTask(): Any?
-    fun instantiateSinkElements()
+    fun instantiateSinkElements(): List<Map<String, String>>
     fun sinkElements(): Map<Topic, List<FloodplainSink>>
 }
 
+abstract class AbstractSinkConfig: SinkConfig {
+    // var instantiatedSinkElements: Map<Topic, List<FloodplainSink>>? = null
+    var sinkTask: SinkTask? = null
+    var floodplainSink: FloodplainSink? = null
+    var connector: SinkConnector? = null
+    override fun instantiateSinkElements(): List<Map<String, String>> {
+        val configs = materializeConnectorConfig()
+        if(configs.size > 1) {
+            throw RuntimeException("Multiple configs not supported for now")
+        }
+        return configs.map {
+            it.settings
+        }.toList()
+
+    }
+}
+
+
 class MaterializedConfig(val name: String, val topics: List<Topic>, val settings: Map<String, String>)
 
 interface SourceTopic {
@@ -197,7 +223,6 @@ fun PartialStream.joinAttributes(withTopic: String, nameAttribute: String, value
             scan(
                 keyExtract,
                 {
-                    _ ->
                     empty()
                 },
                 {
@@ -552,3 +577,38 @@ class Block(rootTopology: Stream, override val topologyContext: TopologyContext)
     //     throw UnsupportedOperationException("Blocks shouldn't add sources")
     // }
 }
+
+
+class AbstractFloodplainSink(private val task: SinkTask, private val config: SinkConfig) : FloodplainSink {
+    private val offsetCounter = AtomicLong(System.currentTimeMillis())
+
+    override fun send(topic: Topic, elements: List<Pair<String, Map<String, Any>?>>) {
+        val list = elements.map { (key, value) ->
+            SinkRecord(topic.qualifiedString(), 0, null, key, null, value, offsetCounter.incrementAndGet())
+        }.toList()
+        val insertTime = measureTimeMillis {
+            try {
+                task.put(list)
+            } catch (e: Throwable) {
+                e.printStackTrace()
+            }
+        }
+        logger.info("Inserting into mongodb size: ${list.size} duration: $insertTime")
+    }
+
+    override fun config(): SinkConfig {
+        return config
+    }
+
+    override fun flush() {
+        task.flush(emptyMap())
+    }
+
+    override fun close() {
+        task.close(emptyList())
+    }
+
+    override fun taskObject(): Any? {
+        TODO("Not yet implemented")
+    }
+}

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

@@ -99,12 +99,14 @@ fun startConstructor(
             deleteConnector(generatedName, connectURL)
         } else {
             logger.warn("Connector: {} already present, ignoring", generatedName)
+            // TODO return from here?
         }
     }
     postToHttpJava11(connectURL, jsonString)
 }
 
 private fun existingConnectors(url: URL): List<String> {
+    logger.info("Connecting to URL: $url")
     val response = httpClient.send(HttpRequest.newBuilder().uri(url.toURI()).build(), BodyHandlers.ofInputStream())
     val an = objectMapper.readTree(response.body()) as ArrayNode
     val result: MutableList<String> = ArrayList()
@@ -146,10 +148,10 @@ fun floodplainSinkFromTask(task: SinkTask, config: SinkConfig): FloodplainSink {
     return LocalConnectorSink(task, config)
 }
 
-fun instantiateSinkConfig(config: SinkConfig, connector: () -> SinkConnector): Map<Topic, MutableList<FloodplainSink>> {
+fun instantiateSinkConfig(config: SinkConfig, connector: () -> SinkConnector): SinkConnector {
     val result = mutableMapOf<Topic, MutableList<FloodplainSink>>()
     val materializedSinks = config.materializeConnectorConfig()
-    materializedSinks.map { materializedSink ->
+    val instance =  materializedSinks.map { materializedSink ->
         val connectorInstance = connector()
         connector().start(materializedSink.settings)
         val task = connectorInstance.taskClass().getDeclaredConstructor().newInstance() as SinkTask
@@ -160,11 +162,13 @@ fun instantiateSinkConfig(config: SinkConfig, connector: () -> SinkConnector): M
             val list = result.computeIfAbsent(topic) { mutableListOf() }
             list.add(localSink)
         }
-    }
-    return result
+        connectorInstance
+    }.first()
+    return instance
+    // return connectorInstance to instance
 }
 
-private class LocalConnectorSink(private val task: SinkTask, val config: SinkConfig) : FloodplainSink {
+class LocalConnectorSink(private val task: SinkTask, val config: SinkConfig) : FloodplainSink {
     private val offsetCounter = AtomicLong(System.currentTimeMillis())
     override fun send(topic: Topic, elements: List<Pair<String, Map<String, Any>?>>) {
         logger.info("Inserting # of documents ${elements.size} for topic: $topic")

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

@@ -50,6 +50,7 @@ import kotlinx.coroutines.launch
 import kotlinx.coroutines.newSingleThreadContext
 import kotlinx.coroutines.runBlocking
 import org.apache.kafka.common.serialization.Serdes
+import org.apache.kafka.common.utils.Utils
 import org.apache.kafka.connect.json.JsonDeserializer
 import org.apache.kafka.streams.KeyValue
 import org.apache.kafka.streams.StreamsConfig
@@ -132,7 +133,8 @@ fun runLocalTopology(
     props.setProperty(StreamsConfig.STATE_DIR_CONFIG, storageFolder)
     props.setProperty(StreamsConfig.BUILT_IN_METRICS_VERSION_CONFIG, StreamsConfig.METRICS_LATEST)
     val driver = TopologyTestDriver(topology, props)
-    val contextInstance = LocalDriverContext(driver, context, topologyConstructor, sourceConfigs, sinkConfigs, sinks, bufferTime)
+
+    val contextInstance = LocalDriverContext(driver, context, topologyConstructor, Utils.propsToStringMap(props), sourceConfigs, sinkConfigs, sinks, bufferTime)
     val jobs = contextInstance.connectSourceAndSink()
     contextInstance.connectJobs.addAll(jobs)
     try {
@@ -158,6 +160,7 @@ class LocalDriverContext(
     private val driver: TopologyTestDriver,
     private val topologyContext: TopologyContext,
     private val topologyConstructor: TopologyConstructor,
+    private val config: Map<String,String>,
     private val sourceConfigs: List<SourceConfig>,
     private val sinkConfigs: List<SinkConfig>,
     private val sinks: List<Pair<String, String>>,