support confluent cloud

Author: flyaruu

buildSrc/src/main/kotlin/Dependencies.kt

@@ -16,7 +16,7 @@ fun isReleaseVersion(): Boolean {
 }
 
 object FloodplainDeps {
-    const val floodplain_version = "1.3.2-SNAPSHOT"
+    const val floodplain_version = "1.3.4-SNAPSHOT"
     const val jackson_version = "2.11.3"
     const val kafka_version = "2.7.0"
     const val slf4j_version = "1.7.30"

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

@@ -33,6 +33,7 @@ import org.apache.kafka.streams.KafkaStreams
 import org.apache.kafka.streams.StreamsConfig
 import org.apache.kafka.streams.Topology
 import org.apache.kafka.streams.processor.WallclockTimestampExtractor
+import java.io.InputStream
 import java.lang.StringBuilder
 import java.net.URL
 import java.util.Properties
@@ -140,16 +141,24 @@ class Stream(override val topologyContext: TopologyContext) : FloodplainSourceCo
         )
     }
 
+    fun renderAndSchedule(connectorURL: URL?, settings: InputStream, force: Boolean = false): KafkaStreams {
+        val prop = Properties()
+        prop.load(settings)
+        val propMap = mutableMapOf<String,Any>()
+        prop.forEach { (k,v)-> propMap.put(k as String,v) }
+        return renderAndSchedule(connectorURL,prop[StreamsConfig.BOOTSTRAP_SERVERS_CONFIG] as String,force,propMap)
+    }
+
     /**
      * Will create an executable definition of the str
      * eam (@see render), then will start the topology by starting a streams
      * instance pointing at the kafka cluster at kafkaHosts, using the supplied clientId.
      * Finally, it will POST the supplied
      */
-    fun renderAndSchedule(connectorURL: URL?, kafkaHosts: String, force: Boolean = false): KafkaStreams {
+    fun renderAndSchedule(connectorURL: URL?, kafkaHosts: String,  force: Boolean = false,settings: Map<String,Any>? = null): KafkaStreams {
         val topologyConstructor = TopologyConstructor()
         val (topology, sources, sinks) = render(topologyConstructor)
-        topologyConstructor.createTopicsAsNeeded(topologyContext, kafkaHosts)
+        topologyConstructor.createTopicsAsNeeded(settings?: mapOf(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG to kafkaHosts))
         sources.forEach { (name, json) ->
             connectorURL?.let {
                 startConstructor(name, topologyContext, it, json, force)
@@ -161,7 +170,11 @@ class Stream(override val topologyContext: TopologyContext) : FloodplainSourceCo
             }
         }
         val appId = topologyContext.topicName("@applicationId")
-        val streams = runTopology(topology, appId, kafkaHosts, "storagePath")
+        val extra: MutableMap<String,Any> = mutableMapOf()
+        if(settings!=null) {
+            extra.putAll(settings)
+        }
+        val streams = runTopology(topology, appId, kafkaHosts, "storagePath", extra)
         logger.info { "Topology running!" }
         return streams
     }
@@ -187,8 +200,12 @@ class Stream(override val topologyContext: TopologyContext) : FloodplainSourceCo
         return Triple(topology, sources, sinks)
     }
 
