fix formatting

Author: flyaruu

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

@@ -89,28 +89,25 @@ interface SinkConfig : Config {
     fun sinkElements(): Map<Topic, List<FloodplainSink>>
 }
 
-abstract class AbstractSinkConfig: SinkConfig {
+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) {
+        if (configs.size > 1) {
             throw RuntimeException("Multiple configs not supported for now")
         }
         return configs.map {
             it.settings
         }.toList()
-
     }
     override fun sinkElements(): Map<Topic, List<FloodplainSink>> {
         return instantiateSinkConfig(this)
     }
-
 }
 
-
 class MaterializedConfig(val name: String, val topics: List<Topic>, val settings: Map<String, String>)
 
 interface SourceTopic {
@@ -223,7 +220,6 @@ fun PartialStream.joinMulti(key: (IMessage) -> String?, secondaryKey: (IMessage)
     addTransformer(Transformer(this.rootTopology, jrt, topologyContext))
 }
 
-
 @Deprecated("Don't use unqualified topics")
 fun PartialStream.joinAttributes(withTopic: String, nameAttribute: String, valueAttribute: String, vararg keys: String) {
     return joinAttributes(withTopic, nameAttribute, valueAttribute) { msg ->
@@ -242,7 +238,7 @@ fun PartialStream.joinAttributes(withTopic: String, nameAttribute: String, value
                 },
                 {
                     set {
-                            _, msg, acc ->
+                        _, msg, acc ->
                         val name = msg[nameAttribute] as String?
                         val value = msg[valueAttribute]
                         if (name == null) {
@@ -254,7 +250,7 @@ fun PartialStream.joinAttributes(withTopic: String, nameAttribute: String, value
                 },
                 {
                     set {
-                            _, msg, acc ->
+                        _, msg, acc ->
                         acc.clear(msg[nameAttribute] as String); acc
                     }
                 }
@@ -365,7 +361,7 @@ private fun createSource(
     return source
 }
 
-@Deprecated("Don't use unqualified topics",ReplaceWith("qualifiedTopic"))
+@Deprecated("Don't use unqualified topics", ReplaceWith("qualifiedTopic"))
 fun FloodplainOperator.topic(name: String): Topic {
     return Topic.from(name, topologyContext)
 }
@@ -380,10 +376,10 @@ fun FloodplainOperator.qualifiedTopic(name: String): Topic {
  * alltogether. When that is the case, we can simply use the qualified string instead of the Topic object
  */
 fun FloodplainOperator.generationalTopic(name: String): Topic {
-    if(name.startsWith("@")) {
+    if (name.startsWith("@")) {
         throw RuntimeException("Can't create generationalTopic that starts with '@', remove the '@' from $name")
     }
-    return Topic.from("@$name",topologyContext)
+    return Topic.from("@$name", topologyContext)
 }
 
 /**
@@ -478,7 +474,6 @@ fun PartialStream.externalSinkQualified(topic: String): Transformer {
     return addTransformer(Transformer(this.rootTopology, sink, topologyContext))
 }
 
-
 /**
  * Creates a simple sink that takes a lambda that will choose a **qualified** topic */
 fun PartialStream.dynamicSink(name: String, extractor: (String, IMessage) -> String): Transformer {
@@ -647,10 +642,7 @@ interface FloodplainOperator {
 /**
  * Concrete version of a partial pipe
  */
-class Block(rootTopology: Stream, override val topologyContext: TopologyContext) : PartialStream(rootTopology) {
-
-}
-
+class Block(rootTopology: Stream, override val topologyContext: TopologyContext) : PartialStream(rootTopology)
 
 class AbstractFloodplainSink(private val task: SinkTask, private val config: SinkConfig) : FloodplainSink {
     private val offsetCounter = AtomicLong(System.currentTimeMillis())
@@ -684,4 +676,4 @@ class AbstractFloodplainSink(private val task: SinkTask, private val config: Sin
     override fun taskObject(): Any? {
         TODO("Not yet implemented")
     }
-}
+}

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

@@ -152,7 +152,7 @@ fun instantiateSinkConfig(config: SinkConfig): MutableMap<Topic, MutableList<Flo
     val result = mutableMapOf<Topic, MutableList<FloodplainSink>>()
     val materializedSinks = config.materializeConnectorConfig()
     materializedSinks.forEach { materializedSink ->
-        val connectorInstance =  Class.forName(materializedSink.settings["connector.class"]).getDeclaredConstructor().newInstance() as SinkConnector
+        val connectorInstance = Class.forName(materializedSink.settings["connector.class"]).getDeclaredConstructor().newInstance() as SinkConnector
         connectorInstance.start(materializedSink.settings)
         val task = connectorInstance.taskClass().getDeclaredConstructor().newInstance() as SinkTask
         task.start(materializedSink.settings)

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

@@ -70,7 +70,7 @@ import java.util.Properties
 
 private val logger = mu.KotlinLogging.logger {}
 
-interface InputReceiver: FloodplainOperator {
+interface InputReceiver : FloodplainOperator {
     fun input(topic: Topic, key: String, msg: IMessage)
     fun input(topic: Topic, key: ByteArray, msg: ByteArray)
     fun inputQualified(topic: String, key: String, msg: IMessage)
@@ -164,7 +164,7 @@ class LocalDriverContext(
     override val rootTopology: Stream,
     override val topologyContext: TopologyContext,
     private val topologyConstructor: TopologyConstructor,
-    private val config: Map<String,String>,
+    private val config: Map<String, String>,
     private val sourceConfigs: List<SourceConfig>,
     private val sinkConfigs: List<SinkConfig>,
     private val sinks: List<Pair<String, String>>,
@@ -262,7 +262,7 @@ class LocalDriverContext(
             // it.instantiateSinkElements()
             // it.sinkElements().entries
         }.forEach { entry ->
-            entry.entries.forEach { (key,value)->
+            entry.entries.forEach { (key, value) ->
                 val list = result.computeIfAbsent(key) { mutableListOf() }
                 list.addAll(value)
             }

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

@@ -191,20 +191,20 @@ class Stream(override val topologyContext: TopologyContext, val topologyConstruc
      * 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, initialSettings: Map<String, String>? = null, monitor: (suspend Stream.(KafkaStreams)->Unit)? = null): KafkaStreams {
+    fun renderAndSchedule(connectorURL: URL?, kafkaHosts: String, force: Boolean = false, initialSettings: Map<String, String>? = null, monitor: (suspend Stream.(KafkaStreams) -> Unit)? = null): KafkaStreams {
         val (topology, sources, sinks) = render()
-        val settings = initialSettings?.toMutableMap()?: mutableMapOf()
-        val replicationFactor = initialSettings?.get(StreamsConfig.REPLICATION_FACTOR_CONFIG)?:"1"
+        val settings = initialSettings?.toMutableMap() ?: mutableMapOf()
+        val replicationFactor = initialSettings?.get(StreamsConfig.REPLICATION_FACTOR_CONFIG) ?: "1"
         settings["bootstrap.servers"] = kafkaHosts
-        settings.putIfAbsent("offset.storage.topic","OFFSET_STORAGE")
-        settings.putIfAbsent("config.storage.topic","CONFIG_STORAGE")
-        settings.putIfAbsent("status.storage.topic","STATUS_STORAGE")
-        settings.putIfAbsent("offset.storage.replication.factor",replicationFactor)
-        settings.putIfAbsent("offset.storage.partitions","2")
-        settings.putIfAbsent("status.storage.replication.factor",replicationFactor)
-        settings.putIfAbsent("status.storage.partitions","2")
-        settings.putIfAbsent("config.storage.replication.factor",replicationFactor)
-        settings.putIfAbsent("group.id","${topologyContext.applicationId()}-connectors")
+        settings.putIfAbsent("offset.storage.topic", "OFFSET_STORAGE")
+        settings.putIfAbsent("config.storage.topic", "CONFIG_STORAGE")
+        settings.putIfAbsent("status.storage.topic", "STATUS_STORAGE")
+        settings.putIfAbsent("offset.storage.replication.factor", replicationFactor)
+        settings.putIfAbsent("offset.storage.partitions", "2")
+        settings.putIfAbsent("status.storage.replication.factor", replicationFactor)
+        settings.putIfAbsent("status.storage.partitions", "2")
+        settings.putIfAbsent("config.storage.replication.factor", replicationFactor)
+        settings.putIfAbsent("group.id", "${topologyContext.applicationId()}-connectors")
         topologyConstructor.createTopicsAsNeeded(settings.toMap())
         sources.forEach { (name, json) ->
             connectorURL?.let {
@@ -219,11 +219,11 @@ class Stream(override val topologyContext: TopologyContext, val topologyConstruc
         instantiateLocalSinks(settings)
         val appId = topologyContext.topicName("@applicationId")
         val extra: MutableMap<String, Any> = mutableMapOf()
-            extra.putAll(settings)
+        extra.putAll(settings)
         val streams = runTopology(topology, appId, kafkaHosts, "storagePath", extra)
         logger.info { "Topology running!" }
         runBlocking {
-            monitor?.invoke(this@Stream,streams)
+            monitor?.invoke(this@Stream, streams)
         }
         return streams
     }
@@ -250,14 +250,14 @@ class Stream(override val topologyContext: TopologyContext, val topologyConstruc
         }
     }
 
-    private fun extendWorkerProperties(workerProps: MutableMap<String,String>) {
+    private fun extendWorkerProperties(workerProps: MutableMap<String, String>) {
         workerProps["key.converter"] = "org.apache.kafka.connect.json.JsonConverter"
         workerProps["value.converter"] = "org.apache.kafka.connect.json.JsonConverter"
         // workerProps["offset.storage.file.filename"] = "offset"
         // TODO: Using port 8084 now, that might clash. Random? Don't know.
         workerProps[WorkerConfig.LISTENERS_CONFIG] = "http://127.0.0.1:8084"
         val keys = workerProps.keys.toSet()
-        keys.filter { it.startsWith("security") || it.startsWith("sasl")|| it.startsWith("ssl") || it.startsWith("bootstrap")}
+        keys.filter { it.startsWith("security") || it.startsWith("sasl") || it.startsWith("ssl") || it.startsWith("bootstrap") }
             .forEach { key ->
                 workerProps["consumer.$key"] = workerProps[key]!!
                 workerProps["producer.$key"] = workerProps[key]!!
@@ -272,15 +272,16 @@ class Stream(override val topologyContext: TopologyContext, val topologyConstruc
     }
 
     private fun startLocalConnect(initialWorkerProps: Map<String, String>): Herder {
-        val workerProps = mutableMapOf<String,String>()
+        val workerProps = mutableMapOf<String, String>()
         workerProps.putAll(initialWorkerProps)
         extendWorkerProperties(workerProps)
         val plugins = Plugins(workerProps)
         val config = DistributedConfig(workerProps)
         val time: Time = Time.SYSTEM
         val connectorClientConfigOverridePolicy: ConnectorClientConfigOverridePolicy = plugins.newPlugin(
             config.getString(WorkerConfig.CONNECTOR_CLIENT_POLICY_CLASS_CONFIG),
-            config, ConnectorClientConfigOverridePolicy::class.java
+            config,
+            ConnectorClientConfigOverridePolicy::class.java
         )
         val kafkaClusterId: String = ConnectUtils.lookupKafkaClusterId(config)
         logger.debug("Kafka cluster ID: $kafkaClusterId")
@@ -293,15 +294,19 @@ class Stream(override val topologyContext: TopologyContext, val topologyConstruc
 
         val workerId: String = advertisedUrl.host.toString() + ":" + advertisedUrl.port
         val worker = Worker(
-            workerId, time, plugins, config, offsetBackingStore,
+            workerId,
+            time,
+            plugins,
+            config,
+            offsetBackingStore,
             connectorClientConfigOverridePolicy
         )
 
         val statusBackingStore: StatusBackingStore = KafkaStatusBackingStore(time, worker.getInternalValueConverter())
         statusBackingStore.configure(config)
 
-       val configBackingStore = KafkaConfigBackingStore(worker.getInternalValueConverter(),config,worker.configTransformer())
-        val herder = DistributedHerder(config,time,worker,kafkaClusterId,statusBackingStore,configBackingStore,advertisedUrl.toString(),connectorClientConfigOverridePolicy)
+        val configBackingStore = KafkaConfigBackingStore(worker.getInternalValueConverter(), config, worker.configTransformer())
+        val herder = DistributedHerder(config, time, worker, kafkaClusterId, statusBackingStore, configBackingStore, advertisedUrl.toString(), connectorClientConfigOverridePolicy)
 
         // val herder: Herder = StandaloneHerder(worker, kafkaClusterId, connectorClientConfigOverridePolicy)
         val connect = Connect(herder, rest)
@@ -333,7 +338,7 @@ class Stream(override val topologyContext: TopologyContext, val topologyConstruc
         // }
     }
 
-    private fun renderLocal(): Pair<Topology,List<MaterializedConfig>> {
+    private fun renderLocal(): Pair<Topology, List<MaterializedConfig>> {
         val topology = renderTopology()
         val sources = sourceConfigurations().flatMap { element ->
             element.materializeConnectorConfig()

floodplain-dsl/src/main/kotlin/io/floodplain/kotlindsl/message/IMessage.kt

@@ -62,7 +62,6 @@ interface IMessage {
     fun merge(msg: IMessage): IMessage
     fun messageList(path: String): List<IMessage>?
     fun messageElement(path: String, index: Int): IMessage?
-
 }
 
 private data class IMessageImpl(private val content: MutableMap<String, Any>) : IMessage {

floodplain-dsl/src/test/kotlin/io/floodplain/kotlindsl/IMessageTest.kt

@@ -236,10 +236,10 @@ class IMessageTest {
     fun testQueryMessageListElement() {
         val msg = createComplexMessage()
         val id = msg.messageList("sublist")?.get(0)?.string("astring")
-        val id2 = msg.messageElement("sublist",0)?.string("astring")
-        val nonExistingIndex = msg.messageElement("sublist",15)?.string("astring")
-        val nonExistingList = msg.messageElement("sublist2",0)?.string("astring")
-        assertEquals(id,id2)
+        val id2 = msg.messageElement("sublist", 0)?.string("astring")
+        val nonExistingIndex = msg.messageElement("sublist", 15)?.string("astring")
+        val nonExistingList = msg.messageElement("sublist2", 0)?.string("astring")
+        assertEquals(id, id2)
         assertNull(nonExistingIndex)
         assertNull(nonExistingList)
     }

floodplain-dsl/src/test/kotlin/io/floodplain/kotlindsl/TestTopology.kt

@@ -59,7 +59,7 @@ class TestTopology {
     }
 
     private fun testQualifiedWithTenantAndDeployment(tenant: String?, deployment: String?, generation: String?) {
-        stream(tenant,deployment,generation?:"defaultGeneration") {
+        stream(tenant, deployment, generation ?: "defaultGeneration") {
             from("sometopic") {
                 sinkQualified("outputTopic")
             }
@@ -72,7 +72,7 @@ class TestTopology {
     }
 
     private fun testNonQualifiedWithTenantAndDeployment(tenant: String?, deployment: String?, generation: String?) {
-        stream(tenant,deployment, generation?: "defaultGeneration") {
+        stream(tenant, deployment, generation ?: "defaultGeneration") {
             source("@sometopic") {
                 sink("@outputTopic")
             }
@@ -89,15 +89,15 @@ class TestTopology {
 
         // Test both qualified and non-qualified topic resolution.
         // I intend to drop non-qualified in the future
-        testNonQualifiedWithTenantAndDeployment(null,null,null)
-        testNonQualifiedWithTenantAndDeployment("ten",null,null)
-        testNonQualifiedWithTenantAndDeployment("ten","dep",null)
-        testNonQualifiedWithTenantAndDeployment("ten","dep","somegeneration")
-
-        testQualifiedWithTenantAndDeployment(null,null,null)
-        testQualifiedWithTenantAndDeployment("ten",null,null)
-        testQualifiedWithTenantAndDeployment("ten","dep",null)
-        testQualifiedWithTenantAndDeployment("ten","dep","somegeneration")
+        testNonQualifiedWithTenantAndDeployment(null, null, null)
+        testNonQualifiedWithTenantAndDeployment("ten", null, null)
+        testNonQualifiedWithTenantAndDeployment("ten", "dep", null)
+        testNonQualifiedWithTenantAndDeployment("ten", "dep", "somegeneration")
+
+        testQualifiedWithTenantAndDeployment(null, null, null)
+        testQualifiedWithTenantAndDeployment("ten", null, null)
+        testQualifiedWithTenantAndDeployment("ten", "dep", null)
+        testQualifiedWithTenantAndDeployment("ten", "dep", "somegeneration")
     }
     @Test
     fun testDelete() {
@@ -556,10 +556,9 @@ class TestTopology {
     fun testReplicationParserForDateTime() {
         val data = javaClass.classLoader.getResource("decimalwithscale.json")?.readBytes()
             ?: throw IllegalArgumentException("Missing json file for decimalwithscale.json")
-        val replicationMessage = JSONToReplicationMessage.processDebeziumBody(data, Optional.empty());
+        val replicationMessage = JSONToReplicationMessage.processDebeziumBody(data, Optional.empty())
         val date = replicationMessage.value("payment_date")
         assertNotNull(date)
-
     }
 
     @Test
@@ -883,9 +882,9 @@ class TestTopology {
                 sinkQualified("output")
             }
         }.renderAndExecute {
-            input( qualifiedTopic("external"), originalKey.toByteArray(), body.toByteArray())
-            val (key,value) = output(qualifiedTopic("output"))
-            assertEquals("965",key)
+            input(qualifiedTopic("external"), originalKey.toByteArray(), body.toByteArray())
+            val (key, value) = output(qualifiedTopic("output"))
+            assertEquals("965", key)
             logger.info("Result: $value")
         }
     }

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

@@ -19,7 +19,6 @@
 package io.floodplain.kotlindsl.example
 
 import io.floodplain.kotlindsl.buffer
-import io.floodplain.kotlindsl.from
 import io.floodplain.kotlindsl.joinAttributes
 import io.floodplain.kotlindsl.joinRemote
 import io.floodplain.kotlindsl.set

floodplain-fhir/src/main/kotlin/FhirSource.kt

@@ -51,7 +51,7 @@ fun Stream.fhirGeneric(topic: Topic, init: Source.() -> Unit = {}) {
 }
 
 fun <T : IBaseResource> PartialStream.fhirSource(topic: String, transform: (T) -> IMessage, init: Source.() -> Unit = {}): Source {
-    return fhirSource<T>(Topic.fromQualified(topic,topologyContext),transform,init)
+    return fhirSource<T>(Topic.fromQualified(topic, topologyContext), transform, init)
 }
 
 fun <T : IBaseResource> PartialStream.fhirSource(topic: Topic, transform: (T) -> IMessage, init: Source.() -> Unit = {}): Source {

floodplain-googlesheets/src/main/kotlin/io/floodplain/sink/sheet/GoogleSheetSink.kt

@@ -88,7 +88,6 @@ class GoogleSheetConfiguration(override val topologyContext: TopologyContext, ov
         return instantiatedSinkElements ?: emptyMap()
     }
 
-
     override fun sinkTask(): Any? {
         return googleTask
     }

floodplain-googlesheets/src/test/kotlin/GoogleSheetTest.kt

@@ -19,9 +19,7 @@
 import io.floodplain.kotlindsl.each
 import io.floodplain.kotlindsl.from
 import io.floodplain.kotlindsl.message.empty
-import io.floodplain.kotlindsl.qualifiedTopic
 import io.floodplain.kotlindsl.set
-import io.floodplain.kotlindsl.source
 import io.floodplain.kotlindsl.stream
 import io.floodplain.sink.sheet.GoogleSheetConfiguration
 import io.floodplain.sink.sheet.googleSheetConfig

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

@@ -75,7 +75,7 @@ class FilmSimple {
             )
             postgresSource("film", postgresConfig) {
                 // Clear the last_update field, it makes no sense in a denormalized situation
-                each { _,msg,_ ->
+                each { _, msg, _ ->
                     logger.info("Last update: ${msg["last_update"]?.javaClass}")
                 }
                 toMongo("filmwithactors", "$generation-filmwithcat", mongoConfig)