Improve static code analysis, fix formatting / complexity issues

Author: flyaruu

.editorconfig

@@ -1,2 +1,2 @@
 [*.{kt,kts}]
-max_line_length=130
+max_line_length=120

build.gradle.kts

@@ -1,7 +1,6 @@
 import io.floodplain.build.FloodplainDeps
 import io.floodplain.build.isReleaseVersion
 import io.gitlab.arturbosch.detekt.Detekt
-import org.gradle.kotlin.dsl.detekt as detekt
 
 buildscript {
     repositories {
@@ -31,6 +30,7 @@ plugins {
     signing
     `maven-publish`
     `java-library`
+    jacoco
 }
 
 dependencies {
@@ -67,6 +67,7 @@ subprojects {
     apply(plugin = "org.jetbrains.dokka")
     apply(plugin = "com.github.hierynomus.license-base")
     apply(plugin = "checkstyle")
+    apply(plugin = "jacoco")
     if (!isKotlinModule(this)) {
         apply(plugin = "com.github.spotbugs")
     }
@@ -249,7 +250,46 @@ val detektAll by tasks.registering(Detekt::class) {
     exclude("**/build/**")
     // baseline.set(project.rootDir.resolve("reports/baseline.xml"))
     reports {
-        xml.enabled = false
+        xml.enabled = true
         html.enabled = true
     }
 }
+
+jacoco {
+    toolVersion = "0.8.6"
+}
+
+tasks.test {
+    finalizedBy(tasks.jacocoTestReport) // report is always generated after tests run
+}
+
+tasks.jacocoTestReport {
+    dependsOn(tasks.test) // tests are required to run before generating the report
+}
+
+// tasks.jacocoTestReport {
+//     reports {
+//         xml.isEnabled = true
+//         csv.isEnabled = false
+//         html.isEnabled = true
+//         html.destination = layout.buildDirectory.dir("jacocoHtml").get().asFile
+//     }
+// }
+val codeCoverageReport by tasks.creating(JacocoReport::class) {
+    executionData(fileTree(project.rootDir.absolutePath).include("**/build/jacoco/*.exec"))
+
+    subprojects.onEach {
+        sourceSets(it.sourceSets["main"])
+    }
+
+    reports {
+        // sourceDirectories =  files(sourceSets["main"].allSource.srcDirs)
+        // classDirectories =  files(sourceSets["main"].output)
+        xml.isEnabled = true
+        xml.destination = File("$buildDir/reports/jacoco/report.xml")
+        html.isEnabled = true
+        csv.isEnabled = false
+    }
+
+    dependsOn("test")
+}

buildSrc/build.gradle.kts

@@ -10,4 +10,4 @@ repositories {
     mavenLocal()
     mavenCentral()
     gradlePluginPortal()
-}
+}

detektConfig.yml

@@ -5,7 +5,7 @@ exceptions:
 #    excludes: ['**/Chunk.kt']
 formatting:
   MaximumLineLength:
-    maxLineLength: 140
+    maxLineLength: 120
 
 complexity:
   TooManyFunctions:
@@ -14,6 +14,10 @@ complexity:
     excludes: ['**/test/**']
   LargeClass:
     excludes: ['**/test/**']
+  LongParameterList:
+    functionThreshold: 7
+    constructorThreshold: 10
+    ignoreDefaultParameters: true
 
 style:
   ReturnCount:

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

@@ -56,7 +56,7 @@ internal class EngineKillSwitch(var engine: DebeziumEngine<ChangeEvent<String, S
 
 @OptIn(ExperimentalPathApi::class)
 private fun createOffsetFilePath(offsetId: String?): Path {
-    val tempFile = createTempFile(offsetId ?: UUID.randomUUID().toString().substring(0, 7))
+    val tempFile = createTempFile(offsetId ?: UUID.randomUUID().toString().substring(0..7))
     if (offsetId == null) {
         tempFile.toFile().deleteOnExit()
     }
@@ -96,13 +96,15 @@ private fun createLocalDebeziumSettings(
 
 /**
  * @return A hot flow of ChangeRecord. Perhaps one day there might be a colder one.
- * @param name: The prefix of the outgoing 'topic', basically the destination field of the changerecord is <topicprefix>.<schema>.<table>
+ * @param name: The prefix of the outgoing 'topic', basically the destination
+ * field of the changerecord is <topicprefix>.<schema>.<table>
  * @param hostname: The host of the postgres database
  * @param port: The port of the postgres database
  * @param username: The username of the postgres database
  * @param password: The password of the postgres database
  * @param offsetId: By default, we will save the offsets in a file path
- * @param settings An optional string-string map, that represents any extra parameters you want to pass to Debezium
+ * @param settings An optional string-string map, that represents any extra
+ * parameters you want to pass to Debezium
  * Defaults to empty map.
  *
  */
@@ -117,7 +119,10 @@ fun createDebeziumChangeFlow(
     offsetId: String? = null,
     settings: Map<String, String> = emptyMap()
 ): Flow<ChangeRecord> {
-    val props = createLocalDebeziumSettings(name, taskClass, hostname, port, database, username, password, offsetId, settings)
+    val props = createLocalDebeziumSettings(
+        name, taskClass, hostname, port, database,
+        username, password, offsetId, settings
+    )
     props.list(System.out)
     return runDebeziumServer(props)
 }

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

@@ -103,6 +103,7 @@ abstract class AbstractSinkConfig : SinkConfig {
             it.settings
         }.toList()
     }
+
     override fun sinkElements(): Map<Topic, List<FloodplainSink>> {
         return instantiateSinkConfig(this)
     }
@@ -168,8 +169,8 @@ fun PartialStream.buffer(duration: Duration, maxSize: Int = 10000, inMemory: Boo
 /**
  * Modifies the incoming message before passing it on.
  * You can use it as a simple stateless transformation: Add constant values, remove fields or transform fields.
- * A 'set' is also required for more complicated combinations. Typically after a join you use a 'set' to merge the messages
- * from both sources.
+ * A 'set' is also required for more complicated combinations. Typically after a join you use a 'set' to merge
+ * the messages from both sources.
  * The first message is the original message from the join, the second message the result from the 'secondary' source.
  * The second message will be lost after this operation, so you need to append anything you want to keep from the inner
  * source to the outer source here.
@@ -256,7 +257,12 @@ fun PartialStream.joinMulti(
     addTransformer(Transformer(this.rootTopology, jrt, topologyContext))
 }
 
-fun PartialStream.joinAttributes(withTopic: String, nameAttribute: String, valueAttribute: String, vararg keys: String) {
+fun PartialStream.joinAttributes(
+    withTopic: String,
+    nameAttribute: String,
+    valueAttribute: String,
+    vararg keys: String
+) {
     return joinAttributes(withTopic, nameAttribute, valueAttribute) { msg ->
         keys.joinToString(ReplicationMessage.KEYSEPARATOR) { msg[it].toString() }
     }
@@ -295,6 +301,7 @@ fun PartialStream.joinAttributes(
         }
     }
 }
+
 /**
  * Group a source, using the key from the lambda. The messages will be unchanged,
  * only the key will have the supplied key pre-pended.
@@ -306,8 +313,8 @@ fun PartialStream.group(key: (IMessage) -> String) {
 }
 
 /**
-* Create sub source (without qualifying with tenant / deployment / gen)
-*/
+ * Create sub source (without qualifying with tenant / deployment / gen)
+ */
 fun PartialStream.from(topic: String, init: Source.() -> Unit = {}): Source {
     return createSource(Topic.fromQualified(topic, rootTopology.topologyContext), rootTopology, init)
 }
@@ -373,7 +380,8 @@ private fun existingDebeziumSource(topicSource: String, rootTopology: Stream, in
 }
 
 /**
- * Creates a simple sink that will contain the result of the current transformation. Will not qualify with tenant / deployment
+ * Creates a simple sink that will contain the result of the current transformation.
+ * Will not qualify with tenant / deployment
  */
 fun PartialStream.to(topic: String): Transformer {
     val sink = SinkTransformer(
@@ -472,6 +480,7 @@ fun PartialStream.scan(
 fun PartialStream.createBlock(): Block {
     return Block(rootTopology, topologyContext)
 }
+
 /**
  * Scan is effectively a 'reduce' operator (The 'scan' name is used in Rx, which means a reduce operator that emits a
  * 'running aggregate' every time it consumes a message). A real reduce makes no sense in infinite streams
@@ -524,7 +533,12 @@ fun PartialStream.fork(vararg destinations: Block.() -> Unit): Transformer {
  * with no further meaning within the framework, you can choose what meaning you want to attach. You can increment a
  * number, use a sort of time stamp, or even a git commit.
  */
-fun stream(tenant: String? = null, deployment: String? = null, generation: String = "any", init: Stream.() -> Unit): Stream {
+fun stream(
+    tenant: String? = null,
+    deployment: String? = null,
+    generation: String = "any",
+    init: Stream.() -> Unit
+): Stream {
     val topologyContext = TopologyContext.context(
         Optional.ofNullable(tenant),
         Optional.ofNullable(deployment),

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

@@ -165,14 +165,17 @@ fun instantiateSinkConfig(config: SinkConfig): MutableMap<Topic, MutableList<Flo
     val materializedSinks = config.materializeConnectorConfig()
     materializedSinks.forEach { materializedSink ->
         val connectorInstance =
-            Class.forName(materializedSink.settings["connector.class"]).getDeclaredConstructor().newInstance() as SinkConnector
+            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)
         val keyConverter =
-            Class.forName(materializedSink.settings["key.converter"]).getDeclaredConstructor().newInstance() as Converter
+            Class.forName(materializedSink.settings["key.converter"])
+                .getDeclaredConstructor().newInstance() as Converter
         val valueConverter =
-            Class.forName(materializedSink.settings["value.converter"]).getDeclaredConstructor().newInstance() as Converter
+            Class.forName(materializedSink.settings["value.converter"])
+                .getDeclaredConstructor().newInstance() as Converter
         keyConverter.configure(settingsWithPrefix(materializedSink.settings, "key.converter."), true)
         valueConverter.configure(settingsWithPrefix(materializedSink.settings, "value.converter."), false)
         val localSink = floodplainSinkFromTask(task, config, keyConverter, valueConverter)
@@ -204,7 +207,6 @@ class LocalConnectorSink(
     override fun send(topic: Topic, elements: List<Pair<ByteArray?, ByteArray?>>) {
         logger.info("Inserting # of documents ${elements.size} for topic: $topic")
         val list = elements.map { (key, value) ->
-            // logger.info("Key: ${String(key?: byteArrayOf())} \nValue: ${String(value?: byteArrayOf())}")
             SinkRecord(
                 topic.qualifiedString(),
                 0,

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

@@ -23,7 +23,11 @@ import io.floodplain.reactive.source.topology.CustomTopicSource
 import io.floodplain.replication.factory.ReplicationFactory
 import org.apache.kafka.common.serialization.Serdes
 
-fun PartialStream.genericSource(topic: String, parseMessage: (ByteArray) -> IMessage?, init: Source.() -> Unit = {}): Source {
+fun PartialStream.genericSource(
+    topic: String,
+    parseMessage: (ByteArray) -> IMessage?,
+    init: Source.() -> Unit = {}
+): Source {
     return genericSource(
         topic,
         { Serdes.String().deserializer().deserialize(topic, it) },

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

@@ -57,7 +57,8 @@ class PostgresConfig(
     }
 
     private fun acceptRecord(inputReceiver: InputReceiver, record: ChangeRecord) {
-        val availableSourceTopics = sourceElements.map { sourceElement -> sourceElement.topic().qualifiedString() }.toSet()
+        val availableSourceTopics =
+            sourceElements.map { sourceElement -> sourceElement.topic().qualifiedString() }.toSet()
         if (availableSourceTopics.contains(record.topic)) {
             if (record.value != null) {
                 inputReceiver.input(record.topic, record.key.toByteArray(), record.value!!)
@@ -111,7 +112,7 @@ class PostgresConfig(
 fun Stream.postgresSourceConfig(
     name: String,
     hostname: String,
-    port: Int,
+    port: Int = 5432,
     username: String,
     password: String,
     database: String,
@@ -133,7 +134,12 @@ fun Stream.postgresSourceConfig(
     return postgresConfig
 }
 
-fun PartialStream.postgresSource(table: String, config: PostgresConfig, schema: String? = null, init: Source.() -> Unit): Source {
+fun PartialStream.postgresSource(
+    table: String,
+    config: PostgresConfig,
+    schema: String? = null,
+    init: Source.() -> Unit
+): Source {
     val effectiveSchema = schema ?: config.defaultSchema
         ?: throw IllegalArgumentException("No schema defined, and also no default schema")
     val topic = Topic.from("${config.name}.$effectiveSchema.$table", topologyContext)
@@ -165,7 +171,8 @@ fun Stream.postgresSource(name: String, table: String, schema: String, init: Sou
     addSource(databaseSource)
 }
 
-class DebeziumSourceElement(private val prefix: Topic, private val schema: String? = null, val table: String) : SourceTopic {
+class DebeziumSourceElement(private val prefix: Topic, private val schema: String? = null, val table: String) :
+    SourceTopic {
     override fun topic(): Topic {
         return prefix
     }

floodplain-dsl/src/main/kotlin/io/floodplain/kotlindsl/sink/LogSink.kt

@@ -74,7 +74,7 @@ fun PartialStream.logSink(sinkName: String, topicName: String, config: LogSinkCo
         "connector.class" to "io.floodplain.sink.LogSinkConnector",
         "tasks.max" to "1",
         "value.converter" to "org.apache.kafka.connect.json.JsonConverter",
-        "key.converter" to "org.apache.kafka.connect.storage.StringConverter", // "org.apache.kafka.connect.json.JsonConverter",
+        "key.converter" to "org.apache.kafka.connect.storage.StringConverter",
         "topics" to topicName,
         "schema.ignore" to "true",
         "type.name" to "_doc"

floodplain-dsl/src/main/kotlin/io/floodplain/kotlindsl/transformer/BufferTransformer.kt

@@ -41,7 +41,6 @@ class BufferTransformer(private val duration: Duration, private val maxSize: Int
         val top = transformerNames.peek()
         val name = topologyContext.qualifiedName("buffer", transformerNames.size, currentPipeId)
         if (materialize) {
-            // val prematerialize = topologyContext.qualifiedName("buffer-prematerialize", transformerNames.size, currentPipeId)
             addPersistentCache(
                 topology,
                 topologyContext,

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

@@ -73,7 +73,8 @@ class IMessageTest {
         return convertThereAndBack(afterJSON, protoParser)
     }
 
-    // Utility function to check if a message remains unchanged after serialization and deserialization for a specific parser
+    // Utility function to check if a message remains unchanged after serialization
+    // and deserialization for a specific parser
     private fun convertThereAndBack(input: IMessage, currentParser: ReplicationMessageParser): IMessage {
         val immutable = input.toImmutable()
         val repl = ReplicationFactory.standardMessage(immutable)

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

@@ -30,14 +30,17 @@ import io.floodplain.sink.sheet.googleSheetsSink
 import kotlinx.coroutines.delay
 import java.math.BigDecimal
 
+const val DEFAULT_POSTGRES_PORT = 5432
+const val DEFAULT_TIMEOUT = 200000L
+
 fun main() {
     val spreadsheetId = "1MTAn1d13M8ptb2MkBHOSNK1gbJOOW1sFQoSfqa1JbXU"
 
-    val instance = stream("genxx") {
+    stream("genxx") {
         val postgresConfig = postgresSourceConfig(
             "mypostgres",
             "postgres",
-            5432,
+            DEFAULT_POSTGRES_PORT,
             "postgres",
             "mysecretpassword",
             "dvdrental",
@@ -80,6 +83,6 @@ fun main() {
             )
         }
     }.renderAndExecute {
-        delay(1000000)
+        delay(DEFAULT_TIMEOUT)
     }
 }

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

@@ -27,9 +27,12 @@ import io.floodplain.mongodb.toMongo
 import java.net.URL
 
 private val logger = mu.KotlinLogging.logger {}
+
+const val DEFAULT_MYSQL_PORT = 3306
 fun main() {
     stream {
-        val mysqlConfig = mysqlSourceConfig("mysqlsource", "mysql", 3306, "root", "mysecretpassword", "wpdb")
+        val mysqlConfig =
+            mysqlSourceConfig("mysqlsource", "mysql", DEFAULT_MYSQL_PORT, "root", "mysecretpassword", "wpdb")
         val mongoConfig = mongoConfig("mongosink", "mongodb://mongo", "$generation-mongodumpalt")
         mysqlSource("wpdb.wp_posts", mysqlConfig) {
             each { key, msg, _ ->

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

@@ -59,8 +59,8 @@ class TestTopicCreation {
         logger.info("${adminClient.describeTopics(listOf("mytopic")).all().get()["mytopic"]}")
         val cr = Collections.singleton(ConfigResource(TOPIC, "mytopic"))
         val configsResult = adminClient.describeConfigs(cr)
-        // val cnf = configsResult.all().get().
-        // logger.info("Config: ${cnf.toString()}")
+        val cnf = configsResult.all().get()
+        logger.info("Config: $cnf")
         val configMap: MutableMap<String, String> = HashMap()
         configMap["cleanup.policy"] = "compact"