Add keytransformer (+test), upgrade some deps

Author: flyaruu

build.gradle.kts

@@ -17,8 +17,8 @@ buildscript {
 }
 plugins {
     id("eclipse")
-    id("org.jetbrains.kotlin.jvm") version "1.4.20"
-    id("org.jetbrains.kotlin.plugin.allopen") version "1.4.20"
+    id("org.jetbrains.kotlin.jvm") version "1.4.31"
+    id("org.jetbrains.kotlin.plugin.allopen") version "1.4.31"
     id("com.github.johnrengelman.shadow") version "5.2.0"
     id("org.jlleitschuh.gradle.ktlint") version "9.4.1"
     id("org.jetbrains.dokka") version "0.10.1"

buildSrc/build.gradle.kts

@@ -1,6 +1,6 @@
 plugins {
     `kotlin-dsl`
-    id("org.jetbrains.kotlin.jvm").version("1.4.20")
+    id("org.jetbrains.kotlin.jvm").version("1.4.31")
     id("org.jetbrains.dokka").version("1.4.10.2")
     id("io.gitlab.arturbosch.detekt").version ("1.14.1")
 

buildSrc/src/main/kotlin/Dependencies.kt

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

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

@@ -33,6 +33,7 @@ import io.floodplain.reactive.source.topology.FilterTransformer
 import io.floodplain.reactive.source.topology.GroupTransformer
 import io.floodplain.reactive.source.topology.JoinRemoteTransformer
 import io.floodplain.reactive.source.topology.JoinWithTransformer
+import io.floodplain.reactive.source.topology.KeyTransformer
 import io.floodplain.reactive.source.topology.ScanTransformer
 import io.floodplain.reactive.source.topology.SetTransformer
 import io.floodplain.reactive.source.topology.SinkTransformer
@@ -184,6 +185,10 @@ fun PartialStream.set(transform: (String, IMessage, IMessage) -> IMessage): Tran
     return addTransformer(Transformer(this.rootTopology, set, topologyContext))
 }
 
+fun PartialStream.keyTransform(transform: (String) -> String): Transformer {
+    return addTransformer(Transformer(this.rootTopology, KeyTransformer(transform), topologyContext))
+}
+
 // fun PartialPipe.copy()
 /**
  * Join the current source with another source, that has a different key.

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

@@ -28,6 +28,7 @@ import io.floodplain.streams.debezium.JSONToReplicationMessage
 import io.floodplain.streams.remotejoin.StoreStateProcessor
 import kotlinx.coroutines.delay
 import org.apache.kafka.streams.state.KeyValueStore
+import org.junit.Assert
 import java.math.BigDecimal
 import java.time.Duration
 import java.util.Optional
@@ -579,6 +580,23 @@ class TestTopology {
     }
 
     @Test
+    fun testKeyTransformer() {
+        stream {
+            from("source") {
+                keyTransform { key->"mon${key}" }
+                sinkQualified("output")
+            }
+        }.renderAndExecute {
+            inputQualified("source", "key", empty().set("value", "value1"))
+            val (key, _) = outputQualified("output")
+            Assert.assertEquals("monkey",key)
+            deleteQualified("source","soon")
+            val deleted = deletedQualified("output")
+            Assert.assertEquals("monsoon",deleted)
+        }
+    }
+
+        @Test
     fun testDiff() {
         stream {
             source("@source") {

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

@@ -38,7 +38,6 @@ import kotlin.test.assertEquals
 
 private val logger = mu.KotlinLogging.logger {}
 
-@kotlinx.coroutines.ExperimentalCoroutinesApi
 class FilmToMongoIntegratedSink {
 
     private val containerNetwork = Network.newNetwork()

floodplain-stream-topology/src/main/java/io/floodplain/reactive/source/topology/KeyTransformer.java

@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package io.floodplain.reactive.source.topology;
+
+import io.floodplain.reactive.source.topology.api.TopologyPipeComponent;
+import io.floodplain.streams.api.TopologyContext;
+import io.floodplain.streams.remotejoin.KeyProcessor;
+import io.floodplain.streams.remotejoin.ReplicationTopologyParser;
+import io.floodplain.streams.remotejoin.TopologyConstructor;
+import org.apache.kafka.streams.Topology;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Stack;
+import java.util.function.Function;
+
+public class KeyTransformer implements TopologyPipeComponent {
+
+    private boolean materialize;
+
+    //	private final boolean fromEmpty;
+    private final Function<String,String> keyTransformer;
+
+
+    private final static Logger logger = LoggerFactory.getLogger(KeyTransformer.class);
+    public KeyTransformer(Function<String, String> keyTransformer) {
+        this.keyTransformer = keyTransformer;
+    }
+
+
+    @Override
+    public void addToTopology(Stack<String> transformerNames, int currentPipeId, Topology topology,
+                              TopologyContext topologyContext, TopologyConstructor topologyConstructor) {
+        KeyProcessor fp = new KeyProcessor(this.keyTransformer);
+        String name = topologyContext.qualifiedName("keyTransform", transformerNames.size(), currentPipeId);
+        logger.info("Adding processor: {} to parent: {} hash: {}", name, transformerNames, transformerNames.hashCode());
+        if (this.materialize) {
+            topology.addProcessor(name + "_prematerialize", () -> fp, transformerNames.peek());
+            ReplicationTopologyParser.addMaterializeStore(topology, topologyContext, topologyConstructor, name, name + "_prematerialize");
+        } else {
+            topology.addProcessor(name, () -> fp, transformerNames.peek());
+        }
+        transformerNames.push(name);
+    }
+
+    @Override
+    public boolean materializeParent() {
+        return false;
+    }
+
+    @Override
+    public void setMaterialize() {
+        this.materialize = true;
+    }
+
+
+}

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

@@ -52,7 +52,7 @@ class InstantiatedContainer(image: String, port: Int, env: Map<String, String> =
 // KafkaContainer("5.5.3").withEmbeddedZookeeper().withExposedPorts(9092,9093)
 class InstantiatedKafkaContainer(customizer: ((KafkaContainer) -> KafkaContainer)? = null) {
     // class KGenericContainer(imageName: String) : GenericContainer<KGenericContainer>(DockerImageName.parse(imageName))
-    val container = KafkaContainer(DockerImageName("confluentinc/cp-kafka:6.0.2"))
+    val container = KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:6.0.2"))
         .apply { withExposedPorts(9092, 9093) }
         .apply { withStartupTimeout(Duration.ofMinutes(5)) }
         .apply { withStartupAttempts(50) }
@@ -61,11 +61,11 @@ class InstantiatedKafkaContainer(customizer: ((KafkaContainer) -> KafkaContainer
     var host: String
     var exposedPort: Int = -1
     init {
-        container?.start()
-        host = container?.host ?: "localhost"
-        exposedPort = container?.getMappedPort(9093) ?: -1
+        container.start()
+        host = container.host ?: "localhost"
+        exposedPort = container.getMappedPort(9093) ?: -1
     }
     fun close() {
-        container?.close()
+        container.close()
     }
 }

gradle.properties

@@ -1,4 +1,4 @@
 #Gradle properties
 systemProp.org.gradle.internal.publish.checksums.insecure=true
-kotlinVersion=1.4.20
+kotlinVersion=1.4.31
 org.gradle.jvmargs=-Xmx1024m

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

@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package io.floodplain.streams.remotejoin;
+
+import io.floodplain.replication.api.ReplicationMessage;
+import org.apache.kafka.streams.processor.AbstractProcessor;
+
+import java.util.function.Function;
+
+/**
+ *
+ */
+public class KeyProcessor extends AbstractProcessor<String, ReplicationMessage> {
+
+    private final Function<String, String> keyTransform;
+
+    public KeyProcessor(Function<String,String> keyTransform) {
+        this.keyTransform = keyTransform;
+    }
+    @Override
+    public void process(String key, ReplicationMessage value) {
+        super.context().forward(keyTransform.apply(key), value);
+    }
+
+}

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

@@ -97,6 +97,15 @@ public static void materializeStateStores(TopologyConstructor topologyConstructo
         }
     }
 
+    public static void addKeyTransformProcessor(Topology current, TopologyContext topologyContext,
+                                        TopologyConstructor topologyConstructor, String fromProcessor,
+                                        Function<String,String> keyTransformer) {
+        // TODO I think the topologyContext should be used.
+//        current.addProcessor(diffProcessorNamePrefix, () -> new DiffProcessor(diffProcessorNamePrefix), fromProcessor);
+//        addStateStoreMapping(topologyConstructor.processorStateStoreMapper, diffProcessorNamePrefix, diffProcessorNamePrefix);
+//        logger.info("Granting access for processor: {} to store: {}", diffProcessorNamePrefix, diffProcessorNamePrefix);
+//        topologyConstructor.stateStoreSupplier.put(diffProcessorNamePrefix, createMessageStoreSupplier(diffProcessorNamePrefix, true));
+    }
 
     // I think we need to use the topologyContext
     public static void addDiffProcessor(Topology current, TopologyContext topologyContext,