Add in rekey with processor api into test case

perkss · perkss · commit cffce15732e3 · 2021-05-23T15:57:59.000+01:00
diff --git a/kotlin-kafka-streams-examples/src/main/kotlin/com/perkss/kafka/reactive/AppConfig.kt b/kotlin-kafka-streams-examples/src/main/kotlin/com/perkss/kafka/reactive/AppConfig.kt
@@ -119,17 +119,5 @@ class AppConfig {
         orderProcessingTopology: Topology,
         streamConfig: Properties
     ) = KafkaStreams(orderProcessingTopology, streamConfig)
-//
-//    @Bean
-//    fun bootstrapSemantics(
-//        streamsBuilder: StreamsBuilder,
-//        streamConfig: Properties
-//    ) = KafkaStreams(BootstrapSemanticsSelfJoinTopology.build(streamsBuilder, streamConfig), streamConfig)
-
-//    @Bean
-//    fun bootstrapSemantics(
-//        streamsBuilder: StreamsBuilder,
-//        streamConfig: Properties
-//    ) = KafkaStreams(BootstrapSemanticsJoinOtherTableTopology.build(streamsBuilder, streamConfig), streamConfig)
 
 }
diff --git a/kotlin-kafka-streams-examples/src/main/kotlin/com/perkss/kafka/reactive/KafkaStreamsApp.kt b/kotlin-kafka-streams-examples/src/main/kotlin/com/perkss/kafka/reactive/KafkaStreamsApp.kt
@@ -9,7 +9,6 @@ import org.springframework.boot.runApplication
 @SpringBootApplication
 class KafkaStreamsApp(
     private val orderProcessingApp: KafkaStreams,
-    //private val bootstrapSemantics: KafkaStreams
 ) : CommandLineRunner {
 
     companion object {
@@ -19,7 +18,6 @@ class KafkaStreamsApp(
     override fun run(vararg args: String) {
         logger.info("Running Kotlin Kakfa Streams")
         orderProcessingApp.start()
-        //bootstrapSemantics.start()
     }
 }
 
diff --git a/kotlin-kafka-streams-examples/src/main/kotlin/com/perkss/kafka/reactive/OrderProcessingTopology.kt b/kotlin-kafka-streams-examples/src/main/kotlin/com/perkss/kafka/reactive/OrderProcessingTopology.kt
@@ -1,5 +1,6 @@
 package com.perkss.kafka.reactive
 
+import com.perkss.kafka.reactive.processor.RekeyStream
 import com.perkss.order.model.OrderConfirmed
 import com.perkss.order.model.OrderRejected
 import com.perkss.order.model.OrderRequested
@@ -22,7 +23,6 @@ object OrderProcessingTopology {
               stockSerde: SpecificAvroSerde<Stock>,
               props: AppProperties): KTable<String, Stock> {
         return streamsBuilder.table(props.stockInventory, Consumed.with(Serdes.String(), stockSerde))
-        //, Materialized.`as`(props.stockInventory)
     }
 
 
@@ -44,13 +44,18 @@ object OrderProcessingTopology {
                         orderRejectedSerde: SpecificAvroSerde<OrderRejected>,
                         orderConfirmedSerde: SpecificAvroSerde<OrderConfirmed>,
                         stockSerde: SpecificAvroSerde<Stock>): Topology {
+
+        val rekeyStream = { RekeyStream<String, OrderRequested> { it.productId } }
+
         val split = streamsBuilder
-                .stream(props.orderRequest, Consumed.with(keySerde, orderRequestedSerde))
-                .peek { key, value -> logger.info("Consumed {} {}", key, value) }
-                // Rekey to the Product ID so we can join with Product Table
-                .selectKey { _, value -> value.productId }
-                .peek { key, value -> logger.info("Rekeyed to Product Id {} {}", key, value) }
-                .leftJoin(stockTable, { orderRequest: OrderRequested, stock: Stock? ->
+            .stream(props.orderRequest, Consumed.with(keySerde, orderRequestedSerde))
+            .peek { key, value -> logger.info("Consumed {} {}", key, value) }
+            // Rekey to the Product ID so we can join with Product Table
+            //.selectKey { _, value -> value.productId }
+            // Use the Processor API to rekey so we do dont create another topic
+            .transform(rekeyStream)
+            .peek { key, value -> logger.info("Rekeyed to Product Id {} {}", key, value) }
+            .leftJoin(stockTable, { orderRequest: OrderRequested, stock: Stock? ->
                     if (stock != null && stock.quantityAvailable > 0) {
                         orderRequest
                     } else {
diff --git a/kotlin-kafka-streams-examples/src/test/kotlin/com/perkss/kafka/reactive/integration/StreamIntegrationTest.kt b/kotlin-kafka-streams-examples/src/test/kotlin/com/perkss/kafka/reactive/integration/StreamIntegrationTest.kt
@@ -80,9 +80,11 @@ internal class StreamIntegrationTest @Autowired constructor(private var appPrope
             val kafkaAdminClient: AdminClient = KafkaAdminClient.create(props)
             val result: CreateTopicsResult = kafkaAdminClient
                     .createTopics(
-                            listOf("order-request", "order-processed",
-                                    "stock", "customer", "order-rejected")
-                                    .map { name -> NewTopic(name, 3, 1.toShort()) }
+                            listOf(
+                                "order-request", "order-processed",
+                                "stock", "customer", "order-rejected"
+                            )
+                                .map { name -> NewTopic(name, 12, 1.toShort()) }
                                     .toList())
             logger.info("Topics created {}", result.all().get())
 
@@ -102,45 +104,62 @@ internal class StreamIntegrationTest @Autowired constructor(private var appPrope
             put(ConsumerConfig.GROUP_ID_CONFIG, "test-consumer")
             put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
             put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.bootstrapServers)
-            put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer::class.java)
-            put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, io.confluent.kafka.serializers.KafkaAvroDeserializer::class.java)
+            put(
+                ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
+                org.apache.kafka.common.serialization.StringDeserializer::class.java
+            )
+            put(
+                ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
+                io.confluent.kafka.serializers.KafkaAvroDeserializer::class.java
+            )
             put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, true)
-            put(KafkaAvroDeserializerConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://${schemaRegistry.containerIpAddress}:${schemaRegistry.getMappedPort(8081)}")
+            put(
+                KafkaAvroDeserializerConfig.SCHEMA_REGISTRY_URL_CONFIG,
+                "http://${schemaRegistry.containerIpAddress}:${schemaRegistry.getMappedPort(8081)}"
+            )
         }
 
 
         val testProducer = KafkaProducer<String, Any>(producerProps)
         val testConsumer = KafkaConsumer<String, OrderConfirmed>(consumerProps)
 
-        val orderId = UUID.randomUUID().toString()
-        val productId = UUID.randomUUID().toString()
-        val customerId = UUID.randomUUID().toString()
+        val orderId = "e125125125gasfaseegakjgsaka"
+        val productId = "e124121ffgaavasraseveas"
+        val customerId = "f124f25125212156"
         val value = OrderRequested(orderId, productId, customerId)
 
         // Populate the stock table and the customer table
         testProducer.send(
-                ProducerRecord(appProperties.customerInformation, customerId,
-                        Customer(customerId, "perkss", "london").toGenericRecord(SchemaLoader.loadSchema())))
+            ProducerRecord(
+                appProperties.customerInformation, customerId,
+                Customer(customerId, "perkss", "london").toGenericRecord(SchemaLoader.loadSchema())
+            )
+        )
 
         testProducer.send(
-                ProducerRecord(appProperties.stockInventory, productId, Stock(productId, "Party Poppers", 5)))
+            ProducerRecord(appProperties.stockInventory, productId, Stock(productId, "Party Poppers", 5))
+        )
 
         // send Order Request message to topology
         testProducer.send(
-                ProducerRecord(appProperties.orderRequest, orderId, value))
+            ProducerRecord(appProperties.orderRequest, orderId, value)
+        )
 
         testConsumer.subscribe(listOf(appProperties.orderProcessedTopic))
 
-        val actual = mutableMapOf<String, OrderConfirmed>()
-        val expected = mapOf(orderId to OrderConfirmed(orderId, productId, customerId, true))
+        val actual = mutableMapOf<String, Pair<Int, OrderConfirmed>>()
+        val expectedPartition = 3
+        val expected = mapOf(orderId to (expectedPartition to OrderConfirmed(orderId, productId, customerId, true)))
 
-        val timeout = System.currentTimeMillis() + 60000L;
-        while (actual != expected && System.currentTimeMillis() < timeout) {
-            val records: ConsumerRecords<String, OrderConfirmed> = testConsumer.poll(Duration.ofSeconds(1))
-            records.forEach { record -> actual[record.key()] = record.value() }
+        val timeout = System.currentTimeMillis() + 60000L
+        while (actual.isEmpty() && System.currentTimeMillis() < timeout) {
+            val records: ConsumerRecords<String, OrderConfirmed> = testConsumer.poll(Duration.ofSeconds(5))
+            assertEquals(1, records.count())
+            records.forEach { record -> actual[record.key()] = record.partition() to record.value() }
         }
 
         assertEquals(expected[orderId], actual[orderId])
+        assertEquals(expectedPartition, actual[orderId]?.first)
         orderProcessingApp.close()
     }
 

Original file line number	Diff line number	Diff line change
`@@ -9,7 +9,6 @@ import org.springframework.boot.runApplication`
`9`	`9`	`@SpringBootApplication`
`10`	`10`	`class KafkaStreamsApp(`
`11`	`11`	`private val orderProcessingApp: KafkaStreams,`
`12`		`- //private val bootstrapSemantics: KafkaStreams`
`13`	`12`	`) : CommandLineRunner {`
`14`	`13`
`15`	`14`	`companion object {`
`@@ -19,7 +18,6 @@ class KafkaStreamsApp(`
`19`	`18`	`override fun run(vararg args: String) {`
`20`	`19`	`logger.info("Running Kotlin Kakfa Streams")`
`21`	`20`	`orderProcessingApp.start()`
`22`		`- //bootstrapSemantics.start()`
`23`	`21`	`}`
`24`	`22`	`}`
`25`	`23`