Record Router Streams actual implementation.

* Tested with minimal tests.
* Needs better testing.

Author: mjbmarques

code/kafka-streams-module-demo/src/main/kotlin/com/isel/kafkastreamsmoduledemo/recordRouter/RRTesting.kt

@@ -33,25 +33,29 @@ class RRTesting(
     }
 
     private val mapper = jacksonObjectMapper()
-    private final val consumerExecutor: ExecutorService = Executors.newFixedThreadPool(5)
+    private final val consumerExecutor: ExecutorService = Executors.newFixedThreadPool(20)
 
     fun test() {
 
 
         val systemTopic: String = "SYSTEM_TOPIC"
         val inputTopicA: String = "input-topic-a"
+        val inputTopicB: String = "input-topic-b"
 
         runConsumer(systemTopic)
         runConsumer(inputTopicA)
+        runConsumer(inputTopicB)
         runConsumer("gateway-01-clients-topic")
         runConsumer("gateway-01-keys-topic")
+        runConsumer("gateway-02-clients-topic")
+        runConsumer("gateway-02-keys-topic")
         val props = Properties()
         props["bootstrap.servers"] = bootstrapServers
         props["key.serializer"] = "org.apache.kafka.common.serialization.StringSerializer"
         props["value.serializer"] = "org.apache.kafka.common.serialization.StringSerializer"
         val producer: KafkaProducer<String, String> = KafkaProducer<String, String>(props)
 
-        val newGatewayKeyTopic = SystemGatewayKeyTopic("id", "gateway-01-keys-topic", "gateway-01-clients-topic")
+        val newGatewayKeyTopic = SystemGatewayKeyTopic("id1", "gateway-01-keys-topic", "gateway-01-clients-topic")
         val newGatewayKeyTopicJson: String = mapper.writeValueAsString(newGatewayKeyTopic)
         utils.printRed("TEST newGatewayKeyTopicJson as string: \n $newGatewayKeyTopicJson")
 
@@ -69,8 +73,29 @@ class RRTesting(
         producer.send(ProducerRecord(inputTopicA, "0", "testvalue 3"))
         producer.send(ProducerRecord(inputTopicA, "0", "testvalue 4"))
 
+        Thread.sleep(6000)
 
+        val newGatewayKeyTopic2 = SystemGatewayKeyTopic("id2", "gateway-02-keys-topic", "gateway-02-clients-topic")
+        val newGatewayKeyTopicJson2: String = mapper.writeValueAsString(newGatewayKeyTopic2)
+        producer.send(ProducerRecord(systemTopic, "new-gateway-key-topic", newGatewayKeyTopicJson2))
 
+        Thread.sleep(1000)
+
+        producer.send(ProducerRecord(newGatewayKeyTopic2.keysTopicName, inputTopicA, mapper.writeValueAsString(listOf("1", "2"))))
+        producer.send(ProducerRecord(newGatewayKeyTopic2.keysTopicName, inputTopicB, mapper.writeValueAsString(listOf("5", "6"))))
+
+        Thread.sleep(2000)
+
+        producer.send(ProducerRecord(inputTopicA, "0", "testvalue 10"))
+        producer.send(ProducerRecord(inputTopicA, "0", "testvalue 11"))
+        producer.send(ProducerRecord(inputTopicA, "1", "testvalue 12"))
+        producer.send(ProducerRecord(inputTopicA, "1", "testvalue 13"))
+        producer.send(ProducerRecord(inputTopicA, "2", "testvalue 14"))
+
+        producer.send(ProducerRecord(inputTopicB, "1", "no 13"))
+        producer.send(ProducerRecord(inputTopicB, "2", "no 14"))
+        producer.send(ProducerRecord(inputTopicB, "5", "yes 13"))
+        producer.send(ProducerRecord(inputTopicB, "6", "yes 14"))
 
     }
 

code/kafka-streams-module-demo/src/main/kotlin/com/isel/kafkastreamsmoduledemo/recordRouter/Streams.kt

@@ -42,15 +42,22 @@ class Streams(
 
         val systemTopic: String = "SYSTEM_TOPIC"
         val inputTopicA: String = "input-topic-a"
+        val inputTopicB: String = "input-topic-b"
 
         rrTesting.deleteTopic(systemTopic)
         rrTesting.deleteTopic(inputTopicA)
+        rrTesting.deleteTopic(inputTopicB)
         rrTesting.deleteTopic("gateway-01-clients-topic")
         rrTesting.deleteTopic("gateway-01-keys-topic")
+        rrTesting.deleteTopic("gateway-02-clients-topic")
+        rrTesting.deleteTopic("gateway-02-keys-topic")
         rrTesting.createTopic(systemTopic,3,3)
         rrTesting.createTopic(inputTopicA,3,3)
+        rrTesting.createTopic(inputTopicB,3,3)
         rrTesting.createTopic("gateway-01-clients-topic",3,3)
         rrTesting.createTopic("gateway-01-keys-topic",3,3)
+        rrTesting.createTopic("gateway-02-clients-topic",3,3)
+        rrTesting.createTopic("gateway-02-keys-topic",3,3)
 
         listenSystemTopic()
     }
@@ -75,7 +82,7 @@ class Streams(
                                 val systemGatewayKeyTopic: SystemGatewayKeyTopic =
                                     mapper.readValue<SystemGatewayKeyTopic>(record.value())
                                 updateGatewayKeysTopicsSubscriptions(systemGatewayKeyTopic)
-                                //streamsStorage.restartAllDefaultRoutingStreams()
+                                kClientsStorage.restartAllDefaultRoutingStreams()
                             }
 
                             else -> {
@@ -171,8 +178,12 @@ class Streams(
             consumer = createGatewayKeysConsumer()
             kClientsStorage.setGatewayKeysConsumer(consumer) {createGatewayKeysConsumer()}
         } else {
+            recordRouterUtils.printRed("updateGatewayKeysTopicsSubscriptions before wakeup")
             consumer.wakeup()
+            consumer = createGatewayKeysConsumer() // TODO: Temporary
+            kClientsStorage.setGatewayKeysConsumer(consumer) {createGatewayKeysConsumer()} // TODO: Temporary
         }
+        recordRouterUtils.printRed("updateGatewayKeysTopicsSubscriptions after consumer==null if-else")
 
 
         gatewaysDetails.compute(systemGatewayKeyTopic.gatewayId) { gatewayId, details ->