-    private fun runTopology(topology: Topology, applicationId: String, kafkaHosts: String, storagePath: String): KafkaStreams {
-        val props = createProperties(applicationId, kafkaHosts, storagePath)
+    private fun runTopology(topology: Topology, applicationId: String, kafkaHosts: String, storagePath: String, extra: MutableMap<String,Any>): KafkaStreams {
+        extra[StreamsConfig.BOOTSTRAP_SERVERS_CONFIG] = kafkaHosts
+        extra[StreamsConfig.APPLICATION_ID_CONFIG] = applicationId
+        extra[StreamsConfig.STATE_DIR_CONFIG] = storagePath
+
+        val props = createProperties(extra)
         val stream = KafkaStreams(topology, props)
         logger.info("CurrentTopology:\n ${topology.describe()}")
         stream.setUncaughtExceptionHandler { thread: Thread, exception: Throwable? ->
@@ -202,15 +219,12 @@ class Stream(override val topologyContext: TopologyContext) : FloodplainSourceCo
         return stream
     }
 
-    private fun createProperties(applicationId: String, brokers: String, storagePath: String): Properties {
+    private fun createProperties(extra: Map<String,Any>): Properties {
         val streamsConfiguration = Properties()
         // Give the Streams application a unique name.  The name must be unique in the Kafka cluster
         // against which the application is run.
-        logger.info("Creating application with name: {}", applicationId)
-        logger.info("Creating application id: {}", applicationId)
-        logger.info("Starting instance in storagePath: {}", storagePath)
-        streamsConfiguration[StreamsConfig.APPLICATION_ID_CONFIG] = applicationId
-        streamsConfiguration[StreamsConfig.BOOTSTRAP_SERVERS_CONFIG] = brokers
+        streamsConfiguration.putAll(extra)
+        // logger.info("Starting instance in storagePath: {}", storagePath)
         streamsConfiguration[StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG] = Serdes.String().javaClass
         streamsConfiguration[StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG] = StreamOperators.replicationSerde.javaClass
         streamsConfiguration[ConsumerConfig.AUTO_OFFSET_RESET_CONFIG] = "earliest"
@@ -220,7 +234,7 @@ class Stream(override val topologyContext: TopologyContext) : FloodplainSourceCo
         streamsConfiguration[ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG] = 7200000
         streamsConfiguration[ConsumerConfig.MAX_POLL_RECORDS_CONFIG] = 100
         streamsConfiguration[ProducerConfig.COMPRESSION_TYPE_CONFIG] = "lz4"
-        streamsConfiguration[StreamsConfig.STATE_DIR_CONFIG] = storagePath
+        // streamsConfiguration[StreamsConfig.STATE_DIR_CONFIG] = storagePath
         streamsConfiguration[StreamsConfig.NUM_STREAM_THREADS_CONFIG] = 1
         streamsConfiguration[StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG] = 0
         streamsConfiguration[StreamsConfig.RETRIES_CONFIG] = 50

floodplain-example/src/main/kotlin/io/floodplain/kotlindsl/example/CalendarDate.kt

@@ -1,7 +1,6 @@
 package io.floodplain.kotlindsl.example
 
 import io.floodplain.kotlindsl.each
-import io.floodplain.kotlindsl.set
 import io.floodplain.kotlindsl.source
 import io.floodplain.kotlindsl.stream
 import io.floodplain.mongodb.mongoConfig

floodplain-integration/src/test/kotlin/io.floodplain.integration/TestTopicCreation.kt

@@ -1,16 +1,15 @@
 package io.floodplain.integration
 
-import io.floodplain.test.useIntegraton
+import io.floodplain.test.RedPandaContainer
 import org.apache.kafka.clients.admin.AdminClient
 import org.apache.kafka.clients.admin.CreateTopicsOptions
 import org.apache.kafka.clients.admin.ListTopicsOptions
 import org.apache.kafka.clients.admin.NewTopic
 import org.apache.kafka.common.config.ConfigResource
-import org.apache.kafka.common.config.ConfigResource.*
 import org.apache.kafka.common.config.ConfigResource.Type.TOPIC
+import org.junit.Assert
 import org.junit.Ignore
 import org.junit.Test
-import org.testcontainers.containers.KafkaContainer
 import java.util.Collections
 import java.util.UUID
 import java.util.HashMap
@@ -24,14 +23,21 @@ class TestTopicCreation {
         // .withEnv(env)
 
     // Infra for testing Kafka interaction
+    private val panda = RedPandaContainer("vectorized/redpanda:latest", 9092)
+    // val kafkaContainer = KafkaContainer("5.5.3").withEmbeddedZookeeper().withExposedPorts(9092,9093)
+
+    // Not functional yet
     @Test @Ignore
     fun testCreateTopic() {
-        val kafkaContainer = KafkaContainer("5.5.3").withEmbeddedZookeeper().withExposedPorts(9092,9093)
 
-        kafkaContainer.start()
-        logger.info("Bootstrap: ${kafkaContainer.bootstrapServers}")
+
+        // kafkaContainer.start()
+        // logger.info("Bootsx§trap: ${kafkaContainer.bootstrapServers}")
         val config: MutableMap<String, Any> = HashMap()
-        config["bootstrap.servers"] = kafkaContainer.bootstrapServers
+        // val exposedPort = panda.exposedPort
+        val host = "localhost:${panda.exposedPort}"
+        logger.info("Exposed host: $host")
+        config["bootstrap.servers"] = host //"localhost:51347"
         config["client.id"] = UUID.randomUUID().toString()
         val adminClient = AdminClient.create(config)
         adminClient.listTopics().names().get().forEach {
@@ -42,15 +48,16 @@ class TestTopicCreation {
         val topicDescription = adminClient.describeTopics(listOf("mytopic")).all().get()["mytopic"]
     //    topicDescription?
         val cr =  Collections.singleton(ConfigResource(TOPIC, "mytopic"))
-        val configsResult = adminClient.describeConfigs(cr);
+        val configsResult = adminClient.describeConfigs(cr)
         val cnf = configsResult.all().get()["mytopic"]
 
 //        adminClient.createTopics()
         val configMap: MutableMap<String, String> = HashMap()
         configMap["cleanup.policy"] = "compact"
 
 
-    //        adminClient.listTopics(ListTopicsOptions())
+        Assert.assertEquals(1,adminClient.listTopics(ListTopicsOptions()).names().get().size)
+
     }
 
     // KafkaContainer kafka = new KafkaContainer(KAFKA_TEST_IMAGE)

floodplain-mongodb/src/main/kotlin/io/floodplain/mongodb/MongoSink.kt

@@ -65,6 +65,10 @@ class MongoConfig(val name: String, private val uri: String, private val databas
             "document.id.strategy" to "com.mongodb.kafka.connect.sink.processor.id.strategy.FullKeyStrategy",
             "delete.on.null.values" to "true",
             "debug" to "true",
+            "topic.creation.default.replication.factor" to "3",
+            "topic.creation.default.partitions" to "1",
+            "topic.creation.default.cleanup.policy" to "compact",
+            "topic.creation.default.compression.type" to "lz4",
             "connection.uri" to uri,
             "database" to generationalDatabase,
             "collection" to collections,

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

@@ -31,13 +31,12 @@ val useIntegraton: Boolean by lazy {
 class InstantiatedContainer(image: String, port: Int, env: Map<String, String> = emptyMap()) {
 
     class KGenericContainer(imageName: String) : GenericContainer<KGenericContainer>(DockerImageName.parse(imageName))
-    var container: KGenericContainer?
+    var container: KGenericContainer? = KGenericContainer(image)
+        .apply { withExposedPorts(port) }
+        .apply { withEnv(env) }
     var host: String
     var exposedPort: Int = -1
     init {
-        container = KGenericContainer(image)
-            .apply { withExposedPorts(port) }
-            .apply { withEnv(env) }
         container?.start()
         host = container?.host ?: "localhost"
         exposedPort = container?.firstMappedPort ?: -1

streams/src/main/java/io/floodplain/streams/remotejoin/TopologyConstructor.java

@@ -67,9 +67,10 @@ public void ensureTopicExists(Topic topicName, Optional<Integer> partitionCount)
         desiredTopics.put(topicName, partitionCount);
     }
 
-    public void createTopicsAsNeeded(TopologyContext topologyContext, String kafkaHosts) {
-        Map<String, Object> config = new HashMap<>();
-        config.put("bootstrap.servers", kafkaHosts);
+    public void createTopicsAsNeeded(Map<String,Object> settings) {
+        Map<String, Object> config = new HashMap<>(settings);
+//        config.put("bootstrap.servers", kafkaHosts);
+        // TODO remove these?
         config.put("client.id", UUID.randomUUID().toString());
         config.put("cleanup.policy","compact");
         AdminClient adminClient = AdminClient.create(config);