class Customer( val id: Long, val firstName: String, val lastName: String, val email: String, val address: String, val level: String = "bronze" // possible values: platinum, gold, silver, bronze ) enum class OrderState { CREATED, VALIDATED, FAILED, SHIPPED } enum class Product { JUMPERS, UNDERPANTS, STOCKINGS } class Order( val id: String, val customerId: Long, val state: OrderState, val product: Product, val quantity: Int, val price: Double ) class OrderValue( val order: Order, val value: Double ) class OrderEnriched( val id: String, val customerId: Long, val customerLevel: String ) enum class OrderValidationType { INVENTORY_CHECK, FRAUD_CHECK, ORDER_DETAILS_CHECK } enum class OrderValidationResult { PASS, FAIL, ERROR } class OrderValidation( val orderId: String, val checkType: OrderValidationType, val validationResult: OrderValidationResult ) class Payment( val id: String, val orderId: String, val ccy: String, val amount: Double )

fun configureSerdes() { val config = hashMapOf<String, Any>( AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG to topicsProps.schemaRegistryUrl ) for (topic in ALL.values) { configureSerde(topic.keySerde, config, true) configureSerde(topic.valueSerde, config, false) } configureSerde(ORDER_VALUE_SERDE, config, false) } fun configureSerde(serde: Serde<*>, config: Map<String, Any>, isKey: Boolean) { if (serde is SpecificAvroSerde) { serde.configure(config, isKey) } }

@EnableKafkaStreams @Configuration class KafkaConfig( private val kafkaProps: KafkaProps, private val serviceUtils: ServiceUtils, private val topicProps: TopicsProps ) { @Bean fun outstandingRequests(): MutableMap<String, FilteredResponse<String, Order, OrderDto>> { return ConcurrentHashMap() } @Bean(name = [KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME]) fun kStreamsConfig(): KafkaStreamsConfiguration { val props = hashMapOf<String, Any>( StreamsConfig.APPLICATION_ID_CONFIG to kafkaProps.appId, // must be specified to enable InteractiveQueries and checking metadata of Kafka Cluster StreamsConfig.APPLICATION_SERVER_CONFIG to serviceUtils.getServerAddress(), StreamsConfig.BOOTSTRAP_SERVERS_CONFIG to kafkaProps.bootstrapServers, StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG to Serdes.String().javaClass.name, // instances MUST have different stateDir StreamsConfig.STATE_DIR_CONFIG to kafkaProps.stateDir, ConsumerConfig.AUTO_OFFSET_RESET_CONFIG to kafkaProps.autoOffsetReset, StreamsConfig.PROCESSING_GUARANTEE_CONFIG to kafkaProps.processingGuarantee, StreamsConfig.COMMIT_INTERVAL_MS_CONFIG to kafkaProps.commitInterval, AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG to topicProps.schemaRegistryUrl, StreamsConfig.consumerPrefix(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG) to kafkaProps.sessionTimeout ) return KafkaStreamsConfiguration(props) } }

@Component class OrdersServiceTopology( private val orderProps: OrdersProps, private val mapper: OrderMapper, private val schemas: Schemas, private val outstandingRequests: MutableMap<String, FilteredResponse<String, Order, OrderDto>> ) { @Autowired fun ordersTableTopology(builder: StreamsBuilder) { log.info("Calling OrdersServiceTopology.ordersTableTopology()") builder.table( orderProps.topic, Consumed.with(schemas.ORDERS.keySerde, schemas.ORDERS.valueSerde), Materialized.`as`(orderProps.storeName) ).toStream().foreach(::maybeCompleteLongPollGet) } private fun maybeCompleteLongPollGet(id: String, order: Order) { val callback = outstandingRequests[id] if (callback?.asyncResponse?.isSetOrExpired == true) { outstandingRequests.remove(id) } else if (callback != null && callback.predicate(id, order)) { callback.asyncResponse.setResult(mapper.toDto(order)) outstandingRequests.remove(id) } } }

@GetMapping("/{id}") fun getOrder( @PathVariable("id") id: String, @RequestParam("timeout", defaultValue = "2000") timeout: Long ): DeferredResult<OrderDto> { log.info("Received request to GET order with id = {}", id) val deferredResult = DeferredResult<OrderDto>(timeout) ordersService.getOrderDto(id, deferredResult, timeout) return deferredResult }

 fun getOrderDto(id: String, asyncResponse: DeferredResult<OrderDto>, timeout: Long) { initOrderStore() CompletableFuture.runAsync { val hostForKey = getKeyLocationOrBlock(id, asyncResponse) ?: return@runAsync //request timed out so return if (thisHost(hostForKey)) { fetchLocal(id, asyncResponse) { _, _ -> true } } else { val path = Paths(hostForKey.host, hostForKey.port).urlGet(id) fetchFromOtherHost(path, asyncResponse, timeout) } } } private fun fetchLocal( id: String, asyncResponse: DeferredResult<OrderDto>, predicate: (String, Order) -> Boolean ) { log.info("running GET on this node") try { val order = orderStore.get(id) if (order == null || !predicate(id, order)) { log.info("Delaying get as order not present for id $id") outstandingRequests[id] = FilteredResponse( asyncResponse = asyncResponse, predicate = predicate ) } else { asyncResponse.setResult(mapper.toDto(order)) } } catch (e: InvalidStateStoreException) { log.error("Exception while querying local state store. {}", e.message) outstandingRequests[id] = FilteredResponse(asyncResponse, predicate) } } private fun fetchFromOtherHost( path: String, asyncResponse: DeferredResult<OrderDto>, timeout: Long ) { log.info("Chaining GET to a different instance: {}", path) try { val order = webClientBuilder.build() .get() .uri("$path?timeout=$timeout") .retrieve() .bodyToMono<OrderDto>() .block() ?: throw NotFoundException("FetchFromOtherHost returned null for request: $path") asyncResponse.setResult(order) } catch (swallowed: Exception) { log.warn("FetchFromOtherHost failed.", swallowed) } }

private lateinit var orderStore: ReadOnlyKeyValueStore<String, Order> fun initOrderStore() { if (!::orderStore.isInitialized) { val streams = factoryBean.kafkaStreams ?: throw RuntimeException("Cannot obtain KafkaStreams instance.") orderStore = streams.store( StoreQueryParameters.fromNameAndType( ordersProps.storeName, QueryableStoreTypes.keyValueStore() ) ) } }

 /**  * Find the metadata for the instance of this Kafka Streams Application that has the given  * store and would have the given key if it exists.  * @param store Store to find  * @param key The key to find  * @return {@link HostStoreInfo}  */ fun <K> streamsMetadataForStoreAndKey( store: String, key: K, serializer: Serializer<K> ): HostStoreInfo { // Get metadata for the instances of this Kafka Streams application hosting the store and // potentially the value for key val metadata = factoryBean.kafkaStreams?.queryMetadataForKey(store, key, serializer) ?: throw NotFoundException("Metadata for store $store not found!") return HostStoreInfo( host = metadata.activeHost().host(), port = metadata.activeHost().port(), storeNames = hashSetOf(store) ) }

 @Bean fun kafkaListenerContainerFactory( consumerFactory: ConsumerFactory<String, Order> ): ConcurrentKafkaListenerContainerFactory<String, Order> { val factory = ConcurrentKafkaListenerContainerFactory<String, Order>() factory.consumerFactory = consumerFactory return factory } @Bean fun consumerFactory(): ConsumerFactory<String, Order> { return DefaultKafkaConsumerFactory(consumerProps()) } private fun consumerProps(): Map<String, Any> { val props = hashMapOf<String, Any>( ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG to kafkaProps.bootstrapServers, ConsumerConfig.GROUP_ID_CONFIG to kafkaProps.appId, ConsumerConfig.AUTO_OFFSET_RESET_CONFIG to kafkaProps.autoOffsetReset, ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG to !kafkaProps.enableExactlyOnce, ConsumerConfig.CLIENT_ID_CONFIG to kafkaProps.appId, ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG to schemas.ORDERS.keySerde.deserializer()::class.java, ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG to schemas.ORDERS.valueSerde.deserializer()::class.java, AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG to kafkaProps.schemaRegistryUrl, "specific.avro.reader" to true ) return props }

 @Bean fun producerFactory(): ProducerFactory<String, OrderValidation> { return DefaultKafkaProducerFactory(senderProps()) } private fun senderProps(): Map<String, Any> { val props = hashMapOf<String, Any>( ProducerConfig.BOOTSTRAP_SERVERS_CONFIG to kafkaProps.bootstrapServers, ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG to kafkaProps.enableIdempotence, ProducerConfig.RETRIES_CONFIG to Int.MAX_VALUE.toString(), ProducerConfig.ACKS_CONFIG to kafkaProps.acks, ProducerConfig.CLIENT_ID_CONFIG to kafkaProps.appId, ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG to schemas.ORDER_VALIDATIONS.valueSerde.serializer().javaClass.name, ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG to schemas.ORDER_VALIDATIONS.keySerde.serializer().javaClass.name, AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG to kafkaProps.schemaRegistryUrl, AbstractKafkaSchemaSerDeConfig.AUTO_REGISTER_SCHEMAS to false, AbstractKafkaSchemaSerDeConfig.USE_LATEST_VERSION to true, ) if (kafkaProps.enableExactlyOnce) { // attempt to provide a unique transactional ID which is a recommended practice props[ProducerConfig.TRANSACTIONAL_ID_CONFIG] = "${kafkaProps.appId}-${applicationProps.port}" } return props } @Bean fun kafkaTemplate( producerFactory: ProducerFactory<String, OrderValidation> ): KafkaTemplate<String, OrderValidation> { return KafkaTemplate(producerFactory) }

@Service class EmailService( private val emailer: Emailer, private val schemas: Schemas ) { @Autowired fun processStreams(builder: StreamsBuilder) { val orders: KStream<String, Order> = builder.stream( schemas.ORDERS.name, Consumed.with(schemas.ORDERS.keySerde, schemas.ORDERS.valueSerde) ) //Create the streams/tables for the join val payments: KStream<String, Payment> = builder.stream( schemas.PAYMENTS.name, Consumed.with(schemas.PAYMENTS.keySerde, schemas.PAYMENTS.valueSerde) ) //Rekey payments to be by OrderId for the windowed join .selectKey { s, payment -> payment.orderId } val customers: GlobalKTable<Long, Customer> = builder.globalTable( schemas.CUSTOMERS.name, Consumed.with(schemas.CUSTOMERS.keySerde, schemas.CUSTOMERS.valueSerde) ) val serdes: StreamJoined<String, Order, Payment> = StreamJoined .with(schemas.ORDERS.keySerde, schemas.ORDERS.valueSerde, schemas.PAYMENTS.valueSerde) //Join the two streams and the table then send an email for each orders.join( payments, ::EmailTuple, JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(1)), serdes ) //Next join to the GKTable of Customers .join( customers, { _, value -> value.order.customerId }, // note how, because we use a GKtable, we can join on any attribute of the Customer. EmailTuple::setCustomer ) //Now for each tuple send an email. .peek { _, emailTuple -> emailer.sendEmail(emailTuple) } //Send the order to a topic whose name is the value of customer level orders.join( customers, { orderId, order -> order.customerId }, { order, customer -> OrderEnriched(order.id, order.customerId, customer.level) } ) //TopicNameExtractor to get the topic name (i.e., customerLevel) from the enriched order record being sent .to( TopicNameExtractor { orderId, orderEnriched, record -> orderEnriched.customerLevel }, Produced.with(schemas.ORDERS_ENRICHED.keySerde, schemas.ORDERS_ENRICHED.valueSerde) ) } }

 @Bean(name = [KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME]) fun kStreamsConfig(): KafkaStreamsConfiguration { val props = hashMapOf<String, Any>( StreamsConfig.APPLICATION_ID_CONFIG to kafkaProps.appId, // must be specified to enable InteractiveQueries and checking metadata of Kafka Cluster StreamsConfig.APPLICATION_SERVER_CONFIG to serviceUtils.getServerAddress(), StreamsConfig.BOOTSTRAP_SERVERS_CONFIG to kafkaProps.bootstrapServers, StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG to Serdes.String().javaClass.name, // instances MUST have different stateDir StreamsConfig.STATE_DIR_CONFIG to kafkaProps.stateDir, ConsumerConfig.AUTO_OFFSET_RESET_CONFIG to kafkaProps.autoOffsetReset, StreamsConfig.PROCESSING_GUARANTEE_CONFIG to kafkaProps.processingGuarantee, StreamsConfig.COMMIT_INTERVAL_MS_CONFIG to kafkaProps.commitInterval, AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG to topicProps.schemaRegistryUrl, StreamsConfig.consumerPrefix(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG) to kafkaProps.sessionTimeout, // disable caching to ensure a complete aggregate changelog. // This is a little trick we need to apply // as caching in Kafka Streams will conflate subsequent updates for the same key. // Disabling caching ensures // we get a complete "changelog" from the aggregate(...) // (i.e. every input event will have a corresponding output event. // https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=186878390 StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG to "0" ) return KafkaStreamsConfiguration(props) }

 val orders: KStream<String, Order> = builder .stream( schemas.ORDERS.name, Consumed.with(schemas.ORDERS.keySerde, schemas.ORDERS.valueSerde) ) .peek { key, value -> log.info("Processing Order {} in Fraud Service.", value) } .filter { _, order -> OrderState.CREATED == order.state } // Create an aggregate of the total value by customer and hold it with the order. // We use session windows to detect periods of activity. val aggregate: KTable<Windowed<Long>, OrderValue> = orders // creates a repartition internal topic if the value to be grouped by differs from // the key and downstream nodes need the new key .groupBy( { id, order -> order.customerId }, Grouped.with(Serdes.Long(), schemas.ORDERS.valueSerde) ) .windowedBy(SessionWindows.ofInactivityGapWithNoGrace(Duration.ofHours(1))) .aggregate( ::OrderValue, { custId, order, total -> OrderValue(order, total.value + order.quantity * order.price) }, { k, a, b -> simpleMerge(a, b) },//include a merger as we're using session windows., Materialized.with(null, schemas.ORDER_VALUE_SERDE) ) private fun simpleMerge(a: OrderValue?, b: OrderValue): OrderValue { return OrderValue(b.order, (a?.value ?: 0.0) + b.value) }

 val ordersWithTotals: KStream<String, OrderValue> = aggregate .toStream { windowedKey, orderValue -> windowedKey.key() } //When elements are evicted from a session window they create delete events. Filter these out. .filter { k, v -> v != null } .selectKey { id, orderValue -> orderValue.order.id }

 //Now branch the stream into two, for pass and fail, based on whether the windowed // total is over Fraud Limit val forks: Map<String, KStream<String, OrderValue>> = ordersWithTotals .peek { key, value -> log.info("Processing OrderValue: {} in FraudService BEFORE branching.", value) } .split(Named.`as`("limit-")) .branch( { id, orderValue -> orderValue.value >= FRAUD_LIMIT }, Branched.`as`("above") ) .branch( { id, orderValue -> orderValue.value < FRAUD_LIMIT }, Branched.`as`("below") ) .noDefaultBranch()

val keySerde = schemas.ORDER_VALIDATIONS.keySerde val valueSerde = schemas.ORDER_VALIDATIONS.valueSerde val config = hashMapOf<String, Any>( AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG to topicProps.schemaRegistryUrl, AbstractKafkaSchemaSerDeConfig.AUTO_REGISTER_SCHEMAS to false, AbstractKafkaSchemaSerDeConfig.USE_LATEST_VERSION to true, ) keySerde.configure(config, true) valueSerde.configure(config, false) forks["limit-above"]?.mapValues { orderValue -> OrderValidation( orderValue.order.id, OrderValidationType.FRAUD_CHECK, OrderValidationResult.FAIL ) }?.peek { key, value -> log.info("Sending OrderValidation for failed check in Fraud Service to Kafka. Order: {}", value) }?.to( schemas.ORDER_VALIDATIONS.name, Produced.with( keySerde, valueSerde ) ) forks["limit-below"]?.mapValues { orderValue -> OrderValidation( orderValue.order.id, OrderValidationType.FRAUD_CHECK, OrderValidationResult.PASS ) }?.peek { key, value -> log.info("Sending OrderValidation for passed check in Fraud Service to Kafka. Order: {}", value) }?.to( schemas.ORDER_VALIDATIONS.name, Produced.with( schemas.ORDER_VALIDATIONS.keySerde, schemas.ORDER_VALIDATIONS.valueSerde ) )

val orders: KStream<String, Order> = builder .stream( schemas.ORDERS.name, Consumed.with(schemas.ORDERS.keySerde, schemas.ORDERS.valueSerde) ) .peek { key, value -> log.info("Orders streams record. Key: {}, Order: {}", key, value) }

val warehouseInventory: KTable<Product, Int> = builder .table( schemas.WAREHOUSE_INVENTORY.name, Consumed.with( schemas.WAREHOUSE_INVENTORY.keySerde, schemas.WAREHOUSE_INVENTORY.valueSerde ) )

val reservedStock: StoreBuilder<KeyValueStore<Product, Long>> = Stores .keyValueStoreBuilder( Stores.persistentKeyValueStore(inventoryProps.reservedStockStoreName), schemas.WAREHOUSE_INVENTORY.keySerde, Serdes.Long() ) .withLoggingEnabled(hashMapOf()) builder.addStateStore(reservedStock)

 orders.selectKey { id, order -> order.product } .peek { key, value -> log.info("Orders stream record after SELECT KEY. New key: {}. \nNew value: {}", key, value) } // Limit to newly created orders .filter { id, order -> OrderState.CREATED == order.state } //Join Orders to Inventory so we can compare each order to its corresponding stock value .join( warehouseInventory, ::KeyValue, Joined.with( schemas.WAREHOUSE_INVENTORY.keySerde, schemas.ORDERS.valueSerde, Serdes.Integer() ) ) //Validate the order based on how much stock we have both in the warehouse // and locally 'reserved' stock .transform( TransformerSupplier { InventoryValidator(inventoryProps) }, inventoryProps.reservedStockStoreName ) .peek { key, value -> log.info( "Pushing the result of validation Order record to topic: {} with key: {}. \nResult: {}", schemas.ORDER_VALIDATIONS.name, key, value ) } //Push the result into the Order Validations topic .to( schemas.ORDER_VALIDATIONS.name, Produced.with( orderValidationsKeySerde.apply { configureSerde(this, true) }, orderValidationsValueSerde.apply { configureSerde(this, false) } ) )

private val log = LoggerFactory.getLogger(InventoryKafkaService::class.java) @Service class InventoryKafkaService( private val schemas: Schemas, private val inventoryProps: InventoryProps, private val topicProps: TopicsProps ) { @Autowired fun processStreams(builder: StreamsBuilder) { // Latch onto instances of the orders and inventory topics val orders: KStream<String, Order> = builder .stream( schemas.ORDERS.name, Consumed.with(schemas.ORDERS.keySerde, schemas.ORDERS.valueSerde) ) .peek { key, value -> log.info("Orders streams record. Key: {}, Order: {}", key, value) } val warehouseInventory: KTable<Product, Int> = builder .table( schemas.WAREHOUSE_INVENTORY.name, Consumed.with( schemas.WAREHOUSE_INVENTORY.keySerde, schemas.WAREHOUSE_INVENTORY.valueSerde ) ) // Create a store to reserve inventory whilst the order is processed. // This will be prepopulated from Kafka before the service starts processing val reservedStock: StoreBuilder<KeyValueStore<Product, Long>> = Stores .keyValueStoreBuilder( Stores.persistentKeyValueStore(inventoryProps.reservedStockStoreName), schemas.WAREHOUSE_INVENTORY.keySerde, Serdes.Long() ) .withLoggingEnabled(hashMapOf()) builder.addStateStore(reservedStock) val orderValidationsKeySerde = Serdes.String() val orderValidationsValueSerde = SpecificAvroSerde<OrderValidation>() //First change orders stream to be keyed by Product (so we can join with warehouse inventory) orders.selectKey { id, order -> order.product } .peek { key, value -> log.info("Orders stream record after SELECT KEY. New key: {}. \nNew value: {}", key, value) } // Limit to newly created orders .filter { id, order -> OrderState.CREATED == order.state } //Join Orders to Inventory so we can compare each order to its corresponding stock value .join( warehouseInventory, ::KeyValue, Joined.with( schemas.WAREHOUSE_INVENTORY.keySerde, schemas.ORDERS.valueSerde, Serdes.Integer() ) ) //Validate the order based on how much stock we have both in the warehouse // and locally 'reserved' stock .transform( TransformerSupplier { InventoryValidator(inventoryProps) }, inventoryProps.reservedStockStoreName ) .peek { key, value -> log.info( "Pushing the result of validation Order record to topic: {} with key: {}. \nResult: {}", schemas.ORDER_VALIDATIONS.name, key, value ) } //Push the result into the Order Validations topic .to( schemas.ORDER_VALIDATIONS.name, Produced.with( orderValidationsKeySerde.apply { configureSerde(this, true) }, orderValidationsValueSerde.apply { configureSerde(this, false) } ) ) } private fun configureSerde(serde: Serde<*>, isKey: Boolean) { val serdesConfig = hashMapOf<String, Any>( AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG to topicProps.schemaRegistryUrl, AbstractKafkaSchemaSerDeConfig.AUTO_REGISTER_SCHEMAS to false, AbstractKafkaSchemaSerDeConfig.USE_LATEST_VERSION to true, ) schemas.configureSerde(serde, serdesConfig, isKey) } } 

 private val CREATE_ORDERS_STREAM = "CREATE SOURCE STREAM IF NOT EXISTS ${ksqlDBProps.ordersStream} (" + "ID STRING KEY, " + "CUSTOMERID BIGINT, " + "STATE STRING, " + "PRODUCT STRING, " + "QUANTITY INT, " + "PRICE DOUBLE) WITH (" + "kafka_topic='${topicProps.ordersTopic}'," + " value_format='AVRO'" + ");" private val CREATE_CUSTOMERS_TABLE = """  CREATE SOURCE TABLE IF NOT EXISTS ${ksqlDBProps.customersTable} (  CUSTOMERID BIGINT PRIMARY KEY,   FIRSTNAME STRING,   LASTNAME STRING,   EMAIL STRING,   ADDRESS STRING,   LEVEL STRING  ) WITH (  KAFKA_TOPIC='${topicProps.customersTopic}',  VALUE_FORMAT='AVRO'  );  """ private val CREATE_ORDERS_ENRICHED_STREAM = """  CREATE STREAM IF NOT EXISTS ${ksqlDBProps.ordersEnrichedStream}   AS SELECT ${ksqlDBProps.customersTable}.CUSTOMERID AS customerId,   ${ksqlDBProps.customersTable}.FIRSTNAME,   ${ksqlDBProps.customersTable}.LASTNAME,   ${ksqlDBProps.customersTable}.LEVEL,   ${ksqlDBProps.ordersStream}.PRODUCT,   ${ksqlDBProps.ordersStream}.QUANTITY,   ${ksqlDBProps.ordersStream}.PRICE   FROM ${ksqlDBProps.ordersStream}   LEFT JOIN ${ksqlDBProps.customersTable}   ON ${ksqlDBProps.ordersStream}.CUSTOMERID = ${ksqlDBProps.customersTable}.CUSTOMERID;  """

 private val CREATE_FRAUD_ORDER_TABLE = """  CREATE TABLE IF NOT EXISTS ${ksqlDBProps.fraudOrderTable}   WITH (KEY_FORMAT='json')   AS SELECT CUSTOMERID,   LASTNAME,   FIRSTNAME,   COUNT(*) AS COUNTS   FROM ${ksqlDBProps.ordersEnrichedStream}   WINDOW TUMBLING (SIZE 30 SECONDS)   GROUP BY CUSTOMERID, LASTNAME, FIRSTNAME   HAVING COUNT(*)>2;   """

data class FraudDto( val customerId: Long, val lastName: String, val firstName: String, val counts: Long ) @RestController @RequestMapping("/fraud") class FraudController( private val fraudService: FraudService ) { @GetMapping fun getFraudulentOrders( @RequestParam(name = "limit", defaultValue = "10") limit: Int ): List<FraudDto> { log.info("Received request to GET $limit number of fraudulent orders.") return fraudService.getFraudulentOrders(limit) } } @Service class FraudServiceImpl( private val client: Client, private val ksqlDBProps: KsqlDBProps ) : FraudService { override fun getFraudulentOrders(limit: Int): List<FraudDto> { log.info("Executing getFraudulentOrders stream query.") val frauds = client.executeQuery("SELECT * FROM ${ksqlDBProps.fraudOrderTable} LIMIT $limit;").get() return frauds.map(::mapRowToResponse) } private fun mapRowToResponse(row: Row): FraudDto { log.info("Mapping ksqlDB Query row: {}", row) val cusId = row.getLong("CUSTOMERID") val lastName = row.getString("LASTNAME") val firstName = row.getString("FIRSTNAME") val count = row.getLong("COUNTS") return FraudDto(cusId, lastName, firstName, count) } }

@Service class ValidationAggregatorService( private val schemas: Schemas, private val topicProps: TopicsProps ) { private val serdes1: Consumed<String, OrderValidation> = Consumed.with( schemas.ORDER_VALIDATIONS.keySerde, schemas.ORDER_VALIDATIONS.valueSerde ) private val serdes2: Consumed<String, Order> = Consumed .with(schemas.ORDERS.keySerde, schemas.ORDERS.valueSerde) private val serdes3: Grouped<String, OrderValidation> = Grouped .with(schemas.ORDER_VALIDATIONS.keySerde, schemas.ORDER_VALIDATIONS.valueSerde) private val serdes4: StreamJoined<String, Long, Order> = StreamJoined .with(schemas.ORDERS.keySerde, Serdes.Long(), schemas.ORDERS.valueSerde) private val serdes5: Grouped<String, Order> = Grouped .with(schemas.ORDERS.keySerde, schemas.ORDERS.valueSerde) private val serdes6: StreamJoined<String, OrderValidation, Order> = StreamJoined .with(schemas.ORDERS.keySerde, schemas.ORDER_VALIDATIONS.valueSerde, schemas.ORDERS.valueSerde) // disables auto registering schemas for specified Serde<*> private fun configureSerdes(serde: Serde<*>, isKey: Boolean) { val config = hashMapOf<String, Any>( AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG to topicProps.schemaRegistryUrl, AbstractKafkaSchemaSerDeConfig.AUTO_REGISTER_SCHEMAS to false, AbstractKafkaSchemaSerDeConfig.USE_LATEST_VERSION to true, ) schemas.configureSerde(serde, config, isKey) } @Autowired fun aggregateOrderValidations(builder: StreamsBuilder) { // 1. Количество проверок, которые проходит каждый заказ. Для простоты реализации захардкожено. val numberOfRules = 3 // 2. Настройка Serdes, которые будут участвовать в отправке сообщений в топик orders.v1.  // Запрещаем автоматическую регистрацию схем в Schema Registry, чтобы избежать конфликтов. val ordersKeySerde = Serdes.String() val ordersValueSerde = SpecificAvroSerde<Order>() configureSerdes(ordersKeySerde, true) configureSerdes(ordersValueSerde, false) // 3. стрим из топика order-validations.v1 val validations: KStream<String, OrderValidation> = builder .stream(schemas.ORDER_VALIDATIONS.name, serdes1) // 4. стрим из топика orders.v1, в котором присутствуют только заказы в статусе CREATED val orders = builder .stream(schemas.ORDERS.name, serdes2) .filter { id, order -> OrderState.CREATED == order.state } validations .peek { key, value -> log.info("Starting validation for key: {}, value: {}", key, value) } // 5. группируем сообщения о валидации по orderId .groupByKey(serdes3) // 6. Сообщения также разбиваются по окнам сессионной активности с периодим неактивности  // 5 минут .windowedBy(SessionWindows.ofInactivityGapWithNoGrace(Duration.ofMinutes(5))) // 7. Аггрегируем данные о проверке заказов, если заказ прошел проверку (PASS), // тогда увеличиваем счетчик проверенных заказов  .aggregate( { 0L }, { id, result, total -> if (PASS == result.validationResult) total + 1 else total }, // так как мы используем SessionWindow, нам необходимо добавить merger для сессий // из-за особенностей работы этих окон. По сути, когда в стрим приходит первое сообщение // сразу создается сессия, после этого, если приходит еще обно сообщение, попадающее в // inactivity gap первого сообщения, то для него тоже создается сессия, которая мержится // с сессией первого сообщения. Подробнее тут: // https://kafka.apache.org/36/documentation/streams/developer-guide/dsl-api.html#windowing-session { k, a, b -> b ?: a }, Materialized.with(null, Serdes.Long()) ) // 8. избавляемся от "окон" и формируем новый стрим, где: key = orderId, value = количество пройденных проверок .toStream { windowedKey, total -> windowedKey.key() } // Когда сообщения удаляются из Session Window, публикуется событие delete (null-value), избавляемся от них .filter { _, v -> v != null } // 9. Также берем только те записи, которые прошли все проверки .filter { _, total -> total >= numberOfRules } // 9. // 10. объединяем со стримом orders и обновляем статус заказа на VALIDATED .join( orders, { id, order -> Order.newBuilder(order).setState(OrderState.VALIDATED).build() }, JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)), serdes4 ) .peek { _, value -> log.info("Order {} has been validated and is being sent to Kafka Topic: {}", value, topicProps.ordersTopic) } // 11. Отправляем заказ с обновленным статусом в топик orders.v1 .to(schemas.ORDERS.name, Produced.with(ordersKeySerde, ordersValueSerde)) // 12. также обрабатываем случай, когда заказ провалил хотябы одну проверку validations.filter { id, rule -> FAIL == rule.validationResult } // 13. объединяем со стримом orders и обновляем статус заказа на FAILED .join( orders, { id, order -> Order.newBuilder(order).setState(OrderState.FAILED).build() }, JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)), serdes6 ) .peek { key, value -> log.info("Order {} failed validation. BEFORE groupByKey.", value) } // 14. Группируем по ключу - orderId .groupByKey(serdes5) // 15. так как заказ мог не пройти несколько проверок, в нашем стриме могут присутствовать // несколько записей со статусом FAIL, тут мы избавляемся от дубликатов, так как нам // не важно сколько проверок не прошел заказ .reduce { order, v1 -> order } .toStream() .peek { key, value -> log.info("Order {} failed validation. AFTER groupByKey.", value) } .to( // 16. Отправляем заказ с обновленным статусом в топик orders.v1 schemas.ORDERS.name, Produced.with(ordersKeySerde, ordersValueSerde) ) } }