{"id":360983,"date":"2024-05-21T01:04:08","date_gmt":"2024-05-21T01:04:08","guid":{"rendered":"http:\/\/savepearlharbor.com\/?p=360983"},"modified":"-0001-11-30T00:00:00","modified_gmt":"2024-05-21T01:04:08","slug":"","status":"publish","type":"post","link":"https:\/\/savepearlharbor.com\/?p=360983","title":{"rendered":"<span>\u041c\u0438\u043a\u0440\u043e\u0441\u0435\u0440\u0432\u0438\u0441\u044b \u043d\u0430 \u043e\u0441\u043d\u043e\u0432\u0435 \u0441\u043e\u0431\u044b\u0442\u0438\u0439 \u0441 Kafka Streams \u0438 Spring Boot<\/span>"},"content":{"rendered":"<div><!--[--><!--]--><\/div>\n<div id=\"post-content-body\">\n<div>\n<div class=\"article-formatted-body article-formatted-body article-formatted-body_version-2\">\n<div xmlns=\"http:\/\/www.w3.org\/1999\/xhtml\">\n<h3>\u041f\u0440\u0435\u0434\u0438\u0441\u043b\u043e\u0432\u0438\u0435<\/h3>\n<p>\u0412\u0441\u0435\u043c \u043f\u0440\u0438\u0432\u0435\u0442! \u042d\u0442\u043e \u043c\u043e\u044f \u043f\u0435\u0440\u0432\u0430\u044f \u0441\u0442\u0430\u0442\u044c\u044f \u043d\u0430 Habr, \u0442\u0430\u043a \u0447\u0442\u043e \u043d\u0435 \u0441\u0443\u0434\u0438\u0442\u0435 \u0441\u0442\u0440\u043e\u0433\u043e \u0437\u0430 \u0441\u0442\u0438\u043b\u044c, \u0430 \u0432\u043e\u0442 \u043f\u043e \u0441\u043e\u0434\u0435\u0440\u0436\u0430\u043d\u0438\u044e \u0431\u0443\u0434\u0443 \u0440\u0430\u0434 \u043b\u044e\u0431\u043e\u0439 \u043a\u043e\u043d\u0441\u0442\u0440\u0443\u043a\u0442\u0438\u0432\u043d\u043e\u0439 \u043a\u0440\u0438\u0442\u0438\u043a\u0435. <\/p>\n<p>\u0421\u0442\u0430\u0442\u044c\u044f \u043e\u0441\u043d\u043e\u0432\u0430\u043d\u0430 \u043d\u0430 \u043f\u0443\u0431\u043b\u0438\u043a\u0430\u0446\u0438\u044f\u0445 Confluent \u043e \u0442\u043e\u043c, \u043a\u0430\u043a \u043c\u043e\u0436\u043d\u043e \u0440\u0435\u0430\u043b\u0438\u0437\u043e\u0432\u0430\u0442\u044c \u043c\u0438\u043a\u0440\u043e\u0441\u0435\u0440\u0432\u0438\u0441\u043d\u0443\u044e \u0430\u0440\u0445\u0438\u0442\u0435\u043a\u0442\u0443\u0440\u0443 \u043d\u0430 \u043e\u0441\u043d\u043e\u0432\u0435 \u0441\u043e\u0431\u044b\u0442\u0438\u0439 \u0441 \u043f\u043e\u043c\u043e\u0449\u044c\u044e Kafka Streams \u0438 KSQL. <\/p>\n<ol>\n<li>\n<p><a href=\"https:\/\/docs.confluent.io\/platform\/current\/tutorials\/examples\/microservices-orders\/docs\/index.html\" rel=\"noopener noreferrer nofollow\">Introduction to Streaming Application Development. <\/a><\/p>\n<\/li>\n<li>\n<p><a href=\"https:\/\/www.confluent.io\/blog\/building-a-microservices-ecosystem-with-kafka-streams-and-ksql\/\" rel=\"noopener noreferrer nofollow\">Building a Microservices Ecosystem with Kafka Streams and KSQL.<\/a><\/p>\n<\/li>\n<\/ol>\n<p>\u0418\u0441\u0445\u043e\u0434\u043d\u044b\u0439 \u043a\u043e\u0434 \u0441 \u043f\u0440\u0438\u043c\u0435\u0440\u0430\u043c\u0438 \u0432 \u0441\u0442\u0430\u0442\u044c\u044f\u0445 \u043d\u0430\u043f\u0438\u0441\u0430\u043d \u043d\u0430 Java \u0431\u0435\u0437 \u043f\u0440\u0438\u043c\u0435\u043d\u0435\u043d\u0438\u044f \u043a\u0430\u043a\u0438\u0445-\u043b\u0438\u0431\u043e \u0444\u0440\u0435\u0439\u043c\u0432\u043e\u0440\u043a\u043e\u0432, \u0430 \u0434\u043b\u044f \u0437\u0430\u043f\u0443\u0441\u043a\u0430 \u0432\u0441\u0435\u0445 \u043c\u0438\u043a\u0440\u043e\u0441\u0435\u0440\u0432\u0438\u0441\u043e\u0432 \u0442\u0440\u0435\u0431\u0443\u0435\u0442\u0441\u044f \u0430\u043a\u043a\u0430\u0443\u043d\u0442 Confluent Cloud. \u042f \u0440\u0435\u0448\u0438\u043b \u043f\u0435\u0440\u0435\u043f\u0438\u0441\u0430\u0442\u044c \u0432\u0441\u0435 \u0438\u0441\u0445\u043e\u0434\u043d\u0438\u043a\u0438 \u043d\u0430 Kotlin + Spring Boot for Apache Kafka. \u041c\u043e\u0439 \u043a\u043e\u0434 \u0434\u043e\u0441\u0442\u0443\u043f\u0435\u043d \u043d\u0430 <a href=\"https:\/\/github.com\/aasmc\/event-driven-microservices-kafka\" rel=\"noopener noreferrer nofollow\">Github<\/a>. <\/p>\n<h3>\u0412\u0435\u0440\u0445\u043d\u0435\u0443\u0440\u043e\u0432\u043d\u0435\u0432\u043e\u0435 \u043e\u043f\u0438\u0441\u0430\u043d\u0438\u0435 \u043f\u0440\u043e\u0435\u043a\u0442\u0430<\/h3>\n<figure class=\"full-width\"><img loading=\"lazy\" decoding=\"async\" src=\"https:\/\/habrastorage.org\/r\/w1560\/getpro\/habr\/upload_files\/9fb\/a54\/575\/9fba545759d486d491f7e75943d3fd7b.png\" width=\"1598\" height=\"1188\" data-src=\"https:\/\/habrastorage.org\/getpro\/habr\/upload_files\/9fb\/a54\/575\/9fba545759d486d491f7e75943d3fd7b.png\"\/><\/figure>\n<p>\u0426\u0435\u043d\u0442\u0440\u043e\u043c \u0441\u0438\u0441\u0442\u0435\u043c\u044b \u043c\u0438\u043a\u0440\u043e\u0441\u0435\u0440\u0432\u0438\u0441\u043e\u0432 \u044f\u0432\u043b\u044f\u0435\u0442\u0441\u044f Orders Service, \u043a\u043e\u0442\u043e\u0440\u044b\u0439 \u043f\u0440\u0435\u0434\u043e\u0441\u0442\u0430\u0432\u043b\u044f\u0435\u0442 \u043a\u043b\u0438\u0435\u043d\u0442\u0430\u043c \u0432\u043e\u0437\u043c\u043e\u0436\u043d\u043e\u0441\u0442\u044c \u043f\u043e\u0441\u044b\u043b\u0430\u0442\u044c REST \u0437\u0430\u043f\u0440\u043e\u0441\u044b \u043d\u0430:<\/p>\n<ul>\n<li>\n<p>\u0441\u043e\u0434\u0430\u043d\u0438\u0435 \u0437\u0430\u043a\u0430\u0437\u043e\u0432 (POST <a href=\"http:\/\/localhost:8000\/v1\/orders\" rel=\"noopener noreferrer nofollow\">http:\/\/localhost:8000\/v1\/orders<\/a>)<\/p>\n<\/li>\n<li>\n<p>\u043f\u043e\u043b\u0443\u0447\u0435\u043d\u0438\u0435 \u0437\u0430\u043a\u0430\u0437\u0430 \u043f\u043e \u0435\u0433\u043e ID (GET <a href=\"http:\/\/localhost:8000\/v1\/orders\/%7Bid%7D\" rel=\"noopener noreferrer nofollow\">http:\/\/localhost:8000\/v1\/orders\/{id}<\/a>)<\/p>\n<\/li>\n<li>\n<p>\u043f\u043e\u043b\u0443\u0447\u0435\u043d\u0438\u0435 \u0432\u0430\u043b\u0438\u0434\u0438\u0440\u043e\u0432\u0430\u043d\u043d\u043e\u0433\u043e \u0437\u0430\u043a\u0430\u0437\u0430 \u043f\u043e \u0435\u0433\u043e ID (GET <a href=\"http:\/\/localhost:8000\/v1\/orders\/%7Bid%7D\/validated\" rel=\"noopener noreferrer nofollow\">http:\/\/localhost:8000\/v1\/orders\/{id}\/validated<\/a>)<\/p>\n<\/li>\n<\/ul>\n<p>\u041f\u0440\u0438 \u0441\u043e\u0437\u0434\u0430\u043d\u0438\u0438 \u0437\u0430\u043a\u0430\u0437\u0430 \u0432 Kafka \u0442\u043e\u043f\u0438\u043a <code>orders.v1<\/code> \u043f\u043e\u0441\u044b\u043b\u0430\u0435\u0442\u0441\u044f \u0441\u043e\u043e\u0442\u0432\u0435\u0442\u0441\u0442\u0432\u0443\u044e\u0449\u0435\u0435 \u0441\u043e\u0431\u044b\u0442\u0438\u0435, \u043e\u043d\u043e \u0432\u044b\u0447\u0438\u0442\u044b\u0432\u0430\u0435\u0442\u0441\u044f \u0440\u0430\u0437\u043b\u0438\u0447\u043d\u044b\u043c\u0438 \u0441\u0435\u0440\u0432\u0438\u0441\u0430\u043c\u0438, \u043e\u0442\u0432\u0435\u0447\u0430\u044e\u0449\u0438\u043c\u0438 \u0437\u0430 \u0432\u0430\u043b\u0438\u0434\u0430\u0446\u0438\u044e \u0437\u0430\u043a\u0430\u0437\u0430: Inventory Service, Fraud Service \u0438 Order Details Service. \u042d\u0442\u0438 \u0441\u0435\u0440\u0432\u0438\u0441\u044b \u043f\u0430\u0440\u0430\u043b\u043b\u0435\u043b\u044c\u043d\u043e \u0438 \u043d\u0435\u0437\u0430\u0432\u0438\u0441\u0438\u043c\u043e \u0434\u0440\u0443\u0433 \u043e\u0442 \u0434\u0440\u0443\u0433\u0430 \u043e\u0441\u0443\u0449\u0435\u0441\u0442\u0432\u043b\u044f\u044e\u0442 \u0432\u0430\u043b\u0438\u0434\u0430\u0446\u0438\u044e \u0437\u0430\u043a\u0430\u0437\u0430, \u0430 \u0440\u0435\u0437\u0443\u043b\u044c\u0442\u0430\u0442 PASS \u0438\u043b\u0438 FAIL \u0437\u0430\u043f\u0438\u0441\u044b\u0432\u0430\u044e\u0442 \u0432 \u0442\u043e\u043f\u0438\u043a Kafka <code>order-validations.v1<\/code>. Validation Aggregator Service \u0432\u044b\u0447\u0438\u0442\u044b\u0432\u0430\u0435\u0442 \u044d\u0442\u043e\u0442 \u0442\u043e\u043f\u0438\u043a, \u0430\u0433\u0433\u0440\u0435\u0433\u0438\u0440\u0443\u0435\u0442 \u0440\u0435\u0437\u0443\u043b\u044c\u0442\u0430\u0442\u044b \u043f\u043e \u043a\u0430\u0436\u0434\u043e\u043c\u0443 \u0437\u0430\u043a\u0430\u0437\u0443 \u0438 \u043e\u0442\u043f\u0440\u0430\u0432\u043b\u044f\u0435\u0442 \u043a\u043e\u043d\u0435\u0447\u043d\u044b\u0439 \u0440\u0435\u0437\u0443\u043b\u044c\u0442\u0430\u0442 \u0432 \u0442\u043e\u043f\u0438\u043a <code>orders.v1<\/code>.<\/p>\n<p>\u0414\u043b\u044f \u0440\u0435\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u0438 GET \u0437\u0430\u043f\u0440\u043e\u0441\u043e\u0432 Orders Service \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u0435\u0442 \u043c\u0430\u0442\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u043e\u0432\u0430\u043d\u043d\u043e\u0435 \u043f\u0440\u0435\u0434\u0441\u0442\u0430\u0432\u043b\u0435\u043d\u0438\u0435 (materialized view), \u043a\u043e\u0442\u043e\u0440\u043e\u0435 \u0432\u0441\u0442\u0440\u043e\u0435\u043d\u043e (embedded) \u0432 \u043a\u0430\u0436\u0434\u044b\u0439 \u0438\u043d\u0441\u0442\u0430\u043d\u0441 Orders Service. \u042d\u0442\u043e \u043c\u0430\u0442\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u043e\u0432\u0430\u043d\u043d\u043e\u0435 \u043f\u0440\u0435\u0434\u0441\u0442\u0430\u0432\u043b\u0435\u043d\u0438\u0435 \u0440\u0435\u0430\u043b\u0438\u0437\u043e\u0432\u0430\u043d\u043e \u043d\u0430 \u043e\u0441\u043d\u043e\u0432\u0435 Kafka Streams state store. \u0422\u0430\u043a \u043a\u0430\u043a state store \u043e\u0441\u043d\u043e\u0432\u0430\u043d \u043d\u0430 \u0442\u043e\u043f\u0438\u043a\u0430\u0445 Kafka, \u0438 \u0432\u0441\u0442\u0440\u043e\u0435\u043d \u0432 \u043a\u0430\u0436\u0434\u044b\u0439 \u0438\u043d\u0441\u0442\u0430\u043d\u0441, \u043e\u043d \u0445\u0440\u0430\u043d\u0438\u0442 \u0434\u0430\u043d\u043d\u044b\u0435 \u0442\u043e\u043b\u044c\u043a\u043e \u0442\u0435\u0445 \u043f\u0430\u0440\u0442\u0438\u0446\u0438\u0439 \u0442\u043e\u043f\u0438\u043a\u0430, \u043a\u043e\u0442\u043e\u0440\u044b\u0435 \u0432\u044b\u0447\u0438\u0442\u044b\u0432\u0430\u044e\u0442\u0441\u044f \u0434\u0430\u043d\u043d\u044b\u043c \u0438\u043d\u0441\u0442\u0430\u043d\u0441\u043e\u043c. \u041e\u0434\u043d\u0430\u043a\u043e Kafka Streams \u043f\u0440\u0435\u0434\u043e\u0441\u0442\u0430\u0432\u043b\u044f\u0435\u0442 API (Interactive Queries) \u0434\u043b\u044f \u0442\u043e\u0433\u043e, \u0447\u0442\u043e\u0431\u044b &#171;\u043d\u0430\u0439\u0442\u0438&#187; \u0438\u043d\u0441\u0442\u0430\u043d\u0441, \u043d\u0430 \u043a\u043e\u0442\u043e\u0440\u043e\u043c \u0445\u0440\u0430\u043d\u044f\u0442\u0441\u044f \u0434\u0430\u043d\u043d\u044b\u0435, \u0435\u0441\u043b\u0438 \u0438\u0445 \u043d\u0435\u0442 \u043b\u043e\u043a\u0430\u043b\u044c\u043d\u043e. \u042d\u0442\u043e \u043f\u043e\u0437\u0432\u043e\u043b\u044f\u0435\u0442 \u0440\u0435\u0430\u043b\u0438\u0437\u043e\u0432\u0430\u0442\u044c \u0433\u0430\u0440\u0430\u043d\u0442\u0438\u044e \u0434\u043b\u044f \u043a\u043b\u0438\u0435\u043d\u0442\u043e\u0432 read-your-own-writes (\u0447\u0442\u0435\u043d\u0438\u0435 \u0441\u0432\u043e\u0438\u0445 \u0437\u0430\u043f\u0438\u0441\u0435\u0439).<\/p>\n<p>\u0422\u0430\u043a\u0436\u0435 \u0432 \u0441\u0438\u0441\u0442\u0435\u043c\u0435 \u0435\u0441\u0442\u044c \u043f\u0440\u043e\u0441\u0442\u043e\u0439 \u0441\u0435\u0440\u0432\u0438\u0441 \u0434\u043b\u044f \u043e\u0442\u043f\u0440\u0430\u0432\u043a\u0438 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0439 \u043d\u0430 \u043f\u043e\u0447\u0442\u0443 Email Service (\u0432 \u0442\u0435\u043a\u0443\u0449\u0435\u0439 \u0440\u0435\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u0438 \u043f\u0440\u043e\u0441\u0442\u043e \u043e\u0441\u0443\u0449\u0435\u0441\u0442\u0432\u043b\u044f\u0435\u0442\u0441\u044f \u043b\u043e\u0433\u0438\u0440\u043e\u0432\u0430\u043d\u0438\u0435 \u043e\u0442\u043f\u0440\u0430\u0432\u043b\u044f\u0435\u043c\u044b\u0445 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0439).<\/p>\n<p>\u0421\u0435\u0440\u0432\u0438\u0441 Order Enrichment \u043e\u0442\u0432\u0435\u0447\u0430\u0435\u0442 \u0437\u0430 \u0441\u043e\u0437\u0434\u0430\u043d\u0438\u0435 <code>orders_enriched_stream<\/code> \u0432 <code>ksqlDB<\/code>, \u0438\u0437 \u043a\u043e\u0442\u043e\u0440\u043e\u0433\u043e \u043c\u043e\u0436\u043d\u043e \u043f\u043e\u043b\u0443\u0447\u0438\u0442\u044c \u0434\u0430\u043d\u043d\u044b\u0435 \u043e \u043f\u043e\u043a\u0443\u043f\u0430\u0442\u0435\u043b\u0435 \u0438 \u0435\u0433\u043e \u0437\u0430\u043a\u0430\u0437\u0435. \u0422\u0430\u043a\u0436\u0435 \u044d\u0442\u043e\u0442 \u0441\u0435\u0440\u0432\u0438\u0441 \u043f\u0440\u0435\u0434\u043e\u0441\u0442\u0430\u0432\u043b\u044f\u0435\u0442 REST \u0440\u0443\u0447\u043a\u0443 \u0434\u043b\u044f \u043f\u043e\u043b\u0443\u0447\u0435\u043d\u0438\u044f \u0438\u043d\u0444\u043e\u0440\u043c\u0430\u0446\u0438\u0438 \u043e \u0444\u043e\u0437\u043c\u043e\u0436\u043d\u044b\u0445 \u043c\u043e\u0448\u0435\u043d\u043d\u0438\u0447\u0435\u0441\u043a\u0438\u0445 \u0434\u0435\u0439\u0441\u0442\u0432\u0438\u044f\u0445 (\u0435\u0441\u043b\u0438 \u043f\u043e\u043a\u0443\u043f\u0430\u0442\u0435\u043b\u044c \u043e\u0441\u0443\u0449\u0435\u0441\u0442\u0432\u0438\u043b \u0431\u043e\u043b\u0435\u0435 2 \u0437\u0430\u043a\u0430\u0437\u043e\u0432 \u0437\u0430 30 \u0441\u0435\u043a\u0443\u043d\u0434):<\/p>\n<ul>\n<li>\n<p>GET <a href=\"http:\/\/localhost:8010\/fraud\" rel=\"noopener noreferrer nofollow\">http:\/\/localhost:8010\/fraud<\/a><\/p>\n<\/li>\n<\/ul>\n<p>\u0414\u0430\u043d\u043d\u044b\u0435 \u043e \u043f\u043e\u043a\u0443\u043f\u0430\u0442\u0435\u043b\u044f\u0445 \u0432\u044b\u0447\u0438\u0442\u044b\u0432\u0430\u044e\u0442\u0441\u044f \u0438\u0437 \u0411\u0414 SQLite \u0441 \u043f\u043e\u043c\u043e\u0449\u044c\u044e Kafka Connect <code>io.confluent.connect.jdbc.JdbcSourceConnector<\/code> \u0432 \u0442\u043e\u043f\u0438\u043a Kafka customers. \u041a\u043e\u043d\u0444\u0438\u0433\u0443\u0440\u0430\u0446\u0438\u044f \u043a\u043e\u043d\u043d\u0435\u043a\u0442\u043e\u0440\u0430: <a href=\"https:\/\/github.com\/aasmc\/event-driven-microservices-kafka\/blob\/master\/infra\/connectors\/connector_jdbc_customers_template.config\" rel=\"noopener noreferrer nofollow\">connector_jdbc_customers_template.config<\/a><\/p>\n<p>\u0414\u0430\u043d\u043d\u044b\u0435 \u043e \u0437\u0430\u043a\u0430\u0437\u0430\u0445 \u0437\u0430\u043f\u0438\u0441\u044b\u0432\u0430\u044e\u0442\u0441\u044f \u0438\u0437 \u0442\u043e\u043f\u0438\u043a\u0430 Kafka <code>orders.v1<\/code> \u0432 \u0438\u043d\u0434\u0435\u043a\u0441 <code>orders.v1<\/code> ElasticSearch \u0441 \u043f\u043e\u043c\u043e\u0449\u044c\u044e Kafka Connect <code>io.confluent.connect.elasticsearch.ElasticsearchSinkConnector<\/code>. \u041a\u043e\u043d\u0444\u0438\u0433\u0443\u0440\u0430\u0446\u0438\u044f \u043a\u043e\u043d\u043d\u0435\u043a\u0442\u043e\u0440\u0430: <a href=\"https:\/\/github.com\/aasmc\/event-driven-microservices-kafka\/blob\/master\/infra\/connectors\/connector_elasticsearch_template.config\" rel=\"noopener noreferrer nofollow\">connector_elasticsearch_template.config<\/a><\/p>\n<figure class=\"full-width\"><img loading=\"lazy\" decoding=\"async\" src=\"https:\/\/habrastorage.org\/r\/w1560\/getpro\/habr\/upload_files\/f83\/0ab\/3e6\/f830ab3e6369083b28fc43b997b32979.png\" width=\"570\" height=\"326\" data-src=\"https:\/\/habrastorage.org\/getpro\/habr\/upload_files\/f83\/0ab\/3e6\/f830ab3e6369083b28fc43b997b32979.png\"\/><\/figure>\n<h3>\u0414\u0438\u0430\u0433\u0440\u0430\u043c\u043c\u0430 \u043c\u0438\u043a\u0440\u043e\u0441\u0435\u0440\u0432\u0438\u0441\u043e\u0432 \u0438 \u0442\u043e\u043f\u0438\u043a\u043e\u0432 Kafka:<\/h3>\n<figure class=\"full-width\"><img loading=\"lazy\" decoding=\"async\" src=\"https:\/\/habrastorage.org\/r\/w1560\/getpro\/habr\/upload_files\/69c\/fca\/13f\/69cfca13f58359be980b44745c16ed07.png\" width=\"3510\" height=\"1711\" data-src=\"https:\/\/habrastorage.org\/getpro\/habr\/upload_files\/69c\/fca\/13f\/69cfca13f58359be980b44745c16ed07.png\"\/><\/figure>\n<h3>\u0414\u0435\u0442\u0430\u043b\u0438 \u0440\u0435\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u0438<\/h3>\n<h4>\u041c\u043e\u0434\u0443\u043b\u044c Common<\/h4>\n<p>\u0414\u0430\u043d\u043d\u044b\u0439 \u043c\u043e\u0434\u0443\u043b\u044c \u0441\u043e\u0431\u0440\u0430\u043d \u043a\u0430\u043a \u0431\u0438\u0431\u043b\u0438\u043e\u0442\u0435\u043a\u0430, \u043a\u043e\u0442\u043e\u0440\u0443\u044e \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u044e\u0442 \u043c\u0438\u043a\u0440\u043e\u0441\u0435\u0440\u0432\u0438\u0441\u044b. \u0421\u043e\u0434\u0435\u0440\u0436\u0438\u0442 \u0441\u0433\u0435\u043d\u0435\u0440\u0438\u0440\u043e\u0432\u0430\u043d\u043d\u044b\u0435 AVRO \u043a\u043b\u0430\u0441\u0441\u044b \u0430 \u0442\u0430\u043a\u0436\u0435 \u0443\u0442\u0438\u043b\u043b\u0438\u0442\u043d\u044b\u0435 \u043a\u043b\u0430\u0441\u0441\u044b, \u043e\u0431\u043b\u0435\u0433\u0447\u0430\u044e\u0449\u0438\u0435 \u0440\u0430\u0431\u043e\u0442\u0443 \u0441 \u0442\u043e\u043f\u0438\u043a\u0430\u043c\u0438, \u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u0430\u0442\u043e\u0440\u0430\u043c\u0438 \u0438 \u0434\u0435\u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u0430\u0442\u043e\u0440\u0430\u043c\u0438.<\/p>\n<h3>\u0421\u0443\u0449\u043d\u043e\u0441\u0442\u0438<\/h3>\n<p>\u0413\u0435\u043d\u0435\u0440\u0430\u0446\u0438\u044f \u0441\u0443\u0449\u043d\u043e\u0441\u0442\u0435\u0439, \u0441 \u043a\u043e\u0442\u043e\u0440\u044b\u043c\u0438 \u0440\u0430\u0431\u043e\u0442\u0430\u0435\u0442 Kafka \u043e\u0441\u0443\u0449\u0435\u0441\u0442\u0432\u043b\u044f\u0435\u0442\u0441\u044f <a href=\"https:\/\/github.com\/aasmc\/event-driven-microservices-kafka\/tree\/master\/common\/src\/main\/avro\" rel=\"noopener noreferrer nofollow\">\u043d\u0430 \u043e\u0441\u043d\u043e\u0432\u0435 \u0441\u0445\u0435\u043c Avro<\/a> \u0441 \u043f\u043e\u043c\u043e\u0449\u044c\u044e Gradle plugin <code>id 'com.github.davidmc24.gradle.plugin.avro' version \"1.9.1\"<\/code>, \u043d\u0430 \u044d\u0442\u0430\u043f\u0435 \u0441\u0431\u043e\u0440\u043a\u0438 \u0431\u0438\u0431\u043b\u0438\u043e\u0442\u0435\u043a\u0438 <code>common<\/code>. \u041d\u0438\u0436\u0435 \u043f\u0440\u0438\u0432\u0435\u0434\u0435\u043d\u043e \u043e\u043f\u0438\u0441\u0430\u043d\u0438\u0435 \u0441\u0443\u0449\u043d\u043e\u0441\u0442\u0435\u0439 \u0431\u0435\u0437 \u0434\u043e\u043f\u043e\u043b\u043d\u0438\u0442\u0435\u043b\u044c\u043d\u043e\u0433\u043e \u0441\u0433\u0435\u043d\u0435\u0440\u0438\u0440\u043e\u0432\u0430\u043d\u043d\u043e\u0433\u043e \u043a\u043e\u0434\u0430.<\/p>\n<pre><code class=\"kotlin\">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 )<\/code><\/pre>\n<p>\u0420\u0435\u0433\u0438\u0441\u0442\u0440\u0430\u0446\u0438\u044f \u0441\u0445\u0435\u043c Avro \u0432 Confluent Schema Registry \u043e\u0441\u0443\u0449\u0435\u0441\u0442\u0432\u043b\u044f\u0435\u0442\u0441\u044f \u0441 \u043f\u043e\u043c\u043e\u0449\u044c\u044e Gradle Plugin: <code>id 'com.github.imflog.kafka-schema-registry-gradle-plugin' version \"1.12.0\" <\/code> \u043f\u043e\u0441\u043b\u0435 \u0442\u043e\u0433\u043e, \u043a\u0430\u043a \u043f\u043e\u0434\u043d\u044f\u0442 docker \u043a\u043e\u043d\u0442\u0435\u0439\u043d\u0435\u0440 schema-registry. \u0427\u0442\u043e\u0431\u044b \u0437\u0430\u0440\u0435\u0433\u0438\u0441\u0442\u0440\u0438\u0440\u043e\u0432\u0430\u0442\u044c \u0441\u0445\u0435\u043c\u044b \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e \u043d\u0430\u0441\u0442\u0440\u043e\u0438\u0442\u044c \u043f\u043b\u0430\u0433\u0438\u043d. <\/p>\n<p>\u041c\u0438\u043d\u0438\u043c\u0430\u043b\u044c\u043d\u0430\u044f \u043d\u0430\u0441\u0442\u0440\u043e\u0439\u043a\u0430 \u0432 Gradle:<\/p>\n<pre><code>schemaRegistry {     url = 'http:\/\/localhost:8081'     quiet = true     register {         subject('customers-value', 'common\/src\/main\/avro\/customer.avsc', 'AVRO')         subject('orders.v1-value', 'common\/src\/main\/avro\/order.avsc', 'AVRO')         subject('orders-enriched.v1-value', 'common\/src\/main\/avro\/orderenriched.avsc', 'AVRO')         subject('order-validations.v1-value', 'common\/src\/main\/avro\/ordervalidation.avsc', 'AVRO')         subject('payments.v1-value', 'common\/src\/main\/avro\/payment.avsc', 'AVRO')     } }<\/code><\/pre>\n<p>\u041f\u0440\u0438 \u0440\u0435\u0433\u0438\u0441\u0442\u0440\u0430\u0446\u0438\u0438 \u0443\u043a\u0430\u0437\u044b\u0432\u0430\u044e\u0442\u0441\u044f:<\/p>\n<ol>\n<li>\n<p>\u041d\u0430\u0437\u0432\u0430\u043d\u0438\u0435 \u0441\u0445\u0435\u043c\u044b. \u041e\u0434\u043d\u0438\u043c \u0438\u0437 \u043f\u0430\u0442\u0442\u0435\u0440\u043d\u043e\u0432 \u043d\u0430\u0438\u043c\u0435\u043d\u043e\u0432\u0430\u043d\u0438\u044f \u044f\u0432\u043b\u044f\u0435\u0442\u0441\u044f TopicName-value \/ TopicName-key<\/p>\n<\/li>\n<li>\n<p>\u041f\u0443\u0442\u044c \u0434\u043e \u0444\u0430\u0439\u043b\u0430 \u0441\u043e \u0441\u0445\u0435\u043c\u043e\u0439<\/p>\n<\/li>\n<li>\n<p>\u0424\u043e\u0440\u043c\u0430\u0442 \u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u0438, \u0432 \u043d\u0430\u0448\u0435\u043c \u0441\u043b\u0443\u0447\u0430\u0435 &#8212; AVRO.<\/p>\n<\/li>\n<\/ol>\n<p>\u0414\u0430\u043b\u0435\u0435 \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e \u0437\u0430\u043f\u0443\u0441\u0442\u0438\u0442\u044c \u0442\u0430\u0441\u043a\u0443 Gradle \u0438\u0437 \u043a\u043e\u0440\u043d\u0435\u0432\u043e\u0439 \u0434\u0438\u0440\u0435\u043a\u0442\u043e\u0440\u0438\u0438 \u043f\u0440\u043e\u0435\u043a\u0442\u0430 (\u043d\u0435 \u0438\u0437 \u0434\u0438\u0440\u0435\u043a\u0442\u043e\u0440\u0438\u0438 common):<\/p>\n<pre><code class=\"bash\">.\/gradlew registerSchemaTask<\/code><\/pre>\n<p>\u0414\u043b\u044f \u0440\u0430\u0431\u043e\u0442\u044b \u0441 \u0437\u0430\u043f\u0438\u0441\u044f\u043c\u0438 \u0432 \u0442\u043e\u043f\u0438\u043a\u0430\u0445 \u043f\u0440\u0438\u043b\u043e\u0436\u0435\u043d\u0438\u0435 Kafka Streams \u0434\u043e\u043b\u0436\u043d\u043e \u0437\u043d\u0430\u0442\u044c \u043e \u0444\u043e\u0440\u043c\u0430\u0442\u0435 \u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u0438 \u0434\u0430\u043d\u043d\u044b\u0445. \u0414\u043b\u044f \u044d\u0442\u043e\u0433\u043e \u043d\u0430\u0441\u0442\u0440\u0430\u0438\u0432\u0430\u044e\u0442\u0441\u044f \u0441\u043f\u0435\u0446\u0438\u0430\u043b\u044c\u043d\u044b\u0435 <code>org.apache.kafka.common.serialization.Serde<\/code> \u043a\u043b\u0430\u0441\u0441\u044b, \u043a\u043e\u0442\u043e\u0440\u044b\u0435 \u0437\u043d\u0430\u044e\u0442 \u043a\u0430\u043a \u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u043e\u0432\u0430\u0442\u044c\/\u0434\u0435\u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u043e\u0432\u0430\u0442\u044c \u043a\u043b\u044e\u0447\/\u0437\u043d\u0430\u0447\u0435\u043d\u0438\u0435. \u0414\u043b\u044f \u0443\u0434\u043e\u0431\u0441\u0442\u0432\u0430 \u0432\u0441\u0435 \u043d\u0430\u0441\u0442\u0440\u043e\u0439\u043a\u0438 \u043f\u043e\u043c\u0435\u0449\u0435\u043d\u044b \u0432 \u043e\u0442\u0434\u0435\u043b\u044c\u043d\u044b\u0439 \u043a\u043b\u0430\u0441\u0441 <a href=\"https:\/\/github.com\/aasmc\/event-driven-microservices-kafka\/blob\/master\/common\/src\/main\/kotlin\/ru\/aasmc\/eventdriven\/common\/schemas\/Schemas.kt\" rel=\"noopener noreferrer nofollow\">Schemas<\/a>. \u0414\u043b\u044f \u0441\u0433\u0435\u043d\u0435\u0440\u0438\u0440\u043e\u0432\u0430\u043d\u043d\u044b\u0445 AVRO \u043a\u043b\u0430\u0441\u0441\u043e\u0432 \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u0435\u0442\u0441\u044f <code>io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde<\/code>, \u043f\u0440\u0438 \u043d\u0430\u0441\u0442\u0440\u043e\u0439\u043a\u0435 \u043a\u043e\u0442\u043e\u0440\u043e\u0433\u043e \u0443\u043a\u0430\u0437\u044b\u0432\u0430\u0435\u0442\u0441\u044f \u0430\u0434\u0440\u0435\u0441 Schema Registry &#8212; \u0441\u0435\u0440\u0432\u0435\u0440\u0430, \u0445\u0440\u0430\u043d\u044f\u0449\u0435\u0433\u043e \u0432\u0441\u0435, \u0432 \u043d\u0430\u0448\u0435\u043c \u0441\u043b\u0443\u0447\u0430\u0435 AVRO, \u0441\u0445\u0435\u043c\u044b.<\/p>\n<pre><code class=\"kotlin\">fun configureSerdes() {     val config = hashMapOf&lt;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&lt;*>, config: Map&lt;String, Any>, isKey: Boolean) {   if (serde is SpecificAvroSerde) {     serde.configure(config, isKey)   } }<\/code><\/pre>\n<h3>Orders Service<\/h3>\n<p>\u0412 Spring Boot \u043d\u0435\u0442 \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e\u0441\u0442\u0438 \u0432\u0440\u0443\u0447\u043d\u0443\u044e \u0443\u043f\u0440\u0430\u0432\u043b\u044f\u0442\u044c \u0436\u0438\u0437\u043d\u0435\u043d\u043d\u044b\u043c \u0446\u0438\u043a\u043b\u043e\u043c \u0441\u0443\u0449\u043d\u043e\u0441\u0442\u0438 KafkaStreams. \u0414\u043b\u044f \u044d\u0442\u043e\u0433\u043e \u0435\u0441\u0442\u044c \u0430\u0431\u0441\u0442\u0440\u0430\u043a\u0446\u0438\u044f <code>org.springframework.kafka.config.StreamsBuilderFactoryBean<\/code>, \u043a\u043e\u0442\u043e\u0440\u0430\u044f \u0441\u043e\u0437\u0434\u0430\u0435\u0442 \u0438 \u0443\u043f\u0440\u0430\u0432\u043b\u044f\u0435\u0442 <code>org.apache.kafka.streams.KafkaStreams<\/code> \u0438 <code>org.apache.kafka.streams.StreamsBuilder<\/code>. \u0422\u0435\u043c \u043d\u0435 \u043c\u0435\u043d\u0435\u0435, \u043d\u0430\u043c \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e \u0441\u043a\u043e\u043d\u0444\u0438\u0433\u0443\u0440\u0438\u0440\u043e\u0432\u0430\u0442\u044c \u0431\u0438\u043d <code>org.springframework.kafka.config.KafkaStreamsConfiguration<\/code>:<\/p>\n<pre><code class=\"kotlin\">@EnableKafkaStreams @Configuration class KafkaConfig(     private val kafkaProps: KafkaProps,     private val serviceUtils: ServiceUtils,     private val topicProps: TopicsProps ) {      @Bean     fun outstandingRequests(): MutableMap&lt;String, FilteredResponse&lt;String, Order, OrderDto>> {         return ConcurrentHashMap()     }      @Bean(name = [KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME])     fun kStreamsConfig(): KafkaStreamsConfiguration {         val props = hashMapOf&lt;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,             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)     } }<\/code><\/pre>\n<p>\u0414\u043b\u044f \u0442\u043e\u0433\u043e, \u0447\u0442\u043e\u0431\u044b \u0440\u0430\u0431\u043e\u0442\u0430\u043b\u0438 Interactive Queries (\u0435\u0441\u043b\u0438 \u043c\u044b \u0445\u043e\u0442\u0438\u043c \u0437\u0430\u043f\u0440\u0430\u0448\u0438\u0432\u0430\u0442\u044c \u0434\u0430\u043d\u043d\u044b\u0435 \u043d\u0435 \u0442\u043e\u043b\u044c\u043a\u043e<br \/> \u0438\u0437 \u043b\u043e\u043a\u0430\u043b\u044c\u043d\u043e\u0433\u043e state store, \u043d\u043e \u0438 \u0441 \u0434\u0440\u0443\u0433\u0438\u0445 \u0438\u043d\u0441\u0442\u0430\u043d\u0441\u043e\u0432), \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e \u0432 \u043a\u043e\u043d\u0444\u0438\u0433\u0443\u0440\u0430\u0446\u0438\u044f\u0445 \u0443\u043a\u0430\u0437\u0430\u0442\u044c<br \/> <code>StreamsConfig.APPLICATION_SERVER_CONFIG<\/code> &#8212; \u0430\u0434\u0440\u0435\u0441 \u0442\u0435\u043a\u0443\u0449\u0435\u0433\u043e \u0438\u043d\u0441\u0442\u0430\u043d\u0441\u0430 \u0432 \u0444\u043e\u0440\u043c\u0430\u0442\u0435 <code>host:port<\/code>. \u0422\u0430\u043a\u0436\u0435 \u043c\u044b \u0443\u043a\u0430\u0437\u044b\u0432\u0430\u0435\u043c \u0430\u0434\u0440\u0435\u0441 Schema Registry, \u043f\u0443\u0442\u044c \u0434\u043e \u043a\u0430\u0442\u0430\u043b\u043e\u0433\u0430, \u0433\u0434\u0435 \u0431\u0443\u0434\u0435\u0442 \u0445\u0440\u0430\u043d\u0438\u0442\u0441\u044f state store \u0438 \u0442.\u0434. \u041f\u043e \u0443\u043c\u043e\u043b\u0447\u0430\u043d\u0438\u044e Kafka Streams \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u0435\u0442 \u0432\u0441\u0442\u0440\u043e\u0435\u043d\u043d\u043e\u0435 key-value \u0445\u0440\u0430\u043d\u0438\u043b\u0438\u0449\u0435 RocksDB \u0432 \u043a\u0430\u0447\u0435\u0441\u0442\u0432\u0435 state store.<\/p>\n<p>\u041f\u0440\u0438 \u0441\u0442\u0430\u0440\u0442\u0435 \u043f\u0440\u0438\u043b\u043e\u0436\u0435\u043d\u0438\u044f Orders Service \u043d\u0430 \u043e\u0441\u043d\u043e\u0432\u0435 \u0442\u043e\u043f\u0438\u043a\u0430 Kafka <code>orders.v1<\/code> \u0441\u043e\u0437\u0434\u0430\u0435\u0442\u0441\u044f KTable &#8212; \u0430\u0431\u0441\u0442\u0440\u0430\u043a\u0446\u0438\u044f Kafka Streams, \u043a\u043e\u0442\u043e\u0440\u0430\u044f \u043f\u0440\u0435\u0434\u0441\u0442\u0430\u0432\u043b\u044f\u0435\u0442 \u0438\u0437 \u0441\u0435\u0431\u044f snapshot \u0441\u043e\u0441\u0442\u043e\u044f\u043d\u0438\u044f (\u043c\u043e\u0436\u043d\u043e \u043f\u0440\u043e\u0432\u0435\u0441\u0442\u0438 \u0430\u043d\u0430\u043b\u043e\u0433\u0438\u044e \u0441 compacted topic) &#8212; \u0445\u0440\u0430\u043d\u044f\u0442\u0441\u044f \u043f\u043e\u0441\u043b\u0435\u0434\u043d\u0438\u0435 \u0437\u0430\u043f\u0438\u0441\u0438 \u043f\u043e \u043a\u043b\u044e\u0447\u0443 \u0432 \u0442\u043e\u043f\u0438\u043a\u0435. \u041f\u0440\u0438 \u043f\u043e\u0441\u0442\u0443\u043f\u043b\u0435\u043d\u0438\u0438 \u043d\u043e\u0432\u044b\u0445 \u0437\u0430\u043f\u0438\u0441\u0435\u0439 \u0434\u0430\u043d\u043d\u044b\u0435 \u0432 KTable \u043e\u0431\u043d\u043e\u0432\u043b\u044f\u044e\u0442\u0441\u044f, \u0444\u0430\u043a\u0442\u0438\u0447\u0435\u0441\u043a\u0438 \u043f\u0440\u043e\u0438\u0437\u0432\u043e\u0434\u0438\u0442\u0441\u044f \u043e\u043f\u0435\u0440\u0430\u0446\u0438\u044f <code>UPSERT<\/code> &#8212; \u0435\u0441\u043b\u0438 \u043a\u043b\u044e\u0447 \u0435\u0441\u0442\u044c \u0432 \u0442\u0430\u0431\u043b\u0438\u0446\u0435, \u0434\u0430\u043d\u043d\u044b\u0435 \u043e\u0431\u043d\u043e\u0432\u043b\u044f\u044e\u0442\u0441\u044f, \u0435\u0441\u043b\u0438 \u043d\u0435\u0442 &#8212; \u0441\u043e\u0437\u0434\u0430\u044e\u0442\u0441\u044f, \u0435\u0441\u043b\u0438 \u043f\u0440\u0438\u0445\u043e\u0434\u0438\u0442 \u043a\u043b\u044e\u0447 &#8212; <code>null<\/code> &#8212; \u0442\u043e \u044d\u0442\u043e \u043e\u0437\u043d\u0430\u0447\u0430\u0435\u0442 \u0443\u0434\u0430\u043b\u0435\u043d\u0438\u0435 \u0434\u0430\u043d\u043d\u044b\u0445, \u0431\u043e\u043b\u0435\u0435 \u043e\u043d\u0438 \u043d\u0435 \u0431\u0443\u0434\u0443\u0442 \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u044c\u0441\u044f \u043f\u0440\u0438 \u0432\u044b\u0431\u043e\u0440\u043a\u0435 \u043f\u043e \u043a\u043b\u044e\u0447\u0443 \u0438\u043b\u0438 \u0441\u043e\u0435\u0434\u0438\u043d\u0435\u043d\u0438\u0438 \u0442\u0430\u0431\u043b\u0438\u0446\u044b \u0441 \u0434\u0440\u0443\u0433\u0438\u043c\u0438 \u0441\u0443\u0449\u043d\u043e\u0441\u0442\u044f\u043c\u0438 Kafka Streams (KStream, GlobalKTable). \u041f\u0440\u0438 \u044d\u0442\u043e\u043c Kafka Streams \u0433\u0430\u0440\u0430\u043d\u0442\u0438\u0440\u0443\u0435\u0442, \u0447\u0442\u043e \u0432 \u0441\u043b\u0443\u0447\u0430\u0435 \u043f\u0430\u0434\u0435\u043d\u0438\u044f \u0438\u043d\u0441\u0442\u0430\u043d\u0441\u0430, \u0434\u0430\u043d\u043d\u044b\u0435 \u043d\u0435 \u0431\u0443\u0434\u0443\u0442 \u043f\u043e\u0442\u0435\u0440\u044f\u043d\u044b, \u0442\u0430\u043a \u043a\u0430\u043a \u043e\u043d\u0438 \u043f\u043e\u043c\u0438\u043c\u043e \u043b\u043e\u043a\u0430\u043b\u044c\u043d\u043e\u0433\u043e state store, \u0445\u0440\u0430\u043d\u044f\u0442\u0441\u044f \u0432 change log \u0442\u043e\u043f\u0438\u043a\u0435 Kafka &#8212; \u0434\u043b\u044f \u0441\u043e\u043a\u0440\u0430\u0449\u0435\u043d\u0438\u044f \u043f\u043e\u0442\u0440\u0435\u0431\u043b\u0435\u043d\u0438\u044f \u0440\u0435\u0441\u0443\u0440\u0441\u043e\u0432 \u043f\u043e \u043f\u0430\u043c\u044f\u0442\u0438 \u044d\u0442\u043e\u0442 \u0442\u043e\u043f\u0438\u043a &#8212; compacted.<\/p>\n<p>\u0414\u043b\u044f \u0442\u043e\u0433\u043e, \u0447\u0442\u043e\u0431\u044b \u043f\u0440\u0438\u043b\u043e\u0436\u0435\u043d\u0438\u0435 \u0438\u043c\u0435\u043b\u043e \u0432\u043e\u0437\u043c\u043e\u0436\u043d\u043e\u0441\u0442\u044c \u043f\u043e\u043b\u0443\u0447\u0430\u0442\u044c \u0434\u0430\u043d\u043d\u044b\u0435 \u0438\u0437 state store, \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e \u0435\u0433\u043e \u043c\u0430\u0442\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u043e\u0432\u0430\u0442\u044c, \u0443\u043a\u0430\u0437\u0430\u0432 \u043d\u0430\u0437\u0432\u0430\u043d\u0438\u0435, \u043f\u043e \u043a\u043e\u0442\u043e\u0440\u043e\u043c\u0443 \u0432 \u0434\u0430\u043b\u044c\u043d\u0435\u0439\u0448\u0435\u043c \u043a \u043d\u0435\u043c\u0443 \u043c\u043e\u0436\u043d\u043e \u043e\u0431\u0440\u0430\u0449\u0430\u0442\u044c\u0441\u044f.<\/p>\n<pre><code class=\"kotlin\">@Component class OrdersServiceTopology(     private val orderProps: OrdersProps,     private val mapper: OrderMapper,     private val schemas: Schemas,     private val outstandingRequests: MutableMap&lt;String, FilteredResponse&lt;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 &amp;&amp; callback.predicate(id, order)) {             callback.asyncResponse.setResult(mapper.toDto(order))            outstandingRequests.remove(id)         }     } }<\/code><\/pre>\n<p>\u041c\u0435\u0442\u043e\u0434 <code>OrdersServiceTopology.maybeCompleteLongPollGet(String, Order)<\/code> \u043f\u043e\u043b\u0443\u0447\u0430\u0435\u0442 \u043f\u0440\u0438\u0445\u043e\u0434\u044f\u0449\u0438\u0435 \u0432 KTable \u0434\u0430\u043d\u043d\u044b\u0435 \u043e \u0437\u0430\u043a\u0430\u0437\u0430\u0445, \u0438 \u0435\u0441\u043b\u0438 \u043d\u043e\u0432\u044b\u0439 \u0437\u0430\u043a\u0430\u0437 \u0445\u0440\u0430\u043d\u0438\u0442\u0441\u044f \u0432 ConcurrentHashMap, \u0438 \u043e\u043d \u0443\u0434\u043e\u0432\u043b\u0435\u0442\u0432\u043e\u0440\u044f\u0435\u0442 \u0443\u0441\u043b\u043e\u0432\u0438\u044e FilteredResponse.predicate, \u0442\u043e \u0442\u043e\u0433\u0434\u0430 \u043c\u044b \u043a\u043e\u043c\u043f\u043b\u0438\u0442\u0438\u043c DeferredResult \u0438 \u0443\u0434\u0430\u043b\u044f\u0435\u043c \u0435\u0433\u043e \u0438\u0437 ConcurrentHashMap.<\/p>\n<p>\u041a\u043b\u0430\u0441\u0441 <code>org.springframework.web.context.request.async.DeferredResult<\/code> \u043f\u043e\u0437\u0432\u043e\u043b\u044f\u0435\u0442 \u0430\u0441\u0438\u043d\u0445\u0440\u043e\u043d\u043d\u043e \u043e\u0431\u0440\u0430\u0431\u043e\u0442\u0430\u0442\u044c REST \u0437\u0430\u043f\u0440\u043e\u0441. \u0412 Orders Service \u043e\u043d \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u0435\u0442\u0441\u044f \u0441\u043b\u0435\u0434\u0443\u044e\u0449\u0438\u043c \u043e\u0431\u0440\u0430\u0437\u043e\u043c: \u0432 \u043a\u0430\u0447\u0435\u0441\u0442\u0432\u0435 \u0432\u043e\u0437\u0432\u0440\u0430\u0449\u0430\u0435\u043c\u043e\u0433\u043e \u0437\u043d\u0430\u0447\u0435\u043d\u0438\u044f \u0432 \u043a\u043e\u043d\u0442\u0440\u043e\u043b\u043b\u0435\u0440\u0435 \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u0435\u0442\u0441\u044f DeferredResult. \u041f\u043e\u0442\u043e\u043a, \u043f\u0440\u0438\u043d\u044f\u0432\u0448\u0438\u0439 \u0437\u0430\u043f\u0440\u043e\u0441 \u043c\u043e\u0436\u043d\u043e \u043e\u0441\u0432\u043e\u0431\u043e\u0434\u0438\u0442\u044c, \u0430 \u043f\u0440\u043e\u0441\u0442\u0430\u0432\u0438\u0442\u044c \u0437\u043d\u0430\u0447\u0435\u043d\u0438\u0435 \u0432 DeferredResult \u043c\u043e\u0436\u043d\u043e \u0438\u0437 \u0434\u0440\u0443\u0433\u043e\u0433\u043e \u043f\u043e\u0442\u043e\u043a\u0430. \u041f\u0440\u0438 \u044d\u0442\u043e\u043c \u043c\u043e\u0436\u043d\u043e \u0443\u043a\u0430\u0437\u0430\u0442\u044c \u0442\u0430\u0439\u043c\u0430\u0443\u0442, \u043f\u043e\u0441\u043b\u0435 \u043a\u043e\u0442\u043e\u0440\u043e\u0433\u043e DeferredResult \u0432\u0435\u0440\u043d\u0435\u0442 \u043e\u0448\u0438\u0431\u043a\u0443. \u041f\u0440\u0438 \u043f\u043e\u043b\u0443\u0447\u0435\u043d\u0438\u0438 \u0437\u0430\u043a\u0430\u0437\u0430 &#8212; <code>getOrder(id)<\/code> &#8212; \u0435\u0441\u043b\u0438 \u043d\u0438 \u0432 \u043b\u043e\u043a\u0430\u043b\u044c\u043d\u043e\u043c state store, \u043d\u0438 \u0432 state store \u043d\u0430 \u0434\u0440\u0443\u0433\u0438\u0445 \u0438\u043d\u0441\u0442\u0430\u043d\u0441\u0430\u0445 \u043d\u0435\u0442 \u0437\u0430\u043a\u0430\u0437\u0430 \u0441 \u043d\u0443\u0436\u043d\u044b\u043c ID, \u043c\u044b \u043f\u043e\u043c\u0435\u0449\u0430\u0435\u043c DeferredResult \u0432 ConcurrentHashMap &#8212; \u0432\u043e\u0437\u043c\u043e\u0436\u043d\u043e \u0437\u0430\u043a\u0430\u0437 \u0441 \u0442\u0430\u043a\u0438\u043c ID \u043f\u043e\u044f\u0432\u0438\u0442\u0441\u044f \u0432 \u0441\u043e\u0437\u0434\u0430\u043d\u043d\u043e\u0439 KTable \u0434\u043e \u0438\u0441\u0442\u0435\u0447\u0435\u043d\u0438\u044f \u0442\u0430\u0439\u043c\u0430\u0443\u0442\u0430, \u0442\u043e\u0433\u0434\u0430 \u043c\u0435\u0442\u043e\u0434 <code>OrdersServiceTopology.maybeCompleteLongPollGet(String, Order)<\/code> \u043f\u0440\u043e\u0441\u0442\u0430\u0432\u0438\u0442 \u0440\u0435\u0437\u0443\u043b\u044c\u0442\u0430\u0442 \u0432 DeferredResult \u0438 \u0432\u0435\u0440\u043d\u0435\u0442 \u0435\u0433\u043e \u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u0435\u043b\u044e. <\/p>\n<p>\u041f\u0440\u0438\u043c\u0435\u0440 \u043c\u0435\u0442\u043e\u0434\u0430 \u0432 \u043a\u043e\u043d\u0442\u0440\u043e\u043b\u043b\u0435\u0440\u0435:<\/p>\n<pre><code class=\"kotlin\">@GetMapping(\"\/{id}\") fun getOrder(    @PathVariable(\"id\") id: String,    @RequestParam(\"timeout\", defaultValue = \"2000\") timeout: Long ): DeferredResult&lt;OrderDto> {     log.info(\"Received request to GET order with id = {}\", id)    val deferredResult = DeferredResult&lt;OrderDto>(timeout)    ordersService.getOrderDto(id, deferredResult, timeout)    return deferredResult }<\/code><\/pre>\n<p>\u041f\u0440\u0438\u043c\u0435\u0440 \u0440\u0430\u0431\u043e\u0442\u044b \u0441 DeferredResult:<\/p>\n<pre><code class=\"kotlin\"> fun getOrderDto(id: String, asyncResponse: DeferredResult&lt;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&lt;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&lt;OrderDto>,        timeout: Long     ) {        log.info(\"Chaining GET to a different instance: {}\", path)        try {           val order = webClientBuilder.build()              .get()              .uri(\"$path?timeout=$timeout\")              .retrieve()              .bodyToMono&lt;OrderDto>()              .block() ?: throw NotFoundException(\"FetchFromOtherHost returned null for request: $path\")           asyncResponse.setResult(order)        } catch (swallowed: Exception) {           log.warn(\"FetchFromOtherHost failed.\", swallowed)        }     }<\/code><\/pre>\n<p>\u0414\u043b\u044f \u0442\u043e\u0433\u043e, \u0447\u0442\u043e\u0431\u044b \u043e\u0431\u0440\u0430\u0442\u0438\u0442\u044c\u0441\u044f \u043a state store, \u043d\u0430\u043c \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e \u0437\u0430\u043f\u0440\u043e\u0441\u0438\u0442\u044c \u0435\u0433\u043e \u0438\u0437 Kafka Streams. \u0422\u0443\u0442 \u0441\u0442\u043e\u0438\u0442 \u043e\u0431\u0440\u0430\u0442\u0438\u0442\u044c \u0432\u043d\u0438\u043c\u0430\u043d\u0438\u0435 \u043d\u0430 \u0442\u043e, \u0447\u0442\u043e \u043d\u0430\u043c \u043f\u0440\u0435\u0434\u043e\u0441\u0442\u0430\u0432\u043b\u044f\u0435\u0442\u0441\u044f \u0442\u043e\u043b\u044c\u043a\u043e read-only \u0445\u0440\u0430\u043d\u0438\u043b\u0438\u0449\u0435.<\/p>\n<pre><code class=\"kotlin\">private lateinit var orderStore: ReadOnlyKeyValueStore&lt;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()           )        )     } }<\/code><\/pre>\n<p>\u0414\u043b\u044f \u043f\u043e\u043b\u0443\u0447\u0435\u043d\u0438\u044f \u0438\u043d\u0444\u043e\u0440\u043c\u0430\u0446\u0438\u0438 \u043e \u0442\u043e\u043c, \u043d\u0430 \u043a\u0430\u043a\u043e\u043c \u0438\u043d\u0441\u0442\u0430\u043d\u0441\u0435 \u0445\u0440\u0430\u043d\u044f\u0442\u0441\u044f \u0434\u0430\u043d\u043d\u044b\u0435 \u043e \u0437\u0430\u043a\u0430\u0437\u0435 \u0441 \u043a\u043e\u043d\u043a\u0440\u0435\u0442\u043d\u044b\u043c ID, Kafka Streams \u043f\u0440\u0435\u0434\u043e\u0441\u0442\u0430\u0432\u043b\u044f\u0435\u0442 API:<\/p>\n<pre><code class=\"kotlin\">    fun &lt;K> streamsMetadataForStoreAndKey(         store: String,         key: K,         serializer: Serializer&lt;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)         )     }<\/code><\/pre>\n<p>\u041f\u0440\u0438 \u0441\u043e\u0437\u0434\u0430\u043d\u0438\u0438 \u0437\u0430\u043a\u0430\u0437\u0430, Orders Service \u043e\u0442\u043f\u0440\u0430\u0432\u043b\u044f\u0435\u0442 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0435 \u0432 \u0442\u043e\u043f\u0438\u043a Kafka <code>orders.v1<\/code> \u0441 \u043f\u043e\u043c\u043e\u0449\u044c\u044e <code>KafkaTemplate&lt;String, Order><\/code>. \u041d\u0430\u0441\u0442\u0440\u043e\u0439\u043a\u0438 KafkaProducer \u0432 \u0434\u0430\u043d\u043d\u043e\u043c \u0441\u043b\u0443\u0447\u0430\u0435 \u0443\u043a\u0430\u0437\u044b\u0432\u0430\u044e\u0442\u0441\u044f \u0432 application.yml \u0444\u0430\u0439\u043b\u0435:<\/p>\n<pre><code class=\"yaml\">spring:   application:     name: order-service   kafka:     bootstrap-servers: ${kafkaprops.bootstrapServers}     producer:       key-serializer: org.apache.kafka.common.serialization.StringSerializer       value-serializer: io.confluent.kafka.serializers.KafkaAvroSerializer       bootstrap-servers: ${kafkaprops.bootstrapServers}       acks: ${kafkaprops.acks}       client-id: ${kafkaprops.clientId}       properties:         enable.idempotence: ${kafkaprops.enableIdempotence}         schema.registry.url: ${kafkaprops.schemaRegistryUrl}         auto.register.schemas: false         use.latest.version: true<\/code><\/pre>\n<p>\u041e\u0441\u043e\u0431\u043e\u0435 \u0432\u043d\u0438\u043c\u0430\u043d\u0438\u0435 \u0441\u0442\u043e\u0438\u0442 \u043e\u0431\u0440\u0430\u0442\u0438\u0442\u044c \u043d\u0430 \u043d\u0430\u0441\u0442\u0440\u043e\u0439\u043a\u0438 <code>auto.register.schema: false<\/code> \u0438 <code>use.latest.version: true<\/code>. \u041f\u043e \u0443\u043c\u043e\u043b\u0447\u0430\u043d\u0438\u044e \u043e\u043d\u0438 \u0432\u044b\u0441\u0442\u0430\u0432\u043b\u0435\u043d\u044b \u043d\u0430\u043e\u0431\u043e\u0440\u043e\u0442. \u0418 \u0435\u0441\u043b\u0438 \u043e\u0441\u0442\u0430\u0432\u0438\u0442\u044c \u0438\u0445 \u0431\u0435\u0437 \u0438\u0437\u043c\u0435\u043d\u0435\u043d\u0438\u044f, \u0442\u043e \u043c\u043e\u0433\u0443\u0442 \u0432\u043e\u0437\u043d\u0438\u043a\u043d\u0443\u0442\u044c \u043e\u0448\u0438\u0431\u043a\u0438 \u043f\u0440\u0438 \u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u0438 \/ \u0434\u0435\u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u0438 \u0434\u0430\u043d\u043d\u044b\u0445 \u0438\u0437-\u0437\u0430 \u0432\u043e\u0437\u043c\u043e\u0436\u043d\u043e\u0439 \u043d\u0435\u0432\u043e\u0441\u043c\u0435\u0441\u0442\u0438\u043c\u043e\u0441\u0442\u0438 \u0441\u0445\u0435\u043c AVRO. Confluent \u0440\u0435\u043a\u043e\u043c\u0435\u043d\u0434\u0443\u0435\u0442 \u0432 \u043f\u0440\u043e\u0434\u0435 \u043d\u0435 \u043f\u0440\u0435\u0434\u043e\u0441\u0442\u0430\u0432\u043b\u044f\u0442\u044c \u0432\u043e\u0437\u043c\u043e\u0436\u043d\u043e\u0441\u0442\u044c \u043f\u0440\u043e\u0434\u044a\u044e\u0441\u0435\u0440\u0430\u043c \u0430\u0432\u0442\u043e\u043c\u0430\u0442\u0438\u0447\u0441\u043a\u0438 \u0440\u0435\u0433\u0438\u0441\u0442\u0440\u0438\u0440\u043e\u0432\u0430\u0442\u044c \u0441\u0445\u0435\u043c\u0443.<\/p>\n<h3>Order Details Service<\/h3>\n<p>\u0421\u0435\u0440\u0432\u0438\u0441 \u043e\u0442\u0432\u0435\u0447\u0430\u0435\u0442 \u0437\u0430 \u043f\u0440\u043e\u0432\u0435\u0440\u043a\u0443 \u0434\u0435\u0442\u0430\u043b\u0435\u0439 \u0437\u0430\u043a\u0430\u0437\u0430:<\/p>\n<ul>\n<li>\n<p>\u043a\u043e\u043b\u0438\u0447\u0435\u0441\u0442\u0432\u043e \u0442\u043e\u0432\u0430\u0440\u043e\u0432 \u043d\u0435 \u043c\u0435\u043d\u044c\u0448\u0435 0<\/p>\n<\/li>\n<li>\n<p>\u0441\u0442\u043e\u0438\u043c\u043e\u0441\u0442\u044c \u0437\u0430\u043a\u0430\u0437\u0430 \u043d\u0435 \u043c\u0435\u043d\u044c\u0448\u0435 0<\/p>\n<\/li>\n<li>\n<p>\u0442\u043e\u0432\u0430\u0440 \u043f\u0440\u0438\u0441\u0443\u0442\u0441\u0442\u0432\u0443\u0435\u0442 \u0432 \u0437\u0430\u043a\u0430\u0437\u0435<\/p>\n<\/li>\n<\/ul>\n<p>\u0421\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u044f \u0432\u044b\u0447\u0438\u0442\u044b\u0432\u0430\u044e\u0442\u0441\u044f \u0438\u0437 \u0442\u043e\u043f\u0438\u043a\u0430 Kafka <code>orders.v1<\/code>, \u0437\u0430\u043a\u0430\u0437 \u043f\u0440\u043e\u0432\u0435\u0440\u044f\u0435\u0442\u0441\u044f, \u0430 \u0440\u0435\u0437\u0443\u043b\u044c\u0442\u0430\u0442 \u043f\u0440\u043e\u0432\u0435\u0440\u043a\u0438 \u043e\u0442\u043f\u0440\u0430\u0432\u043b\u044f\u0435\u0442\u0441\u044f \u0432 \u0442\u043e\u043f\u0438\u043a Kafka <code>order-validations.v1<\/code>. \u0412 \u044d\u0442\u043e\u043c \u0441\u0435\u0440\u0432\u0438\u0441\u0435 \u043d\u0435 \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u0435\u0442\u0441\u044f Kafka Streams, \u0430 \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u044e\u0442\u0441\u044f \u0441\u0442\u0430\u043d\u0434\u0430\u0440\u0442\u043d\u044b\u0435 Kafka Producer \/ Consumer.<\/p>\n<p>\u041a\u043e\u043d\u0444\u0438\u0433\u0443\u0440\u0430\u0446\u0438\u044f Kafka Consumer:<\/p>\n<pre><code class=\"kotlin\">  @Bean     fun kafkaListenerContainerFactory(         consumerFactory: ConsumerFactory&lt;String, Order>     ): ConcurrentKafkaListenerContainerFactory&lt;String, Order> {         val factory = ConcurrentKafkaListenerContainerFactory&lt;String, Order>()         factory.consumerFactory = consumerFactory         return factory     }      @Bean     fun consumerFactory(): ConsumerFactory&lt;String, Order> {         return DefaultKafkaConsumerFactory(consumerProps())     }      private fun consumerProps(): Map&lt;String, Any> {         val props = hashMapOf&lt;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     }<\/code><\/pre>\n<p>\u0417\u0434\u0435\u0441\u044c \u0432\u0430\u0436\u043d\u043e \u043e\u0431\u0440\u0430\u0442\u0438\u0442\u044c \u0432\u043d\u0438\u043c\u0430\u043d\u0438\u0435 \u043d\u0430 \u043a\u043e\u043d\u0444\u0438\u0433\u0443\u0440\u0430\u0446\u0438\u044e <code>specific.avro.reader: true<\/code>. \u041e\u043d\u0430 \u0441\u043e\u043e\u0431\u0449\u0430\u0435\u0442 \u043a\u043e\u043d\u0441\u044a\u044e\u043c\u0435\u0440\u0443, \u0447\u0442\u043e \u043e\u043d \u0432\u044b\u0447\u0438\u0442\u044b\u0432\u0430\u0435\u0442 \u043d\u0435 \u043e\u0431\u043e\u0431\u0449\u0435\u043d\u043d\u044b\u0439 AVRO \u0442\u0438\u043f \u0437\u0430\u043f\u0438\u0441\u0438 (GenericAvroRecord), \u0430 \u043a\u043e\u043d\u043a\u0440\u0435\u0442\u043d\u044b\u0439 \u0442\u0438\u043f SpecificAvroRecord, \u0438 \u0438\u043c\u0435\u0435\u0442 \u0432\u043e\u0437\u043c\u043e\u0436\u043d\u043e\u0441\u0442\u044c \u043f\u043e\u043b\u0443\u0447\u0430\u0442\u044c \u0434\u043e\u0441\u0442\u0443\u043f \u043a \u043f\u043e\u043b\u044f\u043c \u0438 \u043c\u0435\u0442\u043e\u0434\u0430\u043c \u0431\u0435\u0437 \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e\u0441\u0442\u0438 \u043a\u0430\u0441\u0442\u0438\u0442\u044c \u043a \u043a\u043e\u043d\u043a\u0440\u0435\u0442\u043d\u043e\u043c\u0443 \u0442\u0438\u043f\u0443.<\/p>\n<p>\u041a\u043e\u043d\u0444\u0438\u0433\u0443\u0440\u0430\u0446\u0438\u044f Kafka Producer:<\/p>\n<pre><code class=\"kotlin\">   @Bean     fun producerFactory(): ProducerFactory&lt;String, OrderValidation> {         return DefaultKafkaProducerFactory(senderProps())     }      private fun senderProps(): Map&lt;String, Any> {         val props = hashMapOf&lt;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&lt;String, OrderValidation>     ): KafkaTemplate&lt;String, OrderValidation> {         return KafkaTemplate(producerFactory)     }<\/code><\/pre>\n<p>\u0422\u0443\u0442 \u0441\u0442\u043e\u0438\u0442 \u043e\u0431\u0440\u0430\u0442\u0438\u0442\u044c \u0432\u043d\u0438\u043c\u0430\u043d\u0438\u0435 \u043d\u0430 \u0434\u0432\u0435 \u0432\u0435\u0449\u0438. \u041f\u0435\u0440\u0432\u0430\u044f, \u043a\u0430\u043a \u0443\u0436\u0435 \u0431\u044b\u043b\u043e \u0441\u043a\u0430\u0437\u0430\u043d\u043e \u0432\u044b\u0448\u0435 &#8212; \u043c\u044b \u0437\u0430\u043f\u0440\u0435\u0449\u0430\u0435\u043c \u043f\u0440\u043e\u0434\u044a\u044e\u0441\u0435\u0440\u0443 \u0430\u0432\u0442\u043e\u043c\u0430\u0442\u0438\u0447\u0435\u0441\u043a\u0438 \u0440\u0435\u0433\u0438\u0441\u0442\u0440\u0438\u0440\u043e\u0432\u0430\u0442\u044c \u0441\u0445\u0435\u043c\u044b \u0432 Schema Registry, \u0430 \u0442\u0430\u043a\u0436\u0435 \u0443\u043a\u0430\u0437\u044b\u0432\u0430\u0435\u043c \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u044c \u043f\u043e\u0441\u043b\u0435\u0434\u043d\u044e\u044e \u0432\u0435\u0440\u0441\u0438\u044e \u0441\u0445\u0435\u043c\u044b \u043f\u0440\u0438 \u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u0438 \u0434\u0430\u043d\u043d\u044b\u0445. \u0412\u0442\u043e\u0440\u043e\u0439 \u043c\u043e\u043c\u0435\u043d\u0442 \u043a\u0430\u0441\u0430\u0435\u0442\u0441\u044f \u043d\u0430\u0441\u0442\u0440\u043e\u0439\u043a\u0438 <code>ProducerConfig.TRANSACTIONAL_ID_CONFIG<\/code>,<br \/> \u044d\u0442\u0430 \u043d\u0430\u0441\u0442\u0440\u043e\u0439\u043a\u0430 \u0434\u043e\u043b\u0436\u043d\u0430 \u0431\u044b\u0442\u044c \u0443\u043d\u0438\u043a\u0430\u043b\u044c\u043d\u043e\u0439 \u0434\u043b\u044f \u043a\u0430\u0436\u0434\u043e\u0433\u043e \u0438\u043d\u0441\u0442\u0430\u043d\u0441\u0430 \u043f\u0440\u0438\u043b\u043e\u0436\u0435\u043d\u0438\u044f, \u0447\u0442\u043e\u0431\u044b Kafka \u043c\u043e\u0433\u043b\u0430 \u0433\u0430\u0440\u0430\u043d\u0442\u0438\u0440\u043e\u0432\u0430\u0442\u044c \u043a\u043e\u0440\u0440\u0435\u043a\u0442\u043d\u0443\u044e \u0440\u0430\u0431\u043e\u0442\u0443 \u0441 \u0442\u0440\u0430\u043d\u0437\u0430\u043a\u0446\u0438\u044f\u043c\u0438 \u0432 \u0441\u043b\u0443\u0447\u0430\u0435 \u043f\u0430\u0434\u0435\u043d\u0438\u044f \u043f\u0440\u0438\u043b\u043e\u0436\u0435\u043d\u0438\u044f. \u041f\u0440\u0438 \u044d\u0442\u043e\u043c \u0432\u0430\u0436\u043d\u043e \u043f\u043e\u043d\u0438\u043c\u0430\u0442\u044c, \u0447\u0442\u043e \u0443\u043d\u0438\u043a\u0430\u043b\u044c\u043d\u044b\u0439 &#8212; \u043d\u0435 \u0437\u043d\u0430\u0447\u0438\u0442 \u0440\u0430\u043d\u0434\u043e\u043c\u043d\u044b\u0439. \u0422\u043e \u0435\u0441\u0442\u044c, \u0437\u0430 \u0438\u043d\u0441\u0442\u0430\u043d\u0441\u043e\u043c \u0434\u043e\u043b\u0436\u0435\u043d \u0431\u044b\u0442\u044c \u0437\u0430\u043a\u0440\u0435\u043f\u043b\u0435\u043d \u0441\u0432\u043e\u0439 <code>TRANSACTIONAL_ID<\/code>. \u0412 \u0434\u0430\u043d\u043d\u043e\u043c \u0441\u043b\u0443\u0447\u0430\u0435 \u0434\u043b\u044f \u043f\u0440\u043e\u0441\u0442\u043e\u0442\u044b \u0440\u0435\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u0438 \u044f \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u044e \u0441\u0443\u0444\u0444\u0438\u043a\u0441 \u0432 \u0432\u0438\u0434\u0435 \u043f\u043e\u0440\u0442\u0430, \u043d\u0430 \u043a\u043e\u0442\u043e\u0440\u043e\u043c \u043f\u043e\u0434\u043d\u044f\u0442\u043e \u043f\u0440\u0438\u043b\u043e\u0436\u0435\u043d\u0438\u0435. \u041e\u0434\u043d\u0430\u043a\u043e, \u044d\u0442\u0430 \u043b\u0438\u0448\u044c \u0434\u0435\u043c\u043e \u0440\u0435\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u044f.<\/p>\n<h3>Email Service<\/h3>\n<figure class=\"full-width\"><img loading=\"lazy\" decoding=\"async\" src=\"https:\/\/habrastorage.org\/r\/w1560\/getpro\/habr\/upload_files\/bda\/3db\/95f\/bda3db95f22aed373711f06640c08d17.png\" width=\"570\" height=\"230\" data-src=\"https:\/\/habrastorage.org\/getpro\/habr\/upload_files\/bda\/3db\/95f\/bda3db95f22aed373711f06640c08d17.png\"\/><\/figure>\n<p>\u0421\u0435\u0440\u0432\u0438\u0441 \u0432\u044b\u0447\u0438\u0442\u044b\u0432\u0430\u0435\u0442 \u0434\u0430\u043d\u043d\u044b\u0435 \u0438\u0437 \u043d\u0435\u0441\u043a\u043e\u043b\u044c\u043a\u0438\u0445 \u0442\u043e\u043f\u0438\u043a\u043e\u0432 Kafka: <code>customers<\/code>, <code>payments.v1<\/code>, <code>orders.v1<\/code>, \u043e\u0431\u044a\u0435\u0434\u0438\u043d\u044f\u0435\u0442 \u0438\u0445 \u0441 \u043f\u043e\u043c\u043e\u0449\u044c\u044e Kafka Streams, \u0438 \u043e\u0442\u043f\u0440\u0430\u0432\u043b\u044f\u0435\u0442 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0435 \u0447\u0435\u0440\u0435\u0437 \u0438\u043d\u0442\u0435\u0440\u0444\u0435\u0439\u0441 <a href=\"https:\/\/github.com\/aasmc\/event-driven-microservices-kafka\/blob\/master\/email-service\/src\/main\/kotlin\/ru\/aasmc\/email\/service\/Emailer.kt\" rel=\"noopener noreferrer nofollow\">Emailer<\/a>. \u041a\u0440\u043e\u043c\u0435 \u0442\u043e\u0433\u043e, \u0434\u0430\u043d\u043d\u044b\u0435 \u043e \u0437\u0430\u043a\u0430\u0437\u0435 \u0438 \u043f\u043e\u043a\u0443\u043f\u0430\u0442\u0435\u043b\u0435 \u043e\u0442\u043f\u0440\u0430\u0432\u043b\u044f\u044e\u0442\u0441\u044f \u0432 \u0442\u043e\u043f\u0438\u043a Kafka, \u0441\u043e\u0432\u043f\u0430\u0434\u0430\u044e\u0449\u0438\u0439 \u0441 \u0440\u0435\u0439\u0442\u0438\u043d\u0433\u043e\u043c \u043f\u043e\u043a\u0443\u043f\u0430\u0442\u0435\u043b\u044f: <code>platinum<\/code>, <code>gold<\/code>, <code>silver<\/code>, <code>bronze<\/code>.<\/p>\n<figure class=\"full-width\"><img loading=\"lazy\" decoding=\"async\" src=\"https:\/\/habrastorage.org\/r\/w1560\/getpro\/habr\/upload_files\/b42\/ea4\/38e\/b42ea438e70bdf47964f4a22a63ceddd.png\" width=\"3374\" height=\"1052\" data-src=\"https:\/\/habrastorage.org\/getpro\/habr\/upload_files\/b42\/ea4\/38e\/b42ea438e70bdf47964f4a22a63ceddd.png\"\/><\/figure>\n<p>\u041d\u0430\u0441\u0442\u0440\u043e\u0439\u043a\u0438 Kafka Streams \u0442\u0443\u0442 \u0430\u043d\u0430\u043b\u043e\u0433\u0438\u0447\u043d\u044b Orders Service.<\/p>\n<p>\u041a\u043b\u0430\u0441\u0441 EmailService \u043e\u0442\u0432\u0435\u0447\u0430\u0435\u0442 \u0437\u0430 \u0441\u043e\u0437\u0434\u0430\u043d\u0438\u0435 \u0438 \u043e\u0431\u0440\u0430\u0431\u043e\u0442\u043a\u0443 <code>KStream<\/code> \u0438 <code>KTable<\/code>. \u0414\u043b\u044f \u0442\u043e\u0433\u043e, \u0447\u0442\u043e\u0431\u044b \u043f\u043e\u0441\u043b\u0435 \u0441\u0442\u0430\u0440\u0442\u0430 \u043f\u0440\u0438\u043b\u043e\u0436\u0435\u043d\u0438\u044f \u0441\u043e\u0437\u0434\u0430\u043b\u0430\u0441\u044c \u0438 \u0437\u0430\u0440\u0430\u0431\u043e\u0442\u0430\u043b\u0430 \u0442\u043e\u043f\u043e\u043b\u043e\u0433\u0438\u044f \u0441\u0442\u0440\u0438\u043c\u043e\u0432 Kafka Streams, \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e \u0435\u0435 \u043d\u0430\u0441\u0442\u0440\u043e\u0438\u0442\u044c \u0441 \u043f\u043e\u043c\u043e\u0449\u044c\u044e <code>StreamsBuilder<\/code>. \u0414\u043b\u044f \u044d\u0442\u043e\u0433\u043e \u0434\u043e\u0441\u0442\u0430\u0442\u043e\u0447\u043d\u043e \u0432 \u043a\u043b\u0430\u0441\u0441, \u043f\u043e\u043c\u0435\u0447\u0435\u043d\u043d\u044b\u0439 \u0430\u043d\u043d\u043e\u0442\u0430\u0446\u0438\u0435\u0439 <code>@Component<\/code> \u0438\u043b\u0438 <code>@Service<\/code> \u0432 \u043e\u0434\u0438\u043d \u0438\u0437 \u043c\u0435\u0442\u043e\u0434\u043e\u0432 \u0437\u0430\u0438\u043d\u0436\u0435\u043a\u0442\u0438\u0442\u044c StreamsBuilder \u0438 \u0432 \u044d\u0442\u043e\u043c \u043c\u0435\u0442\u043e\u0434\u0435 \u0432\u044b\u0441\u0442\u0440\u043e\u0438\u0442\u044c \u043d\u0443\u0436\u043d\u0443\u044e \u0442\u043e\u043f\u043e\u043b\u043e\u0433\u0438\u044e.<\/p>\n<pre><code class=\"kotlin\">@Service class EmailService(     private val emailer: Emailer,     private val schemas: Schemas ) {      @Autowired     fun processStreams(builder: StreamsBuilder) {         val orders: KStream&lt;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&lt;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&lt;Long, Customer> = builder.globalTable(             schemas.CUSTOMERS.name,             Consumed.with(schemas.CUSTOMERS.keySerde, schemas.CUSTOMERS.valueSerde)         )          val serdes: StreamJoined&lt;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)             )     }  }<\/code><\/pre>\n<p>\u0412 \u0434\u0430\u043d\u043d\u043e\u043c \u043f\u0440\u0438\u043c\u0435\u0440\u0435 \u0441\u043b\u0435\u0434\u0443\u0435\u0442 \u043e\u0431\u0440\u0430\u0442\u0438\u0442\u044c \u0432\u043d\u0438\u043c\u0430\u043d\u0438\u0435 \u043d\u0430 \u043d\u0435\u0441\u043a\u043e\u043b\u044c\u043a\u043e \u043c\u043e\u043c\u0435\u043d\u0442\u043e\u0432.<\/p>\n<ol>\n<li>\n<p><code>KStream&lt;String, Payment><\/code> \u0441\u043e\u0437\u0434\u0430\u0435\u0442\u0441\u044f \u043d\u0430 \u043e\u0441\u043d\u043e\u0432\u0435 \u0442\u043e\u043f\u0438\u043a\u0430 <code>payments.v1,<\/code> \u0432 \u043a\u043e\u0442\u043e\u0440\u043e\u043c \u043a\u043b\u044e\u0447\u043e\u043c \u044f\u0432\u043b\u044f\u0435\u0442\u0441\u044f ID \u043f\u043b\u0430\u0442\u0435\u0436\u0430, \u043f\u0440\u0438 \u044d\u0442\u043e\u043c \u0432 \u0438\u043d\u0444\u043e\u0440\u043c\u0430\u0446\u0438\u0438 \u043e \u043f\u043b\u0430\u0442\u0435\u0436\u0435 \u043f\u0440\u0438\u0441\u0443\u0442\u0441\u0442\u0432\u0443\u0435\u0442 \u043f\u043e\u043b\u0435 <code>orderId<\/code>, \u043a\u043e\u0442\u043e\u0440\u043e\u0435 \u0444\u0430\u043a\u0442\u0438\u0447\u0435\u0441\u043a\u0438 \u0441\u0441\u044b\u043b\u0430\u0435\u0442\u0441\u044f \u043d\u0430 \u0441\u0443\u0449\u043d\u043e\u0441\u0442\u044c <code>Order<\/code>. Kafka Streams \u043f\u043e\u0434\u0434\u0435\u0440\u0436\u0438\u0432\u0430\u0435\u0442 \u043e\u0431\u044a\u0435\u0434\u0438\u043d\u0435\u043d\u0438\u0435 <code>KStream<\/code> \u043f\u043e \u043a\u043b\u044e\u0447\u0443 \u0442\u043e\u043f\u0438\u043a\u0430, \u0438\u0437 \u043a\u043e\u0442\u043e\u0440\u043e\u0433\u043e \u0432\u044b\u0447\u0438\u0442\u044b\u0432\u0430\u044e\u0442\u0441\u044f \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u044f. \u042d\u0442\u043e \u043d\u0430\u0437\u044b\u0432\u0430\u0435\u0442\u0441\u044f equi join. \u041d\u043e \u0442\u0430\u043a \u043a\u0430\u043a \u0432 \u0438\u0441\u0445\u043e\u0434\u043d\u044b\u0445 \u0442\u043e\u043f\u0438\u043a\u0430\u0445 \u0440\u0430\u0437\u043d\u044b\u0435 \u043a\u043b\u044e\u0447\u0438, \u043f\u0440\u0438\u0445\u043e\u0434\u0438\u0442\u0441\u044f \u043f\u0440\u043e\u0432\u0435\u0441\u0442\u0438 re-keying &#8212; \u0432\u044b\u0431\u043e\u0440 \u043d\u043e\u0432\u043e\u0433\u043e \u043a\u043b\u044e\u0447\u0430 \u0432 &#171;\u043f\u0440\u0430\u0432\u043e\u043c&#187; \u0441\u0442\u0440\u0438\u043c\u0435 \u0441 \u043f\u043e\u043c\u043e\u0449\u044c\u044e \u043c\u0435\u0442\u043e\u0434\u0430 <code>KStream.selectKey()<\/code>, \u0432 \u043d\u0430\u0448\u0435\u043c \u0441\u043b\u0443\u0447\u0430\u0435 \u0432 <code>KStream&lt;String, Payment><\/code>. \u041f\u043e\u0434 \u043a\u0430\u043f\u043e\u0442\u043e\u043c Kafka Streams \u043f\u043e\u043c\u0435\u0442\u0438\u0442 \u0434\u0430\u043d\u043d\u044b\u0439 KStream \u043a\u0430\u043a \u043f\u043e\u0434\u043b\u0435\u0436\u0430\u0449\u0438\u0439 \u0440\u0435\u043f\u0430\u0440\u0442\u0438\u0446\u0438\u043e\u043d\u0438\u0440\u043e\u0432\u0430\u043d\u0438\u044e, \u0438 \u0435\u0441\u043b\u0438 \u0441\u0440\u0435\u0434\u0438 \u043f\u043e\u0441\u043b\u0435\u0434\u0443\u044e\u0449\u0438\u0445 \u043e\u043f\u0435\u0440\u0430\u0446\u0438\u0439 \u0441\u043e \u0441\u0442\u0440\u0438\u043c\u0430\u043c\u0438 \u0431\u0443\u0434\u0435\u0442 \u043e\u0431\u044a\u0435\u0434\u0438\u043d\u0435\u043d\u0438\u0435 \u0438\u043b\u0438 \u0430\u0433\u0433\u0440\u0435\u0433\u0430\u0446\u0438\u044f, \u0442\u043e \u043f\u0440\u043e\u0438\u0437\u043e\u0439\u0434\u0435\u0442 \u0440\u0435\u043f\u0430\u0440\u0442\u0438\u0446\u0438\u043e\u043d\u0438\u0440\u043e\u0432\u0430\u043d\u0438\u0435: Kafka Streams \u0441\u043e\u0437\u0434\u0430\u0441\u0442 \u0441\u043b\u0443\u0436\u0435\u0431\u043d\u044b\u0439 \u0442\u043e\u043f\u0438\u043a \u0441 \u0442\u0430\u043a\u0438\u043c \u0436\u0435 \u043a\u043e\u043b\u0438\u0447\u0435\u0441\u0442\u0432\u043e\u043c \u043f\u0430\u0440\u0442\u0438\u0446\u0438\u0439, \u043a\u0430\u043a \u0438 \u0432 \u0438\u0441\u0445\u043e\u0434\u043d\u043e\u043c \u0442\u043e\u043f\u0438\u043a\u0435 \u0438 \u0431\u0443\u0434\u0435\u0442 \u043e\u0442\u043f\u0440\u0430\u0432\u043b\u044f\u0442\u044c \u0434\u0430\u043d\u043d\u044b\u0435 \u0438\u0437 \u043f\u0435\u0440\u0432\u043e\u043d\u0430\u0447\u0430\u043b\u044c\u043d\u043e\u0433\u043e \u0442\u043e\u043f\u0438\u043a\u0430 (<code>payments.v1<\/code>) \u0432 \u043d\u043e\u0432\u044b\u0439 \u0442\u043e\u043f\u0438\u043a. \u0422\u0435\u043f\u0435\u0440\u044c \u044d\u0442\u043e\u0442 <code>KStream<\/code>, \u043e\u0441\u043d\u043e\u0432\u0430\u043d\u043d\u044b\u0439 \u043d\u0430 \u0441\u043b\u0443\u0436\u0435\u0431\u043d\u043e\u043c \u0442\u043e\u043f\u0438\u043a\u0435 \u043c\u043e\u0436\u043d\u043e \u0441\u043f\u043e\u043a\u043e\u0439\u043d\u043e \u043e\u0431\u044a\u0435\u0434\u0438\u043d\u044f\u0442\u044c \u0441\u043e \u0441\u0442\u0440\u0438\u043c\u043e\u043c <code>KStream&lt;String, Order><\/code>, \u0442\u0430\u043a \u043a\u0430\u043a \u0443 \u043d\u0438\u0445 \u043e\u0434\u0438\u043d\u0430\u043a\u043e\u0432\u044b\u0435 \u043a\u043b\u044e\u0447\u0438 \u0437\u0430\u043f\u0438\u0441\u0435\u0439.<\/p>\n<\/li>\n<\/ol>\n<p><strong>NB! \u0412\u0410\u0416\u041d\u041e<\/strong> \u0422\u0443\u0442 \u0435\u0441\u0442\u044c \u043e\u0434\u0438\u043d \u043e\u0447\u0435\u043d\u044c \u0432\u0430\u0436\u043d\u044b\u0439 \u043d\u044c\u044e\u0430\u043d\u0441: \u0434\u043b\u044f \u0442\u043e\u0433\u043e, \u0447\u0442\u043e\u0431\u044b Kafka Streams \u043c\u043e\u0433\u043b\u0430 \u043f\u0440\u043e\u0432\u0435\u0441\u0442\u0438 \u043e\u0431\u044a\u0435\u0434\u0438\u043d\u0435\u043d\u0438\u0435 \u0434\u0430\u043d\u043d\u044b\u0445 \u0438\u0441\u0445\u043e\u0434\u043d\u044b\u0435 \u0442\u043e\u043f\u0438\u043a\u0438 \u0434\u043e\u043b\u0436\u043d\u044b \u0431\u044b\u0442\u044c \u043a\u043e-\u043f\u0430\u0440\u0442\u0438\u0446\u0438\u043e\u043d\u0438\u0440\u043e\u0432\u0430\u043d\u044b:<\/p>\n<ul>\n<li>\n<p>\u0434\u043e\u043b\u0436\u043d\u044b \u0431\u044b\u0442\u044c \u043e\u0434\u0438\u043d\u0430\u043a\u043e\u0432\u044b\u0435 \u043a\u043b\u044e\u0447\u0438<\/p>\n<\/li>\n<li>\n<p>\u0434\u043e\u043b\u0436\u043d\u044b \u0438\u043c\u0435\u0442\u044c \u043e\u0434\u0438\u043d\u0430\u043a\u043e\u0432\u043e\u0435 \u043a\u043e\u043b\u0438\u0447\u0435\u0441\u0442\u0432\u043e \u043f\u0430\u0440\u0442\u0438\u0446\u0438\u0439<\/p>\n<\/li>\n<li>\n<p>\u0434\u043e\u043b\u0436\u043d\u044b \u0438\u043c\u0435\u0442\u044c \u043e\u0434\u0438\u043d\u0430\u043a\u043e\u0432\u0443\u044e <a href=\"https:\/\/docs.ksqldb.io\/en\/latest\/developer-guide\/joins\/partition-data\/#records-have-same-partitioning-strategy\" rel=\"noopener noreferrer nofollow\">\u0441\u0442\u0440\u0430\u0442\u0435\u0433\u0438\u044e \u043f\u0430\u0440\u0442\u0438\u0446\u0438\u043e\u043d\u0438\u0440\u043e\u0432\u0430\u043d\u0438\u044f<\/a><\/p>\n<\/li>\n<\/ul>\n<p>\u041f\u043e\u0434\u0440\u043e\u0431\u043d\u0435\u0435 \u043e\u0431 \u044d\u0442\u043e\u043c \u0432 \u0441\u0442\u0430\u0442\u044c\u0435 <a href=\"https:\/\/www.confluent.io\/blog\/co-partitioning-in-kafka-streams\/\" rel=\"noopener noreferrer nofollow\">Co-Partitioning with Apache Kafka<\/a>.<\/p>\n<ol start=\"2\">\n<li>\n<p>\u0418\u043d\u0444\u043e\u0440\u043c\u0430\u0446\u0438\u044f \u043e \u043f\u043e\u043a\u0443\u043f\u0430\u0442\u0435\u043b\u044f\u0445 \u0441\u043e\u0431\u0438\u0440\u0430\u0435\u0442\u0441\u044f \u0432 <code>GlobalKTable&lt;Long, Customer><\/code>. \u042d\u0442\u043e \u043e\u0437\u043d\u0430\u0447\u0430\u0435\u0442, \u0447\u0442\u043e \u043a\u0430\u0436\u0434\u044b\u0439 \u0438\u043d\u0441\u0442\u0430\u043d\u0441 \u043f\u0440\u0438\u043b\u043e\u0436\u0435\u043d\u0438\u044f Email Service \u0431\u0443\u0434\u0435\u0442 \u0438\u043c\u0435\u0442\u044c \u0432 \u0441\u0432\u043e\u0435\u043c \u0440\u0430\u0441\u043f\u043e\u0440\u044f\u0436\u0435\u043d\u0438\u0438 \u0434\u0430\u043d\u043d\u044b\u0435 \u043e\u0431\u043e \u0432\u0441\u0435\u0445 \u043f\u043e\u043a\u0443\u043f\u0430\u0442\u0435\u043b\u044f\u0445, \u0432 \u043e\u0442\u043b\u0438\u0447\u0438\u0435 \u043e\u0442 \u043e\u0431\u044b\u0447\u043d\u043e\u0439 KTable, \u043a\u043e\u0442\u043e\u0440\u0430\u044f \u0445\u0440\u0430\u043d\u0438\u0442 \u043d\u0430 \u0438\u043d\u0441\u0442\u0430\u043d\u0441\u0435 \u0442\u043e\u043b\u044c\u043a\u043e \u0434\u0430\u043d\u043d\u044b\u0435 \u0438\u0437 \u0432\u044b\u0447\u0438\u0442\u044b\u0432\u0430\u0435\u043c\u044b\u0445 \u044d\u0442\u0438\u043c \u0438\u043d\u0441\u0442\u0430\u043d\u0441\u043e\u043c \u043f\u0430\u0440\u0442\u0438\u0446\u0438\u0439. \u0422\u0430\u043a\u0438\u043c \u043e\u0431\u0440\u0430\u0437\u043e\u043c \u043c\u044b \u043c\u043e\u0436\u0435\u043c \u043e\u0431\u044a\u0435\u0434\u0438\u043d\u044f\u0442\u044c KStream \u0441 GlobalKTable \u043f\u043e \u043b\u044e\u0431\u043e\u043c\u0443 \u043f\u043e\u043b\u044e, \u043f\u043b\u044e\u0441 \u043d\u0435\u0442 \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e\u0441\u0442\u0438 \u0441\u043e\u0431\u043b\u044e\u0434\u0430\u0442\u044c \u043f\u0440\u0430\u0432\u0438\u043b\u043e \u043e \u043a\u043e-\u043f\u0430\u0440\u0442\u0438\u0446\u0438\u043e\u043d\u0438\u0440\u043e\u0432\u0430\u043d\u0438\u0438 \u0442\u043e\u043f\u0438\u043a\u043e\u0432. \u041e\u0434\u043d\u0430\u043a\u043e \u0442\u0435\u043f\u0435\u0440\u044c \u043d\u0430\u0448 \u0438\u043d\u0441\u0442\u0430\u043d\u0441 \u043c\u043e\u0436\u0435\u0442 \u0431\u044b\u0442\u044c \u043f\u0435\u0440\u0435\u0433\u0440\u0443\u0436\u0435\u043d \u043a\u0430\u043a \u043f\u043e \u043f\u0430\u043c\u044f\u0442\u0438, \u0442\u0430\u043a \u0438 \u043f\u043e \u0441\u0435\u0442\u0435\u0432\u043e\u043c\u0443 \u0442\u0440\u0430\u0444\u0438\u043a\u0443, \u0435\u0441\u043b\u0438 \u0442\u043e\u043f\u0438\u043a <code>customers<\/code> \u0431\u0443\u0434\u0435\u0442 \u0445\u0440\u0430\u043d\u0438\u0442\u044c \u043e\u0447\u0435\u043d\u044c \u043c\u043d\u043e\u0433\u043e \u0434\u0430\u043d\u043d\u044b\u0445 \u0438 \u043f\u043e\u0441\u0442\u043e\u044f\u043d\u043d\u043e \u043f\u043e\u043f\u043e\u043b\u043d\u044f\u0442\u044c\u0441\u044f. \u0422\u0430\u043a\u0436\u0435 \u0441\u0442\u043e\u0438\u0442 \u043e\u0442\u043c\u0435\u0442\u0438\u0442\u044c, \u0447\u0442\u043e GlobalKTable \u0438\u043d\u0438\u0446\u0438\u0430\u043b\u0438\u0437\u0438\u0440\u0443\u0435\u0442\u0441\u044f \u043d\u0430 \u0441\u0442\u0430\u0440\u0442\u0435 \u043f\u0440\u0438\u043b\u043e\u0436\u0435\u043d\u0438\u044f, \u0442\u043e \u0435\u0441\u0442\u044c \u0432\u044b\u0447\u0438\u0442\u044b\u0432\u0430\u044e\u0442\u0441\u044f \u0432\u0441\u0435 \u0434\u0430\u043d\u043d\u044b\u0435 \u0438\u0437 \u0441\u043e\u043e\u0442\u0432\u0435\u0442\u0441\u0442\u0432\u0443\u044e\u0449\u0435\u0433\u043e \u0442\u043e\u043f\u0438\u043a\u0430. \u0414\u043e \u0442\u043e\u0433\u043e, \u043a\u0430\u043a \u043f\u0440\u043e\u0438\u0437\u043e\u0439\u0434\u0435\u0442 \u0438\u043d\u0438\u0446\u0438\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u044f, \u043e\u0431\u044a\u0435\u0434\u0438\u043d\u0435\u043d\u0438\u0435 KStream \u0441 GlobalKTable \u043d\u0435 \u043d\u0430\u0447\u043d\u0435\u0442\u0441\u044f.<\/p>\n<\/li>\n<li>\n<p>\u041f\u0440\u0438 \u043e\u0431\u044a\u0435\u0434\u0438\u043d\u0435\u043d\u0438\u0438 \u0434\u0432\u0443\u0445 \u0441\u0442\u0440\u0438\u043c\u043e\u0432 \u0432\u0430\u0436\u043d\u043e \u043f\u043e\u043d\u0438\u043c\u0430\u0442\u044c, \u0447\u0442\u043e \u0434\u0430\u043d\u043d\u044b\u0435 \u0432 \u0442\u043e\u043f\u0438\u043a\u0438, \u043d\u0430 \u043e\u0441\u043d\u043e\u0432\u0435 \u043a\u043e\u0442\u043e\u0440\u044b\u0445 \u0441\u043e\u0437\u0434\u0430\u044e\u0442\u0441\u044f \u0441\u0442\u0440\u0438\u043c\u044b \u043c\u043e\u0433\u0443\u0442 \u043f\u043e\u043f\u0430\u0434\u0430\u0442\u044c \u043d\u0435 \u043e\u0434\u043d\u043e\u0432\u0440\u0435\u043c\u0435\u043d\u043d\u043e. \u041f\u043e\u044d\u0442\u043e\u043c\u0443 \u043f\u0440\u0438 \u043e\u0431\u044a\u0435\u0434\u0438\u043d\u0435\u043d\u0438\u0438 \u0441\u0442\u0440\u0438\u043c\u043e\u0432 \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u044e\u0442\u0441\u044f &#171;\u043e\u043a\u043d\u0430&#187; \u043e\u0431\u044a\u0435\u0434\u0438\u043d\u0435\u043d\u0438\u044f. \u0412 \u043d\u0430\u0448\u0435\u043c \u0441\u043b\u0443\u0447\u0430\u0435 \u043f\u0440\u0438 \u043e\u0431\u044a\u0435\u0434\u0438\u043d\u0435\u043d\u0438\u0438 \u0441\u0442\u0440\u0438\u043c\u043e\u0432 orders \u0438 payments \u043c\u044b \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u0435\u043c \u043e\u043a\u043d\u043e \u0432 1 \u043c\u0438\u043d\u0443\u0442\u0443, \u0431\u0435\u0437 \u043a\u0430\u043a\u043e\u0433\u043e-\u043b\u0438\u0431\u043e \u0433\u0440\u0435\u0439\u0441 \u043f\u0435\u0440\u0438\u043e\u0434\u0430.  <code>JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(1))<\/code> \u042d\u0442\u043e \u043e\u0437\u043d\u0430\u0447\u0430\u0435\u0442, \u0447\u0442\u043e \u0437\u0430\u043f\u0438\u0441\u0438, timestamp \u043a\u043e\u0442\u043e\u0440\u044b\u0445 \u0432 \u043f\u0440\u0435\u0434\u0435\u043b\u0430\u0445 1 \u043c\u0438\u043d\u0443\u0442\u044b \u0434\u0440\u0443\u0433 \u043e\u0442 \u0434\u0440\u0443\u0433\u0430, \u043c\u043e\u0433\u0443\u0442 \u0431\u044b\u0442\u044c \u043e\u0431\u044a\u0435\u0434\u0435\u043d\u0438\u043d\u044b, \u0442\u043e \u0435\u0441\u0442\u044c \u043c\u044b \u043f\u043e\u0434\u0440\u0430\u0437\u0443\u043c\u0435\u0432\u0430\u0435\u043c, \u0447\u0442\u043e \u0441\u043e\u0437\u0434\u0430\u043d\u0438\u0435 \u0437\u0430\u043a\u0430\u0437\u0430 \u0438 \u0435\u0433\u043e \u043e\u043f\u043b\u0430\u0442\u0430 \u043f\u043e \u0432\u0440\u0435\u043c\u0435\u043d\u0438 \u043d\u0435 \u0440\u0430\u0437\u043b\u0438\u0447\u0430\u044e\u0442\u0441\u044f \u043d\u0430 +-1 \u043c\u0438\u043d\u0443\u0442\u0443. \u0417\u0430\u043f\u0438\u0441\u0438, \u043a\u043e\u0442\u043e\u0440\u044b\u0435 \u043d\u0435 \u043f\u043e\u043f\u0430\u0434\u0430\u044e\u0442 \u0432 \u044d\u0442\u043e \u043e\u043a\u043d\u043e, \u043d\u0435 \u0431\u0443\u0434\u0443\u0442 \u0443\u0447\u0438\u0442\u044b\u0432\u0430\u0442\u044c\u0441\u044f \u043f\u0440\u0438 \u043e\u0431\u044a\u0435\u0434\u0438\u043d\u0435\u043d\u0438\u0438 \u0441\u0442\u0440\u0438\u043c\u043e\u0432. <\/p>\n<\/li>\n<\/ol>\n<pre><code class=\"java\">    \/**      * Specifies that records of the same key are joinable if their timestamps are within {@code timeDifference},      * i.e., the timestamp of a record from the secondary stream is max {@code timeDifference} before or after      * the timestamp of the record from the primary stream.      * &lt;p>      * CAUTION: Using this method implicitly sets the grace period to zero, which means that any out-of-order      * records arriving after the window ends are considered late and will be dropped.      *      * @param timeDifference join window interval      * @return a new JoinWindows object with the window definition and no grace period. Note that this means out-of-order records arriving after the window end will be dropped      * @throws IllegalArgumentException if {@code timeDifference} is negative or can't be represented as {@code long milliseconds}      *\/     public static JoinWindows ofTimeDifferenceWithNoGrace(final Duration timeDifference) {         return ofTimeDifferenceAndGrace(timeDifference, Duration.ofMillis(NO_GRACE_PERIOD));     }<\/code><\/pre>\n<p>\u0418\u0441\u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u043d\u0438\u0435 &#171;\u043e\u043a\u043e\u043d&#187; \u0442\u0430\u043a\u0436\u0435 \u043e\u0431\u0443\u0441\u043b\u043e\u0432\u043b\u0435\u043d\u043e \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e\u0441\u0442\u044c\u044e \u043f\u043e\u0434\u0434\u0435\u0440\u0436\u0438\u0432\u0430\u0442\u044c \u0431\u0443\u0444\u0435\u0440 \u0434\u043b\u044f \u043e\u0431\u044a\u0435\u0434\u0438\u043d\u0435\u043d\u0438\u044f \u0434\u0430\u043d\u043d\u044b\u0445. \u0415\u0441\u043b\u0438 \u0431\u044b \u043d\u0435 \u0431\u044b\u043b\u043e &#171;\u043e\u043a\u043e\u043d&#187; \u044d\u0442\u043e\u0442 \u0431\u0443\u0444\u0435\u0440 \u043c\u043e\u0433 \u0431\u044b \u043f\u043e\u0433\u043b\u043e\u0442\u0438\u0442\u044c \u0432\u0441\u044e \u043f\u0430\u043c\u044f\u0442\u044c. &#171;\u041e\u043a\u043d\u0430&#187; \u0445\u0440\u0430\u043d\u044f\u0442 \u0441\u0432\u043e\u0438 \u0434\u0430\u043d\u043d\u044b\u0435 \u0432 windowing state store, \u0441\u0442\u0430\u0440\u044b\u0435 \u0437\u0430\u043f\u0438\u0441\u0438 \u0432 \u043a\u043e\u0442\u043e\u0440\u043e\u043c \u0443\u0434\u0430\u043b\u044f\u044e\u0442\u0441\u044f \u043f\u043e\u0441\u043b\u0435 \u043d\u0430\u0441\u0442\u0440\u0430\u0438\u0432\u0430\u0435\u043c\u043e\u0433\u043e window retention period, \u043a\u043e\u0442\u043e\u0440\u044b\u0439 \u043f\u043e \u0443\u043c\u043e\u043b\u0447\u0430\u043d\u0438\u044e \u0440\u0430\u0432\u0435\u043d 1 \u0434\u043d\u044e. \u0418\u0437\u043c\u0435\u043d\u0438\u0442\u044c \u044d\u0442\u043e \u0437\u043d\u0430\u0447\u0435\u043d\u0438\u0435 \u043c\u043e\u0436\u043d\u043e \u0441 \u043f\u043e\u043c\u043e\u0449\u044c\u044e: <code>Materialized#withRetention()<\/code>.<\/p>\n<p>\u0411\u043e\u043b\u0435\u0435 \u043f\u043e\u0434\u0440\u043e\u0431\u043d\u043e \u043e \u0442\u043e\u043c, \u043a\u0430\u043a\u0438\u0435 \u0431\u044b\u0432\u0430\u044e\u0442 \u0432\u0438\u0434\u044b \u043e\u0431\u044a\u0435\u0434\u0438\u043d\u0435\u043d\u0438\u0439 \u043c\u043e\u0436\u043d\u043e \u043f\u0440\u043e\u0447\u0438\u0442\u0430\u0442\u044c \u0432 <a href=\"https:\/\/kafka.apache.org\/20\/documentation\/streams\/developer-guide\/dsl-api.html#joining\" rel=\"noopener noreferrer nofollow\">\u0434\u043e\u043a\u0443\u043c\u0435\u043d\u0442\u0430\u0446\u0438\u0438<\/a>.<\/p>\n<h3>Fraud Service <\/h3>\n<figure class=\"full-width\"><img loading=\"lazy\" decoding=\"async\" src=\"https:\/\/habrastorage.org\/r\/w1560\/getpro\/habr\/upload_files\/2fa\/6df\/b32\/2fa6dfb323abe735b579af220b0ea450.png\" width=\"1722\" height=\"873\" data-src=\"https:\/\/habrastorage.org\/getpro\/habr\/upload_files\/2fa\/6df\/b32\/2fa6dfb323abe735b579af220b0ea450.png\"\/><\/figure>\n<p>\u0421\u0435\u0440\u0432\u0438\u0441 \u043e\u0442\u0432\u0435\u0447\u0430\u0435\u0442 \u0437\u0430 \u043f\u043e\u0438\u0441\u043a \u043f\u043e\u0442\u0435\u043d\u0446\u0438\u0430\u043b\u044c\u043d\u043e \u043c\u043e\u0448\u0435\u043d\u043d\u0438\u0447\u0435\u0441\u043a\u0438\u0445 \u0442\u0440\u0430\u043d\u0437\u0430\u043a\u0446\u0438\u0439: \u0435\u0441\u043b\u0438 \u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u0435\u043b\u044c \u0441\u0434\u0435\u043b\u0430\u043b \u0437\u0430\u043a\u0430\u0437\u044b \u043d\u0430 \u0431\u043e\u043b\u0435\u0435 \u0447\u0435\u043c 2 \u0442\u044b\u0441 \u0432 \u0442\u0435\u0447\u0435\u043d\u0438\u0435 \u0441\u0435\u0441\u0441\u0438\u0438 \u0432\u0437\u0430\u0438\u043c\u043e\u0434\u0435\u0439\u0441\u0442\u0432\u0438\u044f \u0441 \u0441\u0435\u0440\u0432\u0435\u0440\u043e\u043c (\u043f\u043e\u0434\u0440\u043e\u0431\u043d\u0435\u0435 \u043d\u0438\u0436\u0435 \u043f\u043e \u0442\u0435\u043a\u0441\u0442\u0443), \u0442\u043e \u0432\u0441\u0435 \u0442\u0430\u043a\u0438\u0435 \u0437\u0430\u043a\u0430\u0437\u044b \u043d\u0435 \u043f\u0440\u043e\u0445\u043e\u0434\u044f\u0442 \u0432\u0430\u043b\u0438\u0434\u0430\u0446\u0438\u044e.<\/p>\n<p>\u041a\u043e\u043d\u0444\u0438\u0433\u0443\u0440\u0430\u0446\u0438\u044f Kafka Streams \u043d\u0435\u043c\u043d\u043e\u0433\u043e \u043e\u0442\u043b\u0438\u0447\u0430\u0435\u0442\u0441\u044f \u043e\u0442 \u043f\u0440\u0435\u0434\u044b\u0434\u0443\u0449\u0438\u0445:<\/p>\n<pre><code class=\"kotlin\">    @Bean(name = [KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME])     fun kStreamsConfig(): KafkaStreamsConfiguration {         val props = hashMapOf&lt;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)     }<\/code><\/pre>\n<p>\u0421\u0442\u043e\u0438\u0442 \u043f\u043e\u044f\u0441\u043d\u0438\u0442\u044c, \u0447\u0442\u043e \u0432 \u044d\u0442\u043e\u043c \u0441\u0435\u0440\u0432\u0438\u0441\u0435 \u0431\u0443\u0434\u0435\u0442 \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u044c\u0441\u044f \u0430\u0433\u0433\u0440\u0435\u0433\u0430\u0446\u0438\u044f \u0434\u0430\u043d\u043d\u044b\u0445, \u0432\u0441\u043b\u0435\u0434\u0441\u0442\u0432\u0438\u0435 \u0447\u0435\u0433\u043e Kafka Streams \u0431\u0443\u0434\u0435\u0442 \u0437\u0430\u0434\u0435\u0439\u0441\u0442\u0432\u043e\u0432\u0430\u0442\u044c state store \u0434\u043b\u044f \u0445\u0440\u0430\u043d\u0435\u043d\u0438\u044f \u0441\u043e\u0441\u0442\u043e\u044f\u043d\u0438\u044f \u0430\u0433\u0433\u0440\u0435\u0433\u0438\u0440\u043e\u0432\u0430\u043d\u043d\u044b\u0445 \u0434\u0430\u043d\u043d\u044b\u0445. \u0412 \u043d\u0430\u0448\u0435\u043c \u0441\u043b\u0443\u0447\u0430\u0435 \u043c\u044b \u043e\u0442\u043a\u043b\u044e\u0447\u0430\u0435\u043c \u043a\u044d\u0448\u0438\u0440\u043e\u0432\u0430\u043d\u0438\u0435 \u0434\u0430\u043d\u043d\u044b\u0445 \u043d\u0430 \u0438\u043d\u0441\u0442\u0430\u043d\u0441\u0435, \u0442\u0430\u043a \u043a\u0430\u043a \u043a\u044d\u0448\u0438\u0440\u043e\u0432\u0430\u043d\u0438\u0435 \u043c\u043e\u0436\u0435\u0442 \u043f\u0440\u0438\u0432\u0435\u0441\u0442\u0438 \u043a \u043e\u0431\u044a\u0435\u0434\u0438\u043d\u0435\u043d\u0438\u044e \u043f\u043e\u0441\u043b\u0435\u0434\u043e\u0432\u0430\u0442\u0435\u043b\u044c\u043d\u044b\u0445 \u0441\u043e\u0431\u044b\u0442\u0438\u0439 \u043f\u0440\u0438 \u0440\u0430\u0431\u043e\u0442\u0435 \u0441\u043e state store \u0432\u043e \u0432\u0440\u0435\u043c\u044f \u0430\u0433\u0433\u0440\u0435\u0433\u0430\u0446\u0438\u0438 \u0434\u0430\u043d\u043d\u044b\u0445, \u0432 \u0442\u043e \u0432\u0440\u0435\u043c\u044f \u043a\u0430\u043a \u043c\u044b \u0445\u043e\u0442\u0438\u043c \u043e\u0431\u0440\u0430\u0431\u0430\u0442\u044b\u0432\u0430\u0442\u044c \u043a\u0430\u0436\u0434\u0443\u044e \u0437\u0430\u043f\u0438\u0441\u044c \u0432 \u043f\u0440\u043e\u0446\u0435\u0441\u0441\u0435 \u0430\u0433\u0433\u0440\u0435\u0433\u0430\u0446\u0438\u0438. \u0411\u043e\u043b\u0435\u0435 \u043f\u043e\u0434\u0440\u043e\u0431\u043d\u043e \u043e\u0431 \u044d\u0442\u043e\u043c \u0432 \u0441\u0442\u0430\u0442\u044c\u0435 <a href=\"https:\/\/docs.confluent.io\/platform\/current\/streams\/developer-guide\/memory-mgmt.html\" rel=\"noopener noreferrer nofollow\">Kafka Streams Memory Management<\/a>.<\/p>\n<p>\u041d\u0430 \u043f\u0435\u0440\u0432\u043e\u043c \u044d\u0442\u0430\u043f\u0435 \u0440\u0430\u0431\u043e\u0442\u044b \u0441\u0435\u0440\u0432\u0438\u0441\u0430 \u0441\u043e\u0437\u0434\u0430\u0435\u0442\u0441\u044f \u0443\u0437\u0435\u043b \u0430\u0433\u0433\u0440\u0435\u0433\u0430\u0446\u0438\u0438 \u0441\u0442\u0440\u0438\u043c\u0430 <code>KStream&lt;String, Order><\/code>:<\/p>\n<pre><code class=\"kotlin\">val orders: KStream&lt;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&lt;Windowed&lt;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)     )<\/code><\/pre>\n<p>\u0421\u0442\u043e\u0438\u0442 \u043e\u0431\u0440\u0430\u0442\u0438\u0442\u044c \u0432\u043d\u0438\u043c\u0430\u043d\u0438\u0435 \u043d\u0430 \u0442\u043e, \u0447\u0442\u043e \u043f\u0440\u0438 \u0433\u0440\u0443\u043f\u043f\u0438\u0440\u043e\u0432\u043a\u0435 \u043f\u043e \u043a\u043b\u044e\u0447\u0443, \u043e\u0442\u043b\u0438\u0447\u0430\u044e\u0449\u0435\u043c\u0443\u0441\u044f \u043e\u0442 \u043a\u043b\u044e\u0447\u0430 \u0438\u0441\u0445\u043e\u0434\u043d\u043e\u0433\u043e \u0442\u043e\u043f\u0438\u043a\u0430 \u0431\u0443\u0434\u0435\u0442 \u0441\u043e\u0437\u0434\u0430\u043d \u0432\u043d\u0443\u0442\u0440\u0435\u043d\u043d\u0438\u0439 \u0441\u043b\u0443\u0436\u0435\u0431\u043d\u044b\u0439 \u0442\u043e\u043f\u0438\u043a Kafka, \u0441 \u0442\u0430\u043a\u0438\u043c \u0436\u0435 \u043a\u043e\u043b\u0438\u0447\u0435\u0441\u0442\u0432\u043e\u043c \u043f\u0430\u0440\u0442\u0438\u0446\u0438\u0439, \u0447\u0442\u043e \u0438 \u0438\u0441\u0445\u043e\u0434\u043d\u044b\u0439 \u0442\u043e\u043f\u0438\u043a. \u0412 \u043d\u0430\u0448\u0435\u043c \u043f\u0440\u0438\u043c\u0435\u0440\u0435 \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u0435\u0442\u0441\u044f \u0435\u0449\u0435 \u043e\u0434\u0438\u043d \u0442\u0438\u043f \u043e\u043a\u043d\u0430 \u0434\u043b\u044f \u0430\u0433\u0433\u0440\u0435\u0433\u0430\u0446\u0438\u0438 \u0434\u0430\u043d\u043d\u044b\u0445 &#8212; <code>SessionWindows.ofInactivityGapWithNoGrace(Duration.ofHours(1))<\/code><br \/> \u042d\u0442\u043e \u043e\u0437\u043d\u0430\u0447\u0430\u0435\u0442, \u0447\u0442\u043e \u0434\u0430\u043d\u043d\u044b\u0435 \u0431\u0443\u0434\u0443\u0442 \u043f\u043e\u043f\u0430\u0434\u0430\u0442\u044c \u0432 \u0430\u0433\u0433\u0440\u0435\u0433\u0430\u0446\u0438\u044e \u0432 \u0442\u0435\u0447\u0435\u043d\u0438\u0435 \u0441\u0435\u0441\u0441\u0438\u0438 \u0430\u043a\u0442\u0438\u0432\u043d\u043e\u0441\u0442\u0438 \u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u0435\u043b\u044f. \u0421\u0435\u0441\u0441\u0438\u044f \u0437\u0430\u043a\u0440\u044b\u0432\u0430\u0435\u0442\u0441\u044f, \u043a\u043e\u0433\u0434\u0430 \u0432 \u0442\u043e\u043f\u0438\u043a \u043d\u0435 \u043f\u043e\u0441\u0442\u0443\u043f\u0430\u044e\u0442 \u0434\u0430\u043d\u043d\u044b\u0435 \u0432 \u0442\u0435\u0447\u0435\u043d\u0438\u0435 \u0443\u043a\u0430\u0437\u0430\u043d\u043d\u043e\u0433\u043e \u043f\u0435\u0440\u0438\u043e\u0434\u0430 \u0432\u0440\u0435\u043c\u0435\u043d\u0438. \u0411\u043e\u043b\u0435\u0435 \u043f\u043e\u0434\u0440\u043e\u0431\u043d\u043e \u0432 \u0441\u0442\u0430\u0442\u044c\u0435 <a href=\"https:\/\/developer.confluent.io\/tutorials\/create-session-windows\/confluent.html\" rel=\"noopener noreferrer nofollow\">Create session windows<\/a>.<\/p>\n<p>\u041d\u0430 \u0432\u0442\u043e\u0440\u043e\u043c \u044d\u0442\u0430\u043f\u0435 \u043c\u044b \u0438\u0437\u0431\u0430\u0432\u043b\u044f\u0435\u043c\u0441\u044f \u043e\u0442 \u0434\u0430\u043d\u043d\u044b\u0445 \u043e\u0431 &#171;\u043e\u043a\u043d\u0430\u0445&#187; \u0438 \u0434\u0435\u043b\u0430\u0435\u043c re-keying \u0437\u0430\u043f\u0438\u0441\u0435\u0439 \u043f\u043e <code>orderId<\/code>:<\/p>\n<pre><code class=\"kotlin\">val ordersWithTotals: KStream&lt;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 }<\/code><\/pre>\n<p>\u041d\u0430 \u0442\u0440\u0435\u0442\u044c\u0435\u043c \u044d\u0442\u0430\u043f\u0435 \u043c\u044b \u0444\u043e\u0440\u043c\u0438\u0440\u0443\u0435\u043c \u0434\u0432\u0435 &#171;\u0432\u0435\u0442\u043a\u0438&#187; \u0441\u0442\u0440\u0438\u043c\u043e\u0432:<\/p>\n<ul>\n<li>\n<p>\u043e\u0434\u043d\u0430 \u0441\u043e\u0434\u0435\u0440\u0436\u0438\u0442 \u0437\u043d\u0430\u0447\u0435\u043d\u0438\u044f \u0431\u043e\u043b\u044c\u0448\u0435 FRAUD_LIMIT<\/p>\n<\/li>\n<li>\n<p>\u0432\u0442\u043e\u0440\u0430\u044f &#8212; \u043c\u0435\u043d\u044c\u0448\u0435 FRAUD_LIMIT<\/p>\n<\/li>\n<\/ul>\n<pre><code class=\"kotlin\">\/\/Now branch the stream into two, for pass and fail, based on whether the windowed \/\/ total is over Fraud Limit val forks: Map&lt;String, KStream&lt;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 &lt; FRAUD_LIMIT },         Branched.`as`(\"below\")     )     .noDefaultBranch()<\/code><\/pre>\n<p>\u041d\u0430 \u0441\u043b\u0435\u0434\u0443\u044e\u0449\u0435\u043c \u044d\u0442\u0430\u043f\u0435 \u0437\u043d\u0430\u0447\u0435\u043d\u0438\u044f \u0438\u0437 \u043a\u0430\u0436\u0434\u043e\u0439 &#171;\u0432\u0435\u0442\u043a\u0438&#187; \u0431\u0443\u0434\u0443\u0442 \u043e\u0442\u043f\u0440\u0430\u0432\u043b\u0435\u043d\u044b \u0432 \u0442\u043e\u043f\u0438\u043a <code>order-validations.v1<\/code>, \u0441 \u0441\u043e\u043e\u0442\u0432\u0435\u0442\u0441\u0442\u0432\u0443\u044e\u0449\u0438\u043c \u0440\u0435\u0437\u0443\u043b\u044c\u0442\u0430\u0442\u043e\u043c: FAIL \u0438\u043b\u0438 PASS.<\/p>\n<pre><code class=\"kotlin\">val keySerde = schemas.ORDER_VALIDATIONS.keySerde val valueSerde = schemas.ORDER_VALIDATIONS.valueSerde val config = hashMapOf&lt;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    ) )<\/code><\/pre>\n<h4>\u041f\u0440\u0438\u043c\u0435\u0447\u0430\u043d\u0438\u0435.<\/h4>\n<p>\u0412 \u043f\u0440\u043e\u0446\u0435\u0441\u0441\u0435 \u0442\u0435\u0441\u0442\u0438\u0440\u043e\u0432\u0430\u043d\u0438\u044f \u0440\u0435\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u0438 \u0443 \u043c\u0435\u043d\u044f \u0432\u043e\u0437\u043d\u0438\u043a\u043b\u0430 \u043f\u0440\u043e\u0431\u043b\u0435\u043c\u0430 \u0441 \u0442\u0435\u043c, \u0447\u0442\u043e \u0432 \u043d\u0435\u043a\u043e\u0442\u043e\u0440\u044b\u0445 \u0441\u043b\u0443\u0447\u0430\u044f\u0445 Kafka Streams \u043f\u044b\u0442\u0430\u043b\u0430\u0441\u044c \u043e\u0442\u043f\u0440\u0430\u0432\u0438\u0442\u044c \u0434\u0430\u043d\u043d\u044b\u0435 \u0432 \u0442\u043e\u043f\u0438\u043a, \u043f\u0440\u0435\u0434\u0432\u0430\u0440\u0438\u0442\u0435\u043b\u044c\u043d\u043e \u0430\u0432\u0442\u043e\u043c\u0430\u0442\u0438\u0447\u0435\u0441\u043a\u0438 \u0437\u0430\u0440\u0435\u0433\u0438\u0441\u0442\u0440\u0438\u0440\u043e\u0432\u0430\u0432 \u043d\u043e\u0432\u0443\u044e \u0441\u0445\u0435\u043c\u0443 \u0434\u0430\u043d\u043d\u044b\u0445 \u0432 Schema Registry, \u043f\u0440\u0438 \u044d\u0442\u043e\u043c \u043d\u043e\u0432\u0430\u044f \u0441\u0445\u0435\u043c\u0430 \u0431\u044b\u043b\u0430 \u043d\u0435\u0441\u043e\u0432\u043c\u0435\u0441\u0442\u0438\u043c\u0430 \u0441\u043e \u0441\u0442\u0430\u0440\u043e\u0439, \u0447\u0442\u043e \u043e\u0441\u0442\u0430\u043d\u0430\u0432\u043b\u0438\u0432\u0430\u043b\u043e Kafka Streams. \u041f\u043e\u043f\u044b\u0442\u043a\u0438 \u043e\u0442\u043a\u043b\u044e\u0447\u0438\u0442\u044c \u0430\u0432\u0442\u043e\u043c\u0430\u0442\u0438\u0447\u0435\u0441\u043a\u0443\u044e \u0440\u0435\u0433\u0438\u0441\u0442\u0440\u0430\u0446\u0438\u044e \u0441\u0445\u0435\u043c \u0434\u043b\u044f \u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u0430\u0442\u043e\u0440\u043e\u0432 \/ \u0434\u0435\u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u0430\u0442\u043e\u0440\u043e\u0432 Kafka Streams \u0442\u0430\u043a\u0436\u0435 \u043f\u0440\u0438\u0432\u043e\u0434\u0438\u043b\u0438 \u043a \u043e\u0441\u0442\u0430\u043d\u043e\u0432\u043a\u0435 \u043f\u0440\u0438\u043b\u043e\u0436\u0435\u043d\u0438\u044f \u0438\u0437-\u0437\u0430 \u0442\u043e\u0433\u043e, \u0447\u0442\u043e Kafka Streams \u0440\u0435\u0433\u0443\u043b\u044f\u0440\u043d\u043e \u043f\u043e\u0434 \u043a\u0430\u043f\u043e\u0442\u043e\u043c \u0441\u043e\u0437\u0434\u0430\u0435\u0442 \u0441\u043b\u0443\u0436\u0435\u0431\u043d\u044b\u0435 \u0442\u043e\u043f\u0438\u043a\u0438 \u0438 \u0434\u043e\u043b\u0436\u043d\u0430 \u0438\u043c\u0435\u0442\u044c \u0432\u043e\u0437\u043c\u043e\u0436\u043d\u043e\u0441\u0442\u044c \u0440\u0435\u0433\u0438\u0441\u0442\u0440\u0438\u0440\u043e\u0432\u0430\u0442\u044c \u0438\u0445 \u0441\u0445\u0435\u043c\u044b. \u041f\u0440\u0438 \u044d\u0442\u043e\u043c \u0441\u0442\u043e\u0438\u0442 \u0437\u0430\u043c\u0435\u0442\u0438\u0442\u044c, \u0447\u0442\u043e \u043e\u0441\u0442\u0430\u043d\u043e\u0432\u043a\u0430 \u043f\u0440\u0438\u043b\u043e\u0436\u0435\u043d\u0438\u044f Kafka Streams \u043d\u0435 \u043f\u0440\u0438\u0432\u043e\u0434\u0438\u0442 \u043a \u043f\u0430\u0434\u0435\u043d\u0438\u044e \u0438\u043d\u0441\u0442\u0430\u043d\u0441\u0430 &#8212; \u043e\u043d-\u0442\u043e \u043a\u0430\u043a \u0440\u0430\u0437 \u0438 \u043f\u0440\u043e\u0434\u043e\u043b\u0436\u0430\u0435\u0442 \u0440\u0430\u0431\u043e\u0442\u0430\u0442\u044c, \u043f\u043e\u044d\u0442\u043e\u043c\u0443 \u0432 \u043f\u0440\u043e\u0434\u0435 \u043f\u0440\u0438\u0434\u0435\u0442\u0441\u044f \u043d\u0430\u0441\u0442\u0440\u0430\u0438\u0432\u0430\u0442\u044c \u043e\u0442\u0434\u0435\u043b\u044c\u043d\u044b\u0439 \u043c\u043e\u043d\u0438\u0442\u043e\u0440\u0438\u043d\u0433 \u0436\u0438\u0437\u043d\u0435\u0441\u043f\u043e\u0441\u043e\u0431\u043d\u043e\u0441\u0442\u0438 \u043f\u0440\u0438\u043b\u043e\u0436\u0435\u043d\u0438\u0439 Kafka Streams, \u0447\u0442\u043e\u0431\u044b \u0432 \u0441\u043b\u0443\u0447\u0430\u0435 \u043e\u0441\u0442\u0430\u043d\u043e\u0432\u043a\u0438 \u0441\u0442\u0440\u0438\u043c\u043e\u0432 \u043f\u0435\u0440\u0435\u0437\u0430\u043f\u0443\u0441\u0442\u0438\u0442\u044c \u0438\u043d\u0441\u0442\u0430\u043d\u0441.<\/p>\n<p>\u041f\u0443\u0442\u0435\u043c \u0434\u043e\u043b\u0433\u0438\u0445 \u043c\u044b\u0442\u0430\u0440\u0441\u0442\u0432 \u0440\u0435\u0448\u0435\u043d\u0438\u0435 (\u0441\u043a\u043e\u0440\u0435\u0435 \u0437\u0430\u043f\u043b\u0430\u0442\u043a\u0430) \u0431\u044b\u043b\u043e \u043d\u0430\u0439\u0434\u0435\u043d\u043e: \u0442\u0430\u043c, \u0433\u0434\u0435 Kafka Streams \u043e\u0442\u043f\u0440\u0430\u0432\u043b\u044f\u0435\u0442 \u0430\u0433\u0433\u0440\u0435\u0433\u0438\u0440\u043e\u0432\u0430\u043d\u043d\u044b\u0435 \u0438\u043b\u0438 \u043e\u0431\u044a\u0435\u0434\u0438\u043d\u0435\u043d\u043d\u044b\u0435 \u0434\u0430\u043d\u043d\u044b\u0435 \u0432 \u0443\u0436\u0435 \u0441\u043e\u0437\u0434\u0430\u043d\u043d\u044b\u0439 \u0442\u043e\u043f\u0438\u043a Kafka \u0441 \u0437\u0430\u0440\u0435\u0433\u0438\u0441\u0442\u0440\u0438\u0440\u043e\u0432\u0430\u043d\u043d\u043e\u0439 \u0432 Schema Registry \u0441\u0445\u0435\u043c\u043e\u0439, \u044f \u043e\u0442\u043a\u043b\u044e\u0447\u0438\u043b \u0432\u043e\u0437\u043c\u043e\u0436\u043d\u043e\u0441\u0442\u044c Serdes \u0430\u0432\u0442\u043e\u043c\u0430\u0442\u0438\u0447\u0435\u0441\u043a\u0438 \u0440\u0435\u0433\u0438\u0441\u0442\u0440\u0438\u0440\u043e\u0432\u0430\u0442\u044c \u043d\u043e\u0432\u044b\u0435 \u0441\u0445\u0435\u043c\u044b.<\/p>\n<h3>Inventory Service<\/h3>\n<p>\u0412\u044b\u0447\u0438\u0442\u044b\u0432\u0430\u0435\u0442 \u0434\u0430\u043d\u043d\u044b\u0435 \u043e \u0437\u0430\u043a\u0430\u0437\u0430\u0445 \u0438\u0437 \u0442\u043e\u043f\u0438\u043a\u0430 <code>orders.v1<\/code> \u0438 \u0432\u0430\u043b\u0438\u0434\u0438\u0440\u0443\u0435\u0442 \u043d\u0430 \u043f\u0440\u0435\u0434\u043c\u0435\u0442 \u043d\u0430\u043b\u0438\u0447\u0438\u044f \u0442\u0440\u0435\u0431\u0443\u0435\u043c\u043e\u0433\u043e \u043a\u043e\u043b\u0438\u0447\u0435\u0441\u0442\u0432\u0430 \u0442\u043e\u0432\u0430\u0440\u043e\u0432 \u0432 \u0437\u0430\u043a\u0430\u0437\u0435. \u0412\u0430\u043b\u0438\u0434\u0430\u0446\u0438\u044f \u0443\u0447\u0438\u0442\u044b\u0432\u0430\u0435\u0442 \u043a\u0430\u043a \u0442\u0435 \u0442\u043e\u0432\u0430\u0440\u044b, \u043a\u043e\u0442\u043e\u0440\u044b\u0435 \u0435\u0441\u0442\u044c \u043d\u0430 &#171;\u0441\u043a\u043b\u0430\u0434\u0435&#187;, \u0442\u0430\u043a \u0438 \u0442\u0435, \u043a\u043e\u0442\u043e\u0440\u044b\u0435 \u0437\u0430\u0440\u0435\u0437\u0435\u0440\u0432\u0438\u0440\u043e\u0432\u0430\u043d\u044b (\u043d\u0430 \u043d\u0438\u0445 \u0435\u0441\u0442\u044c \u0434\u0435\u0439\u0441\u0442\u0432\u0443\u044e\u0449\u0438\u0435 \u0437\u0430\u043a\u0430\u0437\u044b). \u041d\u0430 \u0442\u0435\u043a\u0443\u0449\u0438\u0439 \u043c\u043e\u043c\u0435\u043d\u0442, \u043d\u0435 \u0440\u0435\u0430\u043b\u0438\u0437\u043e\u0432\u0430\u043d\u0430 \u043b\u043e\u0433\u0438\u043a\u0430 \u0441\u043d\u0438\u0436\u0435\u043d\u0438\u044f \u043a\u043e\u043b\u0438\u0447\u0435\u0441\u0442\u0432\u0430 \u0442\u043e\u0432\u0430\u0440\u043e\u0432 \u043d\u0430 &#171;\u0441\u043a\u043b\u0430\u0434\u0435&#187; &#8212; \u044d\u0442\u043e \u0434\u043e\u043b\u0436\u043d\u043e \u043f\u0440\u043e\u0438\u0441\u0445\u043e\u0434\u0438\u0442\u044c, \u043d\u0430\u043f\u0440\u0438\u043c\u0435\u0440, \u043f\u0440\u0438 \u043e\u0442\u043f\u0440\u0430\u0432\u043a\u0435 \u0437\u0430\u043a\u0430\u0437\u0430 \u043f\u043e\u043a\u0443\u043f\u0430\u0442\u0435\u043b\u044e \u0447\u0435\u0440\u0435\u0437 \u0441\u043b\u0443\u0436\u0431\u0443 \u0434\u043e\u0441\u0442\u0430\u0432\u043a\u0438. &#171;\u0421\u043a\u043b\u0430\u0434&#187; &#8212; \u043f\u0440\u0435\u0434\u0441\u0442\u0430\u0432\u043b\u044f\u0435\u0442 \u0438\u0437 \u0441\u0435\u0431\u044f \u0442\u043e\u043f\u0438\u043a Kafka <code>warehouse-inventory.v1<\/code>, \u0433\u0434\u0435 \u043a\u043b\u044e\u0447\u043e\u043c \u0437\u0430\u043f\u0438\u0441\u0438 \u044f\u0432\u043b\u044f\u0435\u0442\u0441\u044f \u043d\u0430\u0437\u0432\u0430\u043d\u0438\u0435 \u0442\u043e\u0432\u0430\u0440\u0430, \u0430 \u0437\u043d\u0430\u0447\u0435\u043d\u0438\u0435\u043c &#8212; \u043a\u043e\u043b\u0438\u0447\u0435\u0441\u0442\u0432\u043e.<\/p>\n<p>\u041e\u0441\u043d\u043e\u0432\u043d\u043e\u0439 \u0441\u0435\u0440\u0432\u0438\u0441 <code>InventoryKafkaService<\/code> \u0441\u043e\u0437\u0434\u0430\u0435\u0442 <code>KStream&lt;String, Order><\/code> \u043d\u0430 \u043e\u0441\u043d\u043e\u0432\u0435 \u0442\u043e\u043f\u0438\u043a\u0430 <code>orders.v1<\/code>:<\/p>\n<pre><code class=\"kotlin\">val orders: KStream&lt;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)    }<\/code><\/pre>\n<p><code>KTable&lt;Product, Int><\/code> \u043d\u0430 \u043e\u0441\u043d\u043e\u0432\u0435 \u0442\u043e\u043f\u0438\u043a\u0430 <code>warehouse-inventory.v1<\/code>:<\/p>\n<pre><code class=\"kotlin\">val warehouseInventory: KTable&lt;Product, Int> = builder     .table(        schemas.WAREHOUSE_INVENTORY.name,        Consumed.with(           schemas.WAREHOUSE_INVENTORY.keySerde,           schemas.WAREHOUSE_INVENTORY.valueSerde        )     )<\/code><\/pre>\n<p>\u0430 \u0442\u0430\u043a\u0436\u0435 \u043c\u0430\u0442\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u043e\u0432\u0430\u043d\u043d\u044b\u0439 <code>KeyValueStore&lt;Product, Long><\/code>, \u0445\u0440\u0430\u043d\u044f\u0449\u0438\u0439 \u0438\u043d\u0444\u043e\u0440\u043c\u0430\u0446\u0438\u044e \u043e \u0437\u0430\u0440\u0435\u0437\u0435\u0440\u0432\u0438\u0440\u043e\u0432\u0430\u043d\u043d\u044b\u0445 \u0442\u043e\u0432\u0430\u0440\u0430\u0445:<\/p>\n<pre><code class=\"kotlin\">val reservedStock: StoreBuilder&lt;KeyValueStore&lt;Product, Long>> = Stores     .keyValueStoreBuilder(        Stores.persistentKeyValueStore(inventoryProps.reservedStockStoreName),        schemas.WAREHOUSE_INVENTORY.keySerde, Serdes.Long()     )     .withLoggingEnabled(hashMapOf()) builder.addStateStore(reservedStock)<\/code><\/pre>\n<p>\u041f\u043e\u0441\u043b\u0435 \u044d\u0442\u043e\u0433\u043e \u043c\u044b \u0434\u0435\u043b\u0430\u0435\u043c re-keying \u0441\u0442\u0440\u0438\u043c\u0430 <code>KStream&lt;String, Order><\/code> \u043f\u043e \u043a\u043b\u044e\u0447\u0443 product. \u0422\u0443\u0442 \u0441\u0442\u043e\u0438\u0442 \u043d\u0430\u043f\u043e\u043c\u043d\u0438\u0442\u044c, \u0447\u0442\u043e \u043f\u043e\u0434 \u043a\u0430\u043f\u043e\u0442\u043e\u043c Kafka Streams \u043f\u043e\u043c\u0435\u0442\u0438\u0442 \u044d\u0442\u043e\u0442 KStream \u043a\u0430\u043a \u043f\u043e\u0434\u043b\u0435\u0436\u0430\u0449\u0438\u0439 \u0440\u0435\u043f\u0430\u0440\u0442\u0438\u0446\u0438\u043e\u043d\u0438\u0440\u043e\u0432\u0430\u043d\u0438\u044e, \u0438 \u0432 \u0441\u043b\u0443\u0447\u0430\u0435 \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e\u0441\u0442\u0438 (\u043d\u0438\u0436\u0435 \u0431\u0443\u0434\u0435\u0442 join \u0438\u043b\u0438 \u0430\u0433\u0433\u0440\u0435\u0433\u0430\u0446\u0438\u044f), \u043f\u0440\u043e\u0438\u0437\u043e\u0439\u0434\u0435\u0442 \u0440\u0435\u043f\u0430\u0440\u0442\u0438\u0446\u0438\u043e\u043d\u0438\u0440\u043e\u0432\u0430\u043d\u0438\u0435. \u0414\u0430\u043b\u0435\u0435 \u0432 \u0441\u0442\u0440\u0438\u043c \u043f\u043e\u043f\u0430\u0434\u0430\u044e\u0442 \u0442\u043e\u043b\u044c\u043a\u043e \u0437\u0430\u043a\u0430\u0437\u044b \u0432 \u0441\u043e\u0441\u0442\u043e\u044f\u043d\u0438\u0438 CREATED, \u043a\u043e\u0442\u043e\u0440\u044b\u0435 \u043e\u0431\u044a\u0435\u0434\u0438\u043d\u044f\u044e\u0442\u0441\u044f \u0441 \u0442\u0430\u0431\u043b\u0438\u0446\u0435\u0439 \u0437\u0430\u0440\u0435\u0437\u0435\u0440\u0432\u0438\u0440\u043e\u0432\u0430\u043d\u043d\u044b\u0445 \u0442\u043e\u0432\u0430\u0440\u043e\u0432. \u041f\u043e\u0441\u043b\u0435 \u043e\u0431\u044a\u0435\u0434\u0438\u043d\u0435\u043d\u0438\u044f \u043f\u043e\u043b\u0443\u0447\u0430\u0435\u043c \u043d\u043e\u0432\u044b\u0439 <code>KStream<\/code>, \u043a\u0430\u0436\u0434\u0430\u044f \u0437\u0430\u043f\u0438\u0441\u044c \u043a\u043e\u0442\u043e\u0440\u043e\u0433\u043e \u043f\u0440\u043e\u0445\u043e\u0434\u0438\u0442 \u0432\u0430\u043b\u0438\u0434\u0430\u0446\u0438\u044e \u0438 \u043e\u0442\u043f\u0440\u0430\u0432\u043b\u044f\u0435\u0442\u0441\u044f \u0432 \u0442\u043e\u043f\u0438\u043a <code>order-validations.v1<\/code>.<\/p>\n<pre><code class=\"kotlin\">   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) }                 )             )<\/code><\/pre>\n<p>\u041f\u043e\u043b\u043d\u0430\u044f \u0440\u0435\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u044f \u043a\u043b\u0430\u0441\u0441\u0430:<\/p>\n<pre><code class=\"kotlin\">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&lt;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&lt;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&lt;KeyValueStore&lt;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&lt;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&lt;*>, isKey: Boolean) {         val serdesConfig = hashMapOf&lt;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)     } }<\/code><\/pre>\n<h3>Order Enrichment Service<\/h3>\n<p>\u0421\u0435\u0440\u0432\u0438\u0441 \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u0435\u0442 Java \u043a\u043b\u0438\u0435\u043d\u0442 ksqlDB, \u0434\u043b\u044f \u0442\u043e\u0433\u043e, \u0447\u0442\u043e\u0431\u044b \u0441\u043e\u0437\u0434\u0430\u0442\u044c \u0432 ksqlDB stream, \u043a\u043e\u0442\u043e\u0440\u044b\u0439 \u0441\u043e\u0434\u0435\u0440\u0436\u0438\u0442 \u0438\u043d\u0444\u043e\u0440\u043c\u0430\u0446\u0438\u044e \u043a\u0430\u043a \u043e \u043f\u043e\u043a\u0443\u043f\u0430\u0442\u0435\u043b\u0435, \u0442\u0430\u043a \u0438 \u043e \u0437\u0430\u043a\u0430\u0437\u0435. \u0414\u0430\u043d\u043d\u044b\u0435 \u0438\u0437 \u044d\u0442\u043e\u0433\u043e \u0441\u0442\u0440\u0438\u043c\u0430 \u043c\u043e\u0436\u043d\u043e \u043f\u043e\u043b\u0443\u0447\u0438\u0442\u044c, \u043d\u0435\u043f\u043e\u0441\u0440\u0435\u0434\u0441\u0442\u0432\u0435\u043d\u043d\u043e \u043e\u0431\u0440\u0430\u0442\u0438\u0432\u0448\u0438\u0441\u044c \u043a ksqlDB \u043a\u0430\u043a \u0447\u0435\u0440\u0435\u0437 Java \u043a\u043b\u0438\u0435\u043d\u0442, \u0442\u0430\u043a \u0438 \u0434\u0440\u0443\u0433\u0438\u043c\u0438 \u0441\u043f\u043e\u0441\u043e\u0431\u0430\u043c\u0438, \u043d\u0430\u043f\u0440\u0438\u043c\u0435\u0440 \u0447\u0435\u0440\u0435\u0437 ksqlDB CLI.<\/p>\n<pre><code class=\"kotlin\">   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;        \"\"\"<\/code><\/pre>\n<p>\u041f\u043e\u043c\u0438\u043c\u043e \u044d\u0442\u043e\u0433\u043e \u0442\u0430\u043a\u0436\u0435 \u0441\u043e\u0437\u0434\u0430\u0435\u0442\u0441\u044f \u0442\u0430\u0431\u043b\u0438\u0446\u0430, \u0441\u043e\u0434\u0435\u0440\u0436\u0430\u0449\u0430\u044f \u0438\u043d\u0444\u043e\u0440\u043c\u0430\u0446\u0438\u044e \u043e \u043f\u043e\u0442\u0435\u043d\u0446\u0438\u0430\u043b\u044c\u043d\u043e \u043c\u043e\u0448\u0435\u043d\u043d\u0438\u0447\u0435\u0441\u043a\u0438\u0445 \u0434\u0435\u0439\u0441\u0442\u0432\u0438\u044f\u0445 \u043a\u043b\u0438\u0435\u043d\u0442\u043e\u0432:<\/p>\n<pre><code class=\"kotlin\">    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;      \"\"\"<\/code><\/pre>\n<p>\u0421\u0435\u0440\u0432\u0438\u0441 \u0432\u044b\u0441\u0442\u0430\u0432\u043b\u044f\u0435\u0442 \u043d\u0430\u0440\u0443\u0436\u0443 REST endpoint, \u043f\u043e \u043a\u043e\u0442\u043e\u0440\u043e\u043c\u0443 \u043c\u043e\u0436\u043d\u043e \u043f\u043e\u043b\u0443\u0447\u0438\u0442\u044c \u0434\u0430\u043d\u043d\u044b\u0435 \u043e \u043f\u043e\u0442\u0435\u043d\u0446\u0438\u0430\u043b\u044c\u043d\u044b\u0445 \u043c\u043e\u0448\u0435\u043d\u043d\u0438\u043a\u0430\u0445:<\/p>\n<pre><code class=\"kotlin\">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&lt;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&lt;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)    } }<\/code><\/pre>\n<h3>Validation Aggregator Service<\/h3>\n<p>\u041f\u0440\u043e\u0441\u0442\u043e\u0439 \u0441\u0435\u0440\u0432\u0438\u0441, \u043e\u0442\u0432\u0435\u0447\u0430\u044e\u0449\u0438\u0439 \u0437\u0430 \u0432\u044b\u0447\u0438\u0442\u044b\u0432\u0430\u043d\u0438\u0435 \u0434\u0430\u043d\u043d\u044b\u0445 \u0438\u0437 \u0442\u043e\u043f\u0438\u043a\u0430 <code>order-validations.v1<\/code>, \u043a\u0443\u0434\u0430 \u043e\u0442\u043f\u0440\u0430\u0432\u043b\u044f\u044e\u0442 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u044f \u0441\u0435\u0440\u0432\u0438\u0441\u044b \u043f\u0440\u043e\u0432\u0435\u0440\u043a\u0438 \u0437\u0430\u043a\u0430\u0437\u043e\u0432: Inventory Service, Fraud Service \u0438 Order Details Service. \u041f\u043e\u0441\u043b\u0435 \u0442\u043e\u0433\u043e, \u043a\u0430\u043a \u043e\u0442 \u043a\u0430\u0436\u0434\u043e\u0433\u043e \u0432\u0430\u043b\u0438\u0434\u0430\u0442\u043e\u0440\u0430 \u0431\u0443\u0434\u0435\u0442 \u043f\u043e\u043b\u0443\u0447\u0435\u043d \u043e\u0442\u0432\u0435\u0442 \u043f\u043e \u043a\u043e\u043d\u043a\u0440\u0435\u0442\u043d\u043e\u043c\u0443 \u0437\u0430\u043a\u0430\u0437\u0443, \u0432 \u0442\u043e\u043f\u0438\u043a <code>orders.v1<\/code> \u043e\u0442\u043f\u0440\u0430\u0432\u043b\u044f\u0435\u0442\u0441\u044f \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0435 \u0441 \u043e\u0431\u043d\u043e\u0432\u043b\u0435\u043d\u043d\u044b\u043c \u0441\u0442\u0430\u0442\u0443\u0441\u043e\u043c \u0437\u0430\u043a\u0430\u0437\u0430.<\/p>\n<p>\u041d\u0430\u0441\u0442\u0440\u043e\u0439\u043a\u0438 Kafka Streams \u0430\u043d\u0430\u043b\u043e\u0433\u0438\u0447\u043d\u044b Orders Service.<\/p>\n<p>\u0420\u0435\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u044f \u043a\u043b\u0430\u0441\u0441\u0430 ValidationAggregatorService:<\/p>\n<pre><code class=\"kotlin\">@Service class ValidationAggregatorService(     private val schemas: Schemas,     private val topicProps: TopicsProps ) {      private val serdes1: Consumed&lt;String, OrderValidation> = Consumed.with(         schemas.ORDER_VALIDATIONS.keySerde,         schemas.ORDER_VALIDATIONS.valueSerde     )      private val serdes2: Consumed&lt;String, Order> = Consumed         .with(schemas.ORDERS.keySerde, schemas.ORDERS.valueSerde)      private val serdes3: Grouped&lt;String, OrderValidation> = Grouped         .with(schemas.ORDER_VALIDATIONS.keySerde, schemas.ORDER_VALIDATIONS.valueSerde)      private val serdes4: StreamJoined&lt;String, Long, Order> = StreamJoined         .with(schemas.ORDERS.keySerde, Serdes.Long(), schemas.ORDERS.valueSerde)      private val serdes5: Grouped&lt;String, Order> = Grouped         .with(schemas.ORDERS.keySerde, schemas.ORDERS.valueSerde)      private val serdes6: StreamJoined&lt;String, OrderValidation, Order> = StreamJoined         .with(schemas.ORDERS.keySerde, schemas.ORDER_VALIDATIONS.valueSerde, schemas.ORDERS.valueSerde)      \/\/ disables auto registering schemas for specified Serde&lt;*>     private fun configureSerdes(serde: Serde&lt;*>, isKey: Boolean) {         val config = hashMapOf&lt;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. \u041a\u043e\u043b\u0438\u0447\u0435\u0441\u0442\u0432\u043e \u043f\u0440\u043e\u0432\u0435\u0440\u043e\u043a, \u043a\u043e\u0442\u043e\u0440\u044b\u0435 \u043f\u0440\u043e\u0445\u043e\u0434\u0438\u0442 \u043a\u0430\u0436\u0434\u044b\u0439 \u0437\u0430\u043a\u0430\u0437. \u0414\u043b\u044f \u043f\u0440\u043e\u0441\u0442\u043e\u0442\u044b \u0440\u0435\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u0438 \u0437\u0430\u0445\u0430\u0440\u0434\u043a\u043e\u0436\u0435\u043d\u043e.         val numberOfRules = 3         \/\/ 2. \u041d\u0430\u0441\u0442\u0440\u043e\u0439\u043a\u0430 Serdes, \u043a\u043e\u0442\u043e\u0440\u044b\u0435 \u0431\u0443\u0434\u0443\u0442 \u0443\u0447\u0430\u0441\u0442\u0432\u043e\u0432\u0430\u0442\u044c \u0432 \u043e\u0442\u043f\u0440\u0430\u0432\u043a\u0435 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0439 \u0432 \u0442\u043e\u043f\u0438\u043a orders.v1.         \/\/ \u0417\u0430\u043f\u0440\u0435\u0449\u0430\u0435\u043c \u0430\u0432\u0442\u043e\u043c\u0430\u0442\u0438\u0447\u0435\u0441\u043a\u0443\u044e \u0440\u0435\u0433\u0438\u0441\u0442\u0440\u0430\u0446\u0438\u044e \u0441\u0445\u0435\u043c \u0432 Schema Registry, \u0447\u0442\u043e\u0431\u044b \u0438\u0437\u0431\u0435\u0436\u0430\u0442\u044c \u043a\u043e\u043d\u0444\u043b\u0438\u043a\u0442\u043e\u0432.         val ordersKeySerde = Serdes.String()          val ordersValueSerde = SpecificAvroSerde&lt;Order>()         configureSerdes(ordersKeySerde, true)         configureSerdes(ordersValueSerde, false)         \/\/ 3. \u0441\u0442\u0440\u0438\u043c \u0438\u0437 \u0442\u043e\u043f\u0438\u043a\u0430 order-validations.v1         val validations: KStream&lt;String, OrderValidation> = builder             .stream(schemas.ORDER_VALIDATIONS.name, serdes1)         \/\/ 4. \u0441\u0442\u0440\u0438\u043c \u0438\u0437 \u0442\u043e\u043f\u0438\u043a\u0430 orders.v1, \u0432 \u043a\u043e\u0442\u043e\u0440\u043e\u043c \u043f\u0440\u0438\u0441\u0443\u0442\u0441\u0442\u0432\u0443\u044e\u0442 \u0442\u043e\u043b\u044c\u043a\u043e \u0437\u0430\u043a\u0430\u0437\u044b \u0432 \u0441\u0442\u0430\u0442\u0443\u0441\u0435 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. \u0433\u0440\u0443\u043f\u043f\u0438\u0440\u0443\u0435\u043c \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u044f \u043e \u0432\u0430\u043b\u0438\u0434\u0430\u0446\u0438\u0438 \u043f\u043e orderId             .groupByKey(serdes3)            \/\/ 6. \u0421\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u044f \u0442\u0430\u043a\u0436\u0435 \u0440\u0430\u0437\u0431\u0438\u0432\u0430\u044e\u0442\u0441\u044f \u043f\u043e \u043e\u043a\u043d\u0430\u043c \u0441\u0435\u0441\u0441\u0438\u043e\u043d\u043d\u043e\u0439 \u0430\u043a\u0442\u0438\u0432\u043d\u043e\u0441\u0442\u0438 \u0441 \u043f\u0435\u0440\u0438\u043e\u0434\u0438\u043c \u043d\u0435\u0430\u043a\u0442\u0438\u0432\u043d\u043e\u0441\u0442\u0438             \/\/ 5 \u043c\u0438\u043d\u0443\u0442             .windowedBy(SessionWindows.ofInactivityGapWithNoGrace(Duration.ofMinutes(5)))             \/\/ 7. \u0410\u0433\u0433\u0440\u0435\u0433\u0438\u0440\u0443\u0435\u043c \u0434\u0430\u043d\u043d\u044b\u0435 \u043e \u043f\u0440\u043e\u0432\u0435\u0440\u043a\u0435 \u0437\u0430\u043a\u0430\u0437\u043e\u0432, \u0435\u0441\u043b\u0438 \u0437\u0430\u043a\u0430\u0437 \u043f\u0440\u043e\u0448\u0435\u043b \u043f\u0440\u043e\u0432\u0435\u0440\u043a\u0443 (PASS),            \/\/ \u0442\u043e\u0433\u0434\u0430 \u0443\u0432\u0435\u043b\u0438\u0447\u0438\u0432\u0430\u0435\u043c \u0441\u0447\u0435\u0442\u0447\u0438\u043a \u043f\u0440\u043e\u0432\u0435\u0440\u0435\u043d\u043d\u044b\u0445 \u0437\u0430\u043a\u0430\u0437\u043e\u0432              .aggregate(                  { 0L },                 { id, result, total ->                     if (PASS == result.validationResult) total + 1 else total                 },                \/\/ \u0442\u0430\u043a \u043a\u0430\u043a \u043c\u044b \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u0435\u043c SessionWindow, \u043d\u0430\u043c \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e \u0434\u043e\u0431\u0430\u0432\u0438\u0442\u044c merger \u0434\u043b\u044f \u0441\u0435\u0441\u0441\u0438\u0439                \/\/ \u0438\u0437-\u0437\u0430 \u043e\u0441\u043e\u0431\u0435\u043d\u043d\u043e\u0441\u0442\u0435\u0439 \u0440\u0430\u0431\u043e\u0442\u044b \u044d\u0442\u0438\u0445 \u043e\u043a\u043e\u043d. \u041f\u043e \u0441\u0443\u0442\u0438, \u043a\u043e\u0433\u0434\u0430 \u0432 \u0441\u0442\u0440\u0438\u043c \u043f\u0440\u0438\u0445\u043e\u0434\u0438\u0442 \u043f\u0435\u0440\u0432\u043e\u0435 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0435                \/\/ \u0441\u0440\u0430\u0437\u0443 \u0441\u043e\u0437\u0434\u0430\u0435\u0442\u0441\u044f \u0441\u0435\u0441\u0441\u0438\u044f, \u043f\u043e\u0441\u043b\u0435 \u044d\u0442\u043e\u0433\u043e, \u0435\u0441\u043b\u0438 \u043f\u0440\u0438\u0445\u043e\u0434\u0438\u0442 \u0435\u0449\u0435 \u043e\u0431\u043d\u043e \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0435, \u043f\u043e\u043f\u0430\u0434\u0430\u044e\u0449\u0435\u0435 \u0432                \/\/ inactivity gap \u043f\u0435\u0440\u0432\u043e\u0433\u043e \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u044f, \u0442\u043e \u0434\u043b\u044f \u043d\u0435\u0433\u043e \u0442\u043e\u0436\u0435 \u0441\u043e\u0437\u0434\u0430\u0435\u0442\u0441\u044f \u0441\u0435\u0441\u0441\u0438\u044f, \u043a\u043e\u0442\u043e\u0440\u0430\u044f \u043c\u0435\u0440\u0436\u0438\u0442\u0441\u044f                \/\/ \u0441 \u0441\u0435\u0441\u0441\u0438\u0435\u0439 \u043f\u0435\u0440\u0432\u043e\u0433\u043e \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u044f. \u041f\u043e\u0434\u0440\u043e\u0431\u043d\u0435\u0435 \u0442\u0443\u0442:                \/\/ 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. \u0438\u0437\u0431\u0430\u0432\u043b\u044f\u0435\u043c\u0441\u044f \u043e\u0442 \"\u043e\u043a\u043e\u043d\" \u0438 \u0444\u043e\u0440\u043c\u0438\u0440\u0443\u0435\u043c \u043d\u043e\u0432\u044b\u0439 \u0441\u0442\u0440\u0438\u043c, \u0433\u0434\u0435: key = orderId, value = \u043a\u043e\u043b\u0438\u0447\u0435\u0441\u0442\u0432\u043e \u043f\u0440\u043e\u0439\u0434\u0435\u043d\u043d\u044b\u0445 \u043f\u0440\u043e\u0432\u0435\u0440\u043e\u043a             .toStream { windowedKey, total -> windowedKey.key() }              \/\/ \u041a\u043e\u0433\u0434\u0430 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u044f \u0443\u0434\u0430\u043b\u044f\u044e\u0442\u0441\u044f \u0438\u0437 Session Window, \u043f\u0443\u0431\u043b\u0438\u043a\u0443\u0435\u0442\u0441\u044f \u0441\u043e\u0431\u044b\u0442\u0438\u0435 delete (null-value), \u0438\u0437\u0431\u0430\u0432\u043b\u044f\u0435\u043c\u0441\u044f \u043e\u0442 \u043d\u0438\u0445             .filter { _, v -> v != null }              \/\/ 9. \u0422\u0430\u043a\u0436\u0435 \u0431\u0435\u0440\u0435\u043c \u0442\u043e\u043b\u044c\u043a\u043e \u0442\u0435 \u0437\u0430\u043f\u0438\u0441\u0438, \u043a\u043e\u0442\u043e\u0440\u044b\u0435 \u043f\u0440\u043e\u0448\u043b\u0438 \u0432\u0441\u0435 \u043f\u0440\u043e\u0432\u0435\u0440\u043a\u0438             .filter { _, total -> total >= numberOfRules } \/\/ 9.             \/\/ 10. \u043e\u0431\u044a\u0435\u0434\u0438\u043d\u044f\u0435\u043c \u0441\u043e \u0441\u0442\u0440\u0438\u043c\u043e\u043c orders \u0438 \u043e\u0431\u043d\u043e\u0432\u043b\u044f\u0435\u043c \u0441\u0442\u0430\u0442\u0443\u0441 \u0437\u0430\u043a\u0430\u0437\u0430 \u043d\u0430 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. \u041e\u0442\u043f\u0440\u0430\u0432\u043b\u044f\u0435\u043c \u0437\u0430\u043a\u0430\u0437 \u0441 \u043e\u0431\u043d\u043e\u0432\u043b\u0435\u043d\u043d\u044b\u043c \u0441\u0442\u0430\u0442\u0443\u0441\u043e\u043c \u0432 \u0442\u043e\u043f\u0438\u043a orders.v1             .to(schemas.ORDERS.name, Produced.with(ordersKeySerde, ordersValueSerde))           \/\/ 12. \u0442\u0430\u043a\u0436\u0435 \u043e\u0431\u0440\u0430\u0431\u0430\u0442\u044b\u0432\u0430\u0435\u043c \u0441\u043b\u0443\u0447\u0430\u0439, \u043a\u043e\u0433\u0434\u0430 \u0437\u0430\u043a\u0430\u0437 \u043f\u0440\u043e\u0432\u0430\u043b\u0438\u043b \u0445\u043e\u0442\u044f\u0431\u044b \u043e\u0434\u043d\u0443 \u043f\u0440\u043e\u0432\u0435\u0440\u043a\u0443         validations.filter { id, rule ->             FAIL == rule.validationResult          }            \/\/ 13. \u043e\u0431\u044a\u0435\u0434\u0438\u043d\u044f\u0435\u043c \u0441\u043e \u0441\u0442\u0440\u0438\u043c\u043e\u043c orders \u0438 \u043e\u0431\u043d\u043e\u0432\u043b\u044f\u0435\u043c \u0441\u0442\u0430\u0442\u0443\u0441 \u0437\u0430\u043a\u0430\u0437\u0430 \u043d\u0430 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. \u0413\u0440\u0443\u043f\u043f\u0438\u0440\u0443\u0435\u043c \u043f\u043e \u043a\u043b\u044e\u0447\u0443 - orderId             .groupByKey(serdes5)            \/\/ 15. \u0442\u0430\u043a \u043a\u0430\u043a \u0437\u0430\u043a\u0430\u0437 \u043c\u043e\u0433 \u043d\u0435 \u043f\u0440\u043e\u0439\u0442\u0438 \u043d\u0435\u0441\u043a\u043e\u043b\u044c\u043a\u043e \u043f\u0440\u043e\u0432\u0435\u0440\u043e\u043a, \u0432 \u043d\u0430\u0448\u0435\u043c \u0441\u0442\u0440\u0438\u043c\u0435 \u043c\u043e\u0433\u0443\u0442 \u043f\u0440\u0438\u0441\u0443\u0442\u0441\u0442\u0432\u043e\u0432\u0430\u0442\u044c             \/\/ \u043d\u0435\u0441\u043a\u043e\u043b\u044c\u043a\u043e \u0437\u0430\u043f\u0438\u0441\u0435\u0439 \u0441\u043e \u0441\u0442\u0430\u0442\u0443\u0441\u043e\u043c FAIL, \u0442\u0443\u0442 \u043c\u044b \u0438\u0437\u0431\u0430\u0432\u043b\u044f\u0435\u043c\u0441\u044f \u043e\u0442 \u0434\u0443\u0431\u043b\u0438\u043a\u0430\u0442\u043e\u0432, \u0442\u0430\u043a \u043a\u0430\u043a \u043d\u0430\u043c             \/\/ \u043d\u0435 \u0432\u0430\u0436\u043d\u043e \u0441\u043a\u043e\u043b\u044c\u043a\u043e \u043f\u0440\u043e\u0432\u0435\u0440\u043e\u043a \u043d\u0435 \u043f\u0440\u043e\u0448\u0435\u043b \u0437\u0430\u043a\u0430\u0437             .reduce { order, v1 -> order }             .toStream()             .peek { key, value ->                 log.info(\"Order {} failed validation. AFTER groupByKey.\", value)             }             .to( \/\/ 16. \u041e\u0442\u043f\u0440\u0430\u0432\u043b\u044f\u0435\u043c \u0437\u0430\u043a\u0430\u0437 \u0441 \u043e\u0431\u043d\u043e\u0432\u043b\u0435\u043d\u043d\u044b\u043c \u0441\u0442\u0430\u0442\u0443\u0441\u043e\u043c \u0432 \u0442\u043e\u043f\u0438\u043a orders.v1                 schemas.ORDERS.name,                 Produced.with(ordersKeySerde, ordersValueSerde)             )     } }<\/code><\/pre>\n<h3>\u0422\u0440\u0435\u0431\u043e\u0432\u0430\u043d\u0438\u044f \u0434\u043b\u044f \u0437\u0430\u043f\u0443\u0441\u043a\u0430:<\/h3>\n<ol>\n<li>\n<p>JDK 17<\/p>\n<\/li>\n<li>\n<p>Docker + docker-compose<\/p>\n<\/li>\n<li>\n<p>\u041e\u0417\u0423 \u0434\u043b\u044f Docker \u043d\u0435 \u043c\u0435\u043d\u0435\u0435 8 \u0413\u0431<\/p>\n<\/li>\n<li>\n<p>\u041e\u0417\u0423 \u0434\u043b\u044f \u043c\u0438\u043a\u0440\u043e\u0441\u0435\u0440\u0432\u0438\u0441\u043e\u0432 \u043d\u0435 \u043c\u0435\u043d\u0435\u0435 8 \u0413\u0431. (\u0422\u043e\u0447\u043d\u043e \u0437\u0430\u043f\u0443\u0441\u043a\u0430\u0435\u0442\u0441\u044f \u043d\u0430 Ubuntu 22.04 \u0441 64 \u0413\u0431 \u041e\u0417\u0423 \u0438 8 \u044f\u0434\u0440\u0430\u043c\u0438 \u043f\u0440\u043e\u0446\u0435\u0441\u0441\u043e\u0440\u0430)<\/p>\n<\/li>\n<\/ol>\n<h3>\u041a\u0430\u043a \u0437\u0430\u043f\u0443\u0441\u0442\u0438\u0442\u044c \u043c\u0438\u043a\u0440\u043e\u0441\u0435\u0440\u0432\u0438\u0441\u044b<\/h3>\n<p>\u0417\u0430\u043f\u0443\u0441\u0442\u0438\u0442\u044c \u0441\u043a\u0440\u0438\u043f\u0442 \u0438\u0437 \u043a\u043e\u0440\u043d\u0435\u0432\u043e\u0439 \u043f\u0430\u043f\u043a\u0438:<\/p>\n<ul>\n<li>\n<p>\u0415\u0441\u043b\u0438 \u0443\u0441\u0442\u0430\u043d\u043e\u0432\u043b\u0435\u043d Docker V2, \u0442\u043e:<\/p>\n<\/li>\n<\/ul>\n<pre><code class=\"bash\">.\/prepare_script_docker_new.sh<\/code><\/pre>\n<ul>\n<li>\n<p>\u0415\u0441\u043b\u0438 \u0431\u043e\u043b\u0435\u0435 \u0441\u0442\u0430\u0440\u0430 \u0432\u0435\u0440\u0441\u0438\u044f, \u0442\u043e:<\/p>\n<\/li>\n<\/ul>\n<pre><code class=\"bash\">.\/prepare_script_docker_old.sh<\/code><\/pre>\n<p>\u0421\u043a\u0440\u0438\u043f\u0442 \u043e\u0441\u0443\u0449\u0435\u0441\u0442\u0432\u043b\u044f\u0435\u0442 \u0441\u043b\u0435\u0434\u0443\u044e\u0449\u0438\u0435 \u0434\u0435\u0439\u0441\u0442\u0432\u0438\u044f:<\/p>\n<ol>\n<li>\n<p>\u041f\u043e\u0434\u043d\u0438\u043c\u0430\u0435\u0442 docker \u043a\u043e\u043d\u0442\u0435\u0439\u043d\u0435\u0440\u044b:<\/p>\n<ul>\n<li>\n<p>Zookeeper<\/p>\n<\/li>\n<li>\n<p>Kafka<\/p>\n<\/li>\n<li>\n<p>Schema Registry (Confluent)<\/p>\n<\/li>\n<li>\n<p>Connect (Kafka Connect)<\/p>\n<\/li>\n<li>\n<p>ksqlDB<\/p>\n<\/li>\n<li>\n<p>Elasticsearch<\/p>\n<\/li>\n<li>\n<p>SQLite<\/p>\n<\/li>\n<li>\n<p>Kibana<\/p>\n<\/li>\n<\/ul>\n<\/li>\n<li>\n<p>\u0420\u0435\u0433\u0438\u0441\u0442\u0440\u0438\u0440\u0443\u0435\u0442 Avro \u0441\u0445\u0435\u043c\u0443 \u0441\u0443\u0449\u043d\u043e\u0441\u0442\u0435\u0439, \u043a\u043e\u0442\u043e\u0440\u044b\u0435 \u0431\u0443\u0434\u0443\u0442 \u043e\u0431\u0440\u0430\u0431\u0430\u0442\u044b\u0432\u0430\u0442\u044c\u0441\u044f \u0432 Kafka. \u0420\u0435\u0433\u0438\u0441\u0442\u0440\u0430\u0446\u0438\u044f \u043e\u0441\u0443\u0449\u0435\u0441\u0442\u0432\u043b\u044f\u0435\u0442\u0441\u044f \u0441 \u043f\u043e\u043c\u043e\u0449\u044c\u044e Gradle Plugin:<br \/> <code>id 'com.github.imflog.kafka-schema-registry-gradle-plugin' version \"1.12.0\"<\/code><\/p>\n<\/li>\n<li>\n<p>\u041f\u0440\u043e\u0432\u043e\u0434\u0438\u0442 \u0447\u0438\u0441\u0442\u0443\u044e \u0441\u0431\u043e\u0440\u043a\u0443 \u043c\u0438\u043a\u0440\u043e\u0441\u0435\u0440\u0432\u0438\u0441\u043e\u0432.<\/p>\n<\/li>\n<li>\n<p>\u0421\u043e\u0437\u0434\u0430\u0435\u0442 \u0438\u043d\u0434\u0435\u043a\u0441 <code>orders.v1<\/code> \u0432 ElasticSearch<\/p>\n<\/li>\n<li>\n<p>\u041a\u043e\u043d\u0444\u0438\u0433\u0443\u0440\u0438\u0440\u0443\u0435\u0442 Kafka Connect \u0434\u043b\u044f:<\/p>\n<ul>\n<li>\n<p>\u043f\u043e\u043b\u0443\u0447\u0435\u043d\u0438\u044f \u0434\u0430\u043d\u043d\u044b\u0445 \u0438\u0437 SQLite (\u0442\u0430\u0431\u043b\u0438\u0446\u0430 customers) \u0438 \u0437\u0430\u043f\u0438\u0441\u0438 \u043b\u043e\u0433\u0430 \u0438\u0437\u043c\u0435\u043d\u0435\u043d\u0438\u0439 \u0432 Kafka (\u043f\u0430\u0442\u0442\u0435\u0440\u043d Change Data Capture)<\/p>\n<\/li>\n<li>\n<p>\u0437\u0430\u043f\u0438\u0441\u0438 \u0434\u0430\u043d\u043d\u044b\u0445 \u0438\u0437 \u0442\u043e\u043f\u0438\u043a\u0430 Kafka <code>orders.v1<\/code> \u0432 \u0438\u043d\u0434\u0435\u043a\u0441 <code>orders.v1<\/code> ElasticSearch<\/p>\n<\/li>\n<\/ul>\n<\/li>\n<li>\n<p>\u041d\u0430\u0441\u0442\u0440\u0430\u0438\u0432\u0430\u0435\u0442 Kibana \u0434\u043b\u044f \u043f\u0440\u043e\u0441\u043c\u043e\u0442\u0440\u0430 \u0437\u0430\u043f\u0438\u0441\u0435\u0439 \u0438\u0437 ElasticSearch.<\/p>\n<\/li>\n<\/ol>\n<h4>\u0412\u043e\u0437\u043c\u043e\u0436\u043d\u044b\u0435 \u043f\u0440\u043e\u0431\u043b\u0435\u043c\u044b<\/h4>\n<p>\u0423 \u043c\u0435\u043d\u044f \u043d\u0430 \u0432\u0438\u0440\u0442\u0443\u0430\u043b\u044c\u043d\u043e\u0439 \u043c\u0430\u0448\u0438\u043d\u0435 Ubuntu 22.04 \u0438\u0437\u043d\u0430\u0447\u0430\u043b\u044c\u043d\u043e \u043d\u0435 \u0437\u0430\u043f\u0443\u0441\u0442\u0438\u043b\u0441\u044f \u043a\u043e\u043d\u0442\u0435\u0439\u043d\u0435\u0440 elasticsearch, \u0442\u0430\u043a \u043a\u0430\u043a \u0442\u0440\u0435\u0431\u043e\u0432\u0430\u043b\u043e\u0441\u044c \u0443\u0441\u0442\u0430\u043d\u043e\u0432\u0438\u0442\u044c \u0437\u043d\u0430\u0447\u0435\u043d\u0438\u0435 vm.max_map_count \u043d\u0435 \u043d\u0438\u0436\u0435 262144. \u0415\u0441\u043b\u0438 \u0432\u043e\u0437\u043d\u0438\u043a\u0430\u0435\u0442 \u0442\u0430\u043a\u0430\u044f \u043f\u0440\u043e\u0431\u043b\u0435\u043c\u0430, \u0442\u043e \u043c\u043e\u0436\u043d\u043e \u0432\u044b\u043f\u043e\u043b\u043d\u0438\u0442\u044c \u0441\u043b\u0435\u0434\u0443\u044e\u0449\u0443\u044e \u043a\u043e\u043c\u0430\u043d\u0434\u0443:<\/p>\n<pre><code>sudo sysctl -w vm.max_map_count=262144<\/code><\/pre>\n<p>\u041f\u043e\u0441\u043b\u0435 \u0442\u043e\u0433\u043e, \u043a\u0430\u043a \u0441\u043a\u0440\u0438\u043f\u0442 \u043e\u0442\u0440\u0430\u0431\u043e\u0442\u0430\u0435\u0442, \u043c\u043e\u0436\u043d\u043e \u0437\u0430\u043f\u0443\u0441\u043a\u0430\u0442\u044c \u043c\u0438\u043a\u0440\u043e\u0441\u0435\u0440\u0432\u0438\u0441\u044b. \u0421\u0430\u043c\u044b\u0439 \u043f\u0440\u043e\u0441\u0442\u043e\u0439 \u0432\u0430\u0440\u0438\u0430\u043d\u0442 &#8212; \u0437\u0430\u043f\u0443\u0441\u0442\u0438\u0442\u044c \u0432 IntellijIDEA \u0432\u0441\u0435 \u043c\u0438\u043a\u0440\u043e\u0441\u0435\u0440\u0432\u0438\u0441\u044b, \u043a\u0440\u043e\u043c\u0435 populator. Populator \u0441\u043b\u0435\u0434\u0443\u0435\u0442 \u0437\u0430\u043f\u0443\u0441\u0442\u0438\u0442\u044c \u0432 \u043f\u043e\u0441\u043b\u0435\u0434\u043d\u044e\u044e \u043e\u0447\u0435\u0440\u0435\u0434\u044c, \u0442\u0430\u043a \u043a\u0430\u043a \u043e\u043d \u0431\u0443\u0434\u0435\u0442 \u043f\u043e\u0441\u0442\u043e\u044f\u043d\u043d\u043e \u0433\u0435\u043d\u0435\u0440\u0438\u0440\u043e\u0432\u0430\u0442\u044c \u0434\u0430\u043d\u043d\u044b\u0435 \u0438 \u043e\u0442\u043f\u0440\u0430\u0432\u043b\u044f\u0442\u044c POST \u0437\u0430\u043f\u0440\u043e\u0441\u044b \u0432 orders-service \u043d\u0430 \u0441\u043e\u0437\u0434\u0430\u043d\u0438\u0435 \u0437\u0430\u043a\u0430\u0437\u0430 (1 \u0437\u0430\u043a\u0430\u0437 \u0432 \u0441\u0435\u043a\u0443\u043d\u0434\u0443), \u0430 \u0442\u0430\u043a\u0436\u0435 \u043f\u043e\u0441\u044b\u043b\u0430\u0442\u044c \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u044f \u0432 Kafka \u0442\u043e\u043f\u0438\u043a\u0438: warehouse-inventory.v1 \u0438 payments.v1.<\/p>\n<p>\u041c\u043e\u0436\u043d\u043e \u0434\u0430\u0442\u044c \u0441\u0435\u0440\u0432\u0438\u0441\u0443 populator \u043f\u043e\u0440\u0430\u0431\u043e\u0442\u0430\u0442\u044c \u043d\u0435\u0441\u043a\u043e\u043b\u044c\u043a\u043e \u0441\u0435\u043a\u0443\u043d\u0434 (15-30) \u0438 \u043f\u043e\u0441\u043b\u0435 \u044d\u0442\u043e\u0433\u043e \u043e\u0441\u0442\u0430\u043d\u043e\u0432\u0438\u0442\u044c. \u0427\u0442\u043e\u0431\u044b \u043f\u0440\u043e\u0447\u0438\u0442\u0430\u0442\u044c \u0441\u043e\u0434\u0435\u0440\u0436\u0438\u043c\u043e\u0435 \u0442\u043e\u043f\u0438\u043a\u043e\u0432 Kafka (\u043f\u043e\u0441\u043b\u0435\u0434\u043d\u0438\u0435 5 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0439 \u0432 \u0442\u043e\u043f\u0438\u043a\u0435) \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e \u0432\u044b\u043f\u043e\u043b\u043d\u0438\u0442\u044c \u0441\u043b\u0435\u0434\u0443\u044e\u0449\u0438\u0439 \u0441\u043a\u0440\u0438\u043f\u0442 (\u043e\u043f\u044f\u0442\u044c \u0436\u0435 \u0432 \u0437\u0430\u0432\u0438\u0441\u0438\u043c\u043e\u0441\u0442\u0438 \u043e\u0442 \u0442\u043e\u0433\u043e, \u043a\u0430\u043a\u043e\u0439 docker \u0443\u0441\u0442\u0430\u043d\u043e\u0432\u043b\u0435\u043d):<\/p>\n<pre><code class=\"bash\">.\/read-topics-docker-new.sh<\/code><\/pre>\n<p>\u0438\u043b\u0438<\/p>\n<pre><code class=\"bash\">.\/read-dopics-docker-old.sh<\/code><\/pre>\n<p>\u0412 \u043a\u043e\u043d\u0446\u0435 \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e \u043e\u0441\u0442\u0430\u043d\u043e\u0432\u0438\u0442\u044c docker \u043a\u043e\u043d\u0442\u0435\u0439\u043d\u0435\u0440\u044b:<\/p>\n<pre><code>docker compose down<\/code><\/pre>\n<\/p>\n<\/div>\n<\/div>\n<\/div>\n<p><!----><!----><\/div>\n<p><!----><!----><br \/> \u0441\u0441\u044b\u043b\u043a\u0430 \u043d\u0430 \u043e\u0440\u0438\u0433\u0438\u043d\u0430\u043b \u0441\u0442\u0430\u0442\u044c\u0438 <a href=\"https:\/\/habr.com\/ru\/articles\/775900\/\"> https:\/\/habr.com\/ru\/articles\/775900\/<\/a><\/p>\n","protected":false},"excerpt":{"rendered":"<div><!--[--><!--]--><\/div>\n<div id=\"post-content-body\">\n<div>\n<div class=\"article-formatted-body article-formatted-body article-formatted-body_version-2\">\n<div xmlns=\"http:\/\/www.w3.org\/1999\/xhtml\">\n<h3>\u041f\u0440\u0435\u0434\u0438\u0441\u043b\u043e\u0432\u0438\u0435<\/h3>\n<p>\u0412\u0441\u0435\u043c \u043f\u0440\u0438\u0432\u0435\u0442! \u042d\u0442\u043e \u043c\u043e\u044f \u043f\u0435\u0440\u0432\u0430\u044f \u0441\u0442\u0430\u0442\u044c\u044f \u043d\u0430 Habr, \u0442\u0430\u043a \u0447\u0442\u043e \u043d\u0435 \u0441\u0443\u0434\u0438\u0442\u0435 \u0441\u0442\u0440\u043e\u0433\u043e \u0437\u0430 \u0441\u0442\u0438\u043b\u044c, \u0430 \u0432\u043e\u0442 \u043f\u043e \u0441\u043e\u0434\u0435\u0440\u0436\u0430\u043d\u0438\u044e \u0431\u0443\u0434\u0443 \u0440\u0430\u0434 \u043b\u044e\u0431\u043e\u0439 \u043a\u043e\u043d\u0441\u0442\u0440\u0443\u043a\u0442\u0438\u0432\u043d\u043e\u0439 \u043a\u0440\u0438\u0442\u0438\u043a\u0435. <\/p>\n<p>\u0421\u0442\u0430\u0442\u044c\u044f \u043e\u0441\u043d\u043e\u0432\u0430\u043d\u0430 \u043d\u0430 \u043f\u0443\u0431\u043b\u0438\u043a\u0430\u0446\u0438\u044f\u0445 Confluent \u043e \u0442\u043e\u043c, \u043a\u0430\u043a \u043c\u043e\u0436\u043d\u043e \u0440\u0435\u0430\u043b\u0438\u0437\u043e\u0432\u0430\u0442\u044c \u043c\u0438\u043a\u0440\u043e\u0441\u0435\u0440\u0432\u0438\u0441\u043d\u0443\u044e \u0430\u0440\u0445\u0438\u0442\u0435\u043a\u0442\u0443\u0440\u0443 \u043d\u0430 \u043e\u0441\u043d\u043e\u0432\u0435 \u0441\u043e\u0431\u044b\u0442\u0438\u0439 \u0441 \u043f\u043e\u043c\u043e\u0449\u044c\u044e Kafka Streams \u0438 KSQL. <\/p>\n<ol>\n<li>\n<p><a href=\"https:\/\/docs.confluent.io\/platform\/current\/tutorials\/examples\/microservices-orders\/docs\/index.html\" rel=\"noopener noreferrer nofollow\">Introduction to Streaming Application Development. <\/a><\/p>\n<\/li>\n<li>\n<p><a href=\"https:\/\/www.confluent.io\/blog\/building-a-microservices-ecosystem-with-kafka-streams-and-ksql\/\" rel=\"noopener noreferrer nofollow\">Building a Microservices Ecosystem with Kafka Streams and KSQL.<\/a><\/p>\n<\/li>\n<\/ol>\n<p>\u0418\u0441\u0445\u043e\u0434\u043d\u044b\u0439 \u043a\u043e\u0434 \u0441 \u043f\u0440\u0438\u043c\u0435\u0440\u0430\u043c\u0438 \u0432 \u0441\u0442\u0430\u0442\u044c\u044f\u0445 \u043d\u0430\u043f\u0438\u0441\u0430\u043d \u043d\u0430 Java \u0431\u0435\u0437 \u043f\u0440\u0438\u043c\u0435\u043d\u0435\u043d\u0438\u044f \u043a\u0430\u043a\u0438\u0445-\u043b\u0438\u0431\u043e \u0444\u0440\u0435\u0439\u043c\u0432\u043e\u0440\u043a\u043e\u0432, \u0430 \u0434\u043b\u044f \u0437\u0430\u043f\u0443\u0441\u043a\u0430 \u0432\u0441\u0435\u0445 \u043c\u0438\u043a\u0440\u043e\u0441\u0435\u0440\u0432\u0438\u0441\u043e\u0432 \u0442\u0440\u0435\u0431\u0443\u0435\u0442\u0441\u044f \u0430\u043a\u043a\u0430\u0443\u043d\u0442 Confluent Cloud. \u042f \u0440\u0435\u0448\u0438\u043b \u043f\u0435\u0440\u0435\u043f\u0438\u0441\u0430\u0442\u044c \u0432\u0441\u0435 \u0438\u0441\u0445\u043e\u0434\u043d\u0438\u043a\u0438 \u043d\u0430 Kotlin + Spring Boot for Apache Kafka. \u041c\u043e\u0439 \u043a\u043e\u0434 \u0434\u043e\u0441\u0442\u0443\u043f\u0435\u043d \u043d\u0430 <a href=\"https:\/\/github.com\/aasmc\/event-driven-microservices-kafka\" rel=\"noopener noreferrer nofollow\">Github<\/a>. <\/p>\n<h3>\u0412\u0435\u0440\u0445\u043d\u0435\u0443\u0440\u043e\u0432\u043d\u0435\u0432\u043e\u0435 \u043e\u043f\u0438\u0441\u0430\u043d\u0438\u0435 \u043f\u0440\u043e\u0435\u043a\u0442\u0430<\/h3>\n<figure class=\"full-width\"><\/figure>\n<p>\u0426\u0435\u043d\u0442\u0440\u043e\u043c \u0441\u0438\u0441\u0442\u0435\u043c\u044b \u043c\u0438\u043a\u0440\u043e\u0441\u0435\u0440\u0432\u0438\u0441\u043e\u0432 \u044f\u0432\u043b\u044f\u0435\u0442\u0441\u044f Orders Service, \u043a\u043e\u0442\u043e\u0440\u044b\u0439 \u043f\u0440\u0435\u0434\u043e\u0441\u0442\u0430\u0432\u043b\u044f\u0435\u0442 \u043a\u043b\u0438\u0435\u043d\u0442\u0430\u043c \u0432\u043e\u0437\u043c\u043e\u0436\u043d\u043e\u0441\u0442\u044c \u043f\u043e\u0441\u044b\u043b\u0430\u0442\u044c REST \u0437\u0430\u043f\u0440\u043e\u0441\u044b \u043d\u0430:<\/p>\n<ul>\n<li>\n<p>\u0441\u043e\u0434\u0430\u043d\u0438\u0435 \u0437\u0430\u043a\u0430\u0437\u043e\u0432 (POST <a href=\"http:\/\/localhost:8000\/v1\/orders\" rel=\"noopener noreferrer nofollow\">http:\/\/localhost:8000\/v1\/orders<\/a>)<\/p>\n<\/li>\n<li>\n<p>\u043f\u043e\u043b\u0443\u0447\u0435\u043d\u0438\u0435 \u0437\u0430\u043a\u0430\u0437\u0430 \u043f\u043e \u0435\u0433\u043e ID (GET <a href=\"http:\/\/localhost:8000\/v1\/orders\/%7Bid%7D\" rel=\"noopener noreferrer nofollow\">http:\/\/localhost:8000\/v1\/orders\/{id}<\/a>)<\/p>\n<\/li>\n<li>\n<p>\u043f\u043e\u043b\u0443\u0447\u0435\u043d\u0438\u0435 \u0432\u0430\u043b\u0438\u0434\u0438\u0440\u043e\u0432\u0430\u043d\u043d\u043e\u0433\u043e \u0437\u0430\u043a\u0430\u0437\u0430 \u043f\u043e \u0435\u0433\u043e ID (GET <a href=\"http:\/\/localhost:8000\/v1\/orders\/%7Bid%7D\/validated\" rel=\"noopener noreferrer nofollow\">http:\/\/localhost:8000\/v1\/orders\/{id}\/validated<\/a>)<\/p>\n<\/li>\n<\/ul>\n<p>\u041f\u0440\u0438 \u0441\u043e\u0437\u0434\u0430\u043d\u0438\u0438 \u0437\u0430\u043a\u0430\u0437\u0430 \u0432 Kafka \u0442\u043e\u043f\u0438\u043a <code>orders.v1<\/code> \u043f\u043e\u0441\u044b\u043b\u0430\u0435\u0442\u0441\u044f \u0441\u043e\u043e\u0442\u0432\u0435\u0442\u0441\u0442\u0432\u0443\u044e\u0449\u0435\u0435 \u0441\u043e\u0431\u044b\u0442\u0438\u0435, \u043e\u043d\u043e \u0432\u044b\u0447\u0438\u0442\u044b\u0432\u0430\u0435\u0442\u0441\u044f \u0440\u0430\u0437\u043b\u0438\u0447\u043d\u044b\u043c\u0438 \u0441\u0435\u0440\u0432\u0438\u0441\u0430\u043c\u0438, \u043e\u0442\u0432\u0435\u0447\u0430\u044e\u0449\u0438\u043c\u0438 \u0437\u0430 \u0432\u0430\u043b\u0438\u0434\u0430\u0446\u0438\u044e \u0437\u0430\u043a\u0430\u0437\u0430: Inventory Service, Fraud Service \u0438 Order Details Service. \u042d\u0442\u0438 \u0441\u0435\u0440\u0432\u0438\u0441\u044b \u043f\u0430\u0440\u0430\u043b\u043b\u0435\u043b\u044c\u043d\u043e \u0438 \u043d\u0435\u0437\u0430\u0432\u0438\u0441\u0438\u043c\u043e \u0434\u0440\u0443\u0433 \u043e\u0442 \u0434\u0440\u0443\u0433\u0430 \u043e\u0441\u0443\u0449\u0435\u0441\u0442\u0432\u043b\u044f\u044e\u0442 \u0432\u0430\u043b\u0438\u0434\u0430\u0446\u0438\u044e \u0437\u0430\u043a\u0430\u0437\u0430, \u0430 \u0440\u0435\u0437\u0443\u043b\u044c\u0442\u0430\u0442 PASS \u0438\u043b\u0438 FAIL \u0437\u0430\u043f\u0438\u0441\u044b\u0432\u0430\u044e\u0442 \u0432 \u0442\u043e\u043f\u0438\u043a Kafka <code>order-validations.v1<\/code>. Validation Aggregator Service \u0432\u044b\u0447\u0438\u0442\u044b\u0432\u0430\u0435\u0442 \u044d\u0442\u043e\u0442 \u0442\u043e\u043f\u0438\u043a, \u0430\u0433\u0433\u0440\u0435\u0433\u0438\u0440\u0443\u0435\u0442 \u0440\u0435\u0437\u0443\u043b\u044c\u0442\u0430\u0442\u044b \u043f\u043e \u043a\u0430\u0436\u0434\u043e\u043c\u0443 \u0437\u0430\u043a\u0430\u0437\u0443 \u0438 \u043e\u0442\u043f\u0440\u0430\u0432\u043b\u044f\u0435\u0442 \u043a\u043e\u043d\u0435\u0447\u043d\u044b\u0439 \u0440\u0435\u0437\u0443\u043b\u044c\u0442\u0430\u0442 \u0432 \u0442\u043e\u043f\u0438\u043a <code>orders.v1<\/code>.<\/p>\n<p>\u0414\u043b\u044f \u0440\u0435\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u0438 GET \u0437\u0430\u043f\u0440\u043e\u0441\u043e\u0432 Orders Service \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u0435\u0442 \u043c\u0430\u0442\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u043e\u0432\u0430\u043d\u043d\u043e\u0435 \u043f\u0440\u0435\u0434\u0441\u0442\u0430\u0432\u043b\u0435\u043d\u0438\u0435 (materialized view), \u043a\u043e\u0442\u043e\u0440\u043e\u0435 \u0432\u0441\u0442\u0440\u043e\u0435\u043d\u043e (embedded) \u0432 \u043a\u0430\u0436\u0434\u044b\u0439 \u0438\u043d\u0441\u0442\u0430\u043d\u0441 Orders Service. \u042d\u0442\u043e \u043c\u0430\u0442\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u043e\u0432\u0430\u043d\u043d\u043e\u0435 \u043f\u0440\u0435\u0434\u0441\u0442\u0430\u0432\u043b\u0435\u043d\u0438\u0435 \u0440\u0435\u0430\u043b\u0438\u0437\u043e\u0432\u0430\u043d\u043e \u043d\u0430 \u043e\u0441\u043d\u043e\u0432\u0435 Kafka Streams state store. \u0422\u0430\u043a \u043a\u0430\u043a state store \u043e\u0441\u043d\u043e\u0432\u0430\u043d \u043d\u0430 \u0442\u043e\u043f\u0438\u043a\u0430\u0445 Kafka, \u0438 \u0432\u0441\u0442\u0440\u043e\u0435\u043d \u0432 \u043a\u0430\u0436\u0434\u044b\u0439 \u0438\u043d\u0441\u0442\u0430\u043d\u0441, \u043e\u043d \u0445\u0440\u0430\u043d\u0438\u0442 \u0434\u0430\u043d\u043d\u044b\u0435 \u0442\u043e\u043b\u044c\u043a\u043e \u0442\u0435\u0445 \u043f\u0430\u0440\u0442\u0438\u0446\u0438\u0439 \u0442\u043e\u043f\u0438\u043a\u0430, \u043a\u043e\u0442\u043e\u0440\u044b\u0435 \u0432\u044b\u0447\u0438\u0442\u044b\u0432\u0430\u044e\u0442\u0441\u044f \u0434\u0430\u043d\u043d\u044b\u043c \u0438\u043d\u0441\u0442\u0430\u043d\u0441\u043e\u043c. \u041e\u0434\u043d\u0430\u043a\u043e Kafka Streams \u043f\u0440\u0435\u0434\u043e\u0441\u0442\u0430\u0432\u043b\u044f\u0435\u0442 API (Interactive Queries) \u0434\u043b\u044f \u0442\u043e\u0433\u043e, \u0447\u0442\u043e\u0431\u044b &#171;\u043d\u0430\u0439\u0442\u0438&#187; \u0438\u043d\u0441\u0442\u0430\u043d\u0441, \u043d\u0430 \u043a\u043e\u0442\u043e\u0440\u043e\u043c \u0445\u0440\u0430\u043d\u044f\u0442\u0441\u044f \u0434\u0430\u043d\u043d\u044b\u0435, \u0435\u0441\u043b\u0438 \u0438\u0445 \u043d\u0435\u0442 \u043b\u043e\u043a\u0430\u043b\u044c\u043d\u043e. \u042d\u0442\u043e \u043f\u043e\u0437\u0432\u043e\u043b\u044f\u0435\u0442 \u0440\u0435\u0430\u043b\u0438\u0437\u043e\u0432\u0430\u0442\u044c \u0433\u0430\u0440\u0430\u043d\u0442\u0438\u044e \u0434\u043b\u044f \u043a\u043b\u0438\u0435\u043d\u0442\u043e\u0432 read-your-own-writes (\u0447\u0442\u0435\u043d\u0438\u0435 \u0441\u0432\u043e\u0438\u0445 \u0437\u0430\u043f\u0438\u0441\u0435\u0439).<\/p>\n<p>\u0422\u0430\u043a\u0436\u0435 \u0432 \u0441\u0438\u0441\u0442\u0435\u043c\u0435 \u0435\u0441\u0442\u044c \u043f\u0440\u043e\u0441\u0442\u043e\u0439 \u0441\u0435\u0440\u0432\u0438\u0441 \u0434\u043b\u044f \u043e\u0442\u043f\u0440\u0430\u0432\u043a\u0438 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0439 \u043d\u0430 \u043f\u043e\u0447\u0442\u0443 Email Service (\u0432 \u0442\u0435\u043a\u0443\u0449\u0435\u0439 \u0440\u0435\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u0438 \u043f\u0440\u043e\u0441\u0442\u043e \u043e\u0441\u0443\u0449\u0435\u0441\u0442\u0432\u043b\u044f\u0435\u0442\u0441\u044f \u043b\u043e\u0433\u0438\u0440\u043e\u0432\u0430\u043d\u0438\u0435 \u043e\u0442\u043f\u0440\u0430\u0432\u043b\u044f\u0435\u043c\u044b\u0445 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0439).<\/p>\n<p>\u0421\u0435\u0440\u0432\u0438\u0441 Order Enrichment \u043e\u0442\u0432\u0435\u0447\u0430\u0435\u0442 \u0437\u0430 \u0441\u043e\u0437\u0434\u0430\u043d\u0438\u0435 <code>orders_enriched_stream<\/code> \u0432 <code>ksqlDB<\/code>, \u0438\u0437 \u043a\u043e\u0442\u043e\u0440\u043e\u0433\u043e \u043c\u043e\u0436\u043d\u043e \u043f\u043e\u043b\u0443\u0447\u0438\u0442\u044c \u0434\u0430\u043d\u043d\u044b\u0435 \u043e \u043f\u043e\u043a\u0443\u043f\u0430\u0442\u0435\u043b\u0435 \u0438 \u0435\u0433\u043e \u0437\u0430\u043a\u0430\u0437\u0435. \u0422\u0430\u043a\u0436\u0435 \u044d\u0442\u043e\u0442 \u0441\u0435\u0440\u0432\u0438\u0441 \u043f\u0440\u0435\u0434\u043e\u0441\u0442\u0430\u0432\u043b\u044f\u0435\u0442 REST \u0440\u0443\u0447\u043a\u0443 \u0434\u043b\u044f \u043f\u043e\u043b\u0443\u0447\u0435\u043d\u0438\u044f \u0438\u043d\u0444\u043e\u0440\u043c\u0430\u0446\u0438\u0438 \u043e \u0444\u043e\u0437\u043c\u043e\u0436\u043d\u044b\u0445 \u043c\u043e\u0448\u0435\u043d\u043d\u0438\u0447\u0435\u0441\u043a\u0438\u0445 \u0434\u0435\u0439\u0441\u0442\u0432\u0438\u044f\u0445 (\u0435\u0441\u043b\u0438 \u043f\u043e\u043a\u0443\u043f\u0430\u0442\u0435\u043b\u044c \u043e\u0441\u0443\u0449\u0435\u0441\u0442\u0432\u0438\u043b \u0431\u043e\u043b\u0435\u0435 2 \u0437\u0430\u043a\u0430\u0437\u043e\u0432 \u0437\u0430 30 \u0441\u0435\u043a\u0443\u043d\u0434):<\/p>\n<ul>\n<li>\n<p>GET <a href=\"http:\/\/localhost:8010\/fraud\" rel=\"noopener noreferrer nofollow\">http:\/\/localhost:8010\/fraud<\/a><\/p>\n<\/li>\n<\/ul>\n<p>\u0414\u0430\u043d\u043d\u044b\u0435 \u043e \u043f\u043e\u043a\u0443\u043f\u0430\u0442\u0435\u043b\u044f\u0445 \u0432\u044b\u0447\u0438\u0442\u044b\u0432\u0430\u044e\u0442\u0441\u044f \u0438\u0437 \u0411\u0414 SQLite \u0441 \u043f\u043e\u043c\u043e\u0449\u044c\u044e Kafka Connect <code>io.confluent.connect.jdbc.JdbcSourceConnector<\/code> \u0432 \u0442\u043e\u043f\u0438\u043a Kafka customers. \u041a\u043e\u043d\u0444\u0438\u0433\u0443\u0440\u0430\u0446\u0438\u044f \u043a\u043e\u043d\u043d\u0435\u043a\u0442\u043e\u0440\u0430: <a href=\"https:\/\/github.com\/aasmc\/event-driven-microservices-kafka\/blob\/master\/infra\/connectors\/connector_jdbc_customers_template.config\" rel=\"noopener noreferrer nofollow\">connector_jdbc_customers_template.config<\/a><\/p>\n<p>\u0414\u0430\u043d\u043d\u044b\u0435 \u043e \u0437\u0430\u043a\u0430\u0437\u0430\u0445 \u0437\u0430\u043f\u0438\u0441\u044b\u0432\u0430\u044e\u0442\u0441\u044f \u0438\u0437 \u0442\u043e\u043f\u0438\u043a\u0430 Kafka <code>orders.v1<\/code> \u0432 \u0438\u043d\u0434\u0435\u043a\u0441 <code>orders.v1<\/code> ElasticSearch \u0441 \u043f\u043e\u043c\u043e\u0449\u044c\u044e Kafka Connect <code>io.confluent.connect.elasticsearch.ElasticsearchSinkConnector<\/code>. \u041a\u043e\u043d\u0444\u0438\u0433\u0443\u0440\u0430\u0446\u0438\u044f \u043a\u043e\u043d\u043d\u0435\u043a\u0442\u043e\u0440\u0430: <a href=\"https:\/\/github.com\/aasmc\/event-driven-microservices-kafka\/blob\/master\/infra\/connectors\/connector_elasticsearch_template.config\" rel=\"noopener noreferrer nofollow\">connector_elasticsearch_template.config<\/a><\/p>\n<figure class=\"full-width\"><\/figure>\n<h3>\u0414\u0438\u0430\u0433\u0440\u0430\u043c\u043c\u0430 \u043c\u0438\u043a\u0440\u043e\u0441\u0435\u0440\u0432\u0438\u0441\u043e\u0432 \u0438 \u0442\u043e\u043f\u0438\u043a\u043e\u0432 Kafka:<\/h3>\n<figure class=\"full-width\"><\/figure>\n<h3>\u0414\u0435\u0442\u0430\u043b\u0438 \u0440\u0435\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u0438<\/h3>\n<h4>\u041c\u043e\u0434\u0443\u043b\u044c Common<\/h4>\n<p>\u0414\u0430\u043d\u043d\u044b\u0439 \u043c\u043e\u0434\u0443\u043b\u044c \u0441\u043e\u0431\u0440\u0430\u043d \u043a\u0430\u043a \u0431\u0438\u0431\u043b\u0438\u043e\u0442\u0435\u043a\u0430, \u043a\u043e\u0442\u043e\u0440\u0443\u044e \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u044e\u0442 \u043c\u0438\u043a\u0440\u043e\u0441\u0435\u0440\u0432\u0438\u0441\u044b. \u0421\u043e\u0434\u0435\u0440\u0436\u0438\u0442 \u0441\u0433\u0435\u043d\u0435\u0440\u0438\u0440\u043e\u0432\u0430\u043d\u043d\u044b\u0435 AVRO \u043a\u043b\u0430\u0441\u0441\u044b \u0430 \u0442\u0430\u043a\u0436\u0435 \u0443\u0442\u0438\u043b\u043b\u0438\u0442\u043d\u044b\u0435 \u043a\u043b\u0430\u0441\u0441\u044b, \u043e\u0431\u043b\u0435\u0433\u0447\u0430\u044e\u0449\u0438\u0435 \u0440\u0430\u0431\u043e\u0442\u0443 \u0441 \u0442\u043e\u043f\u0438\u043a\u0430\u043c\u0438, \u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u0430\u0442\u043e\u0440\u0430\u043c\u0438 \u0438 \u0434\u0435\u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u0430\u0442\u043e\u0440\u0430\u043c\u0438.<\/p>\n<h3>\u0421\u0443\u0449\u043d\u043e\u0441\u0442\u0438<\/h3>\n<p>\u0413\u0435\u043d\u0435\u0440\u0430\u0446\u0438\u044f \u0441\u0443\u0449\u043d\u043e\u0441\u0442\u0435\u0439, \u0441 \u043a\u043e\u0442\u043e\u0440\u044b\u043c\u0438 \u0440\u0430\u0431\u043e\u0442\u0430\u0435\u0442 Kafka \u043e\u0441\u0443\u0449\u0435\u0441\u0442\u0432\u043b\u044f\u0435\u0442\u0441\u044f <a href=\"https:\/\/github.com\/aasmc\/event-driven-microservices-kafka\/tree\/master\/common\/src\/main\/avro\" rel=\"noopener noreferrer nofollow\">\u043d\u0430 \u043e\u0441\u043d\u043e\u0432\u0435 \u0441\u0445\u0435\u043c Avro<\/a> \u0441 \u043f\u043e\u043c\u043e\u0449\u044c\u044e Gradle plugin <code>id 'com.github.davidmc24.gradle.plugin.avro' version \"1.9.1\"<\/code>, \u043d\u0430 \u044d\u0442\u0430\u043f\u0435 \u0441\u0431\u043e\u0440\u043a\u0438 \u0431\u0438\u0431\u043b\u0438\u043e\u0442\u0435\u043a\u0438 <code>common<\/code>. \u041d\u0438\u0436\u0435 \u043f\u0440\u0438\u0432\u0435\u0434\u0435\u043d\u043e \u043e\u043f\u0438\u0441\u0430\u043d\u0438\u0435 \u0441\u0443\u0449\u043d\u043e\u0441\u0442\u0435\u0439 \u0431\u0435\u0437 \u0434\u043e\u043f\u043e\u043b\u043d\u0438\u0442\u0435\u043b\u044c\u043d\u043e\u0433\u043e \u0441\u0433\u0435\u043d\u0435\u0440\u0438\u0440\u043e\u0432\u0430\u043d\u043d\u043e\u0433\u043e \u043a\u043e\u0434\u0430.<\/p>\n<pre><code class=\"kotlin\">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 )<\/code><\/pre>\n<p>\u0420\u0435\u0433\u0438\u0441\u0442\u0440\u0430\u0446\u0438\u044f \u0441\u0445\u0435\u043c Avro \u0432 Confluent Schema Registry \u043e\u0441\u0443\u0449\u0435\u0441\u0442\u0432\u043b\u044f\u0435\u0442\u0441\u044f \u0441 \u043f\u043e\u043c\u043e\u0449\u044c\u044e Gradle Plugin: <code>id 'com.github.imflog.kafka-schema-registry-gradle-plugin' version \"1.12.0\" <\/code> \u043f\u043e\u0441\u043b\u0435 \u0442\u043e\u0433\u043e, \u043a\u0430\u043a \u043f\u043e\u0434\u043d\u044f\u0442 docker \u043a\u043e\u043d\u0442\u0435\u0439\u043d\u0435\u0440 schema-registry. \u0427\u0442\u043e\u0431\u044b \u0437\u0430\u0440\u0435\u0433\u0438\u0441\u0442\u0440\u0438\u0440\u043e\u0432\u0430\u0442\u044c \u0441\u0445\u0435\u043c\u044b \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e \u043d\u0430\u0441\u0442\u0440\u043e\u0438\u0442\u044c \u043f\u043b\u0430\u0433\u0438\u043d. <\/p>\n<p>\u041c\u0438\u043d\u0438\u043c\u0430\u043b\u044c\u043d\u0430\u044f \u043d\u0430\u0441\u0442\u0440\u043e\u0439\u043a\u0430 \u0432 Gradle:<\/p>\n<pre><code>schemaRegistry {     url = 'http:\/\/localhost:8081'     quiet = true     register {         subject('customers-value', 'common\/src\/main\/avro\/customer.avsc', 'AVRO')         subject('orders.v1-value', 'common\/src\/main\/avro\/order.avsc', 'AVRO')         subject('orders-enriched.v1-value', 'common\/src\/main\/avro\/orderenriched.avsc', 'AVRO')         subject('order-validations.v1-value', 'common\/src\/main\/avro\/ordervalidation.avsc', 'AVRO')         subject('payments.v1-value', 'common\/src\/main\/avro\/payment.avsc', 'AVRO')     } }<\/code><\/pre>\n<p>\u041f\u0440\u0438 \u0440\u0435\u0433\u0438\u0441\u0442\u0440\u0430\u0446\u0438\u0438 \u0443\u043a\u0430\u0437\u044b\u0432\u0430\u044e\u0442\u0441\u044f:<\/p>\n<ol>\n<li>\n<p>\u041d\u0430\u0437\u0432\u0430\u043d\u0438\u0435 \u0441\u0445\u0435\u043c\u044b. \u041e\u0434\u043d\u0438\u043c \u0438\u0437 \u043f\u0430\u0442\u0442\u0435\u0440\u043d\u043e\u0432 \u043d\u0430\u0438\u043c\u0435\u043d\u043e\u0432\u0430\u043d\u0438\u044f \u044f\u0432\u043b\u044f\u0435\u0442\u0441\u044f TopicName-value \/ TopicName-key<\/p>\n<\/li>\n<li>\n<p>\u041f\u0443\u0442\u044c \u0434\u043e \u0444\u0430\u0439\u043b\u0430 \u0441\u043e \u0441\u0445\u0435\u043c\u043e\u0439<\/p>\n<\/li>\n<li>\n<p>\u0424\u043e\u0440\u043c\u0430\u0442 \u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u0438, \u0432 \u043d\u0430\u0448\u0435\u043c \u0441\u043b\u0443\u0447\u0430\u0435 &#8212; AVRO.<\/p>\n<\/li>\n<\/ol>\n<p>\u0414\u0430\u043b\u0435\u0435 \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e \u0437\u0430\u043f\u0443\u0441\u0442\u0438\u0442\u044c \u0442\u0430\u0441\u043a\u0443 Gradle \u0438\u0437 \u043a\u043e\u0440\u043d\u0435\u0432\u043e\u0439 \u0434\u0438\u0440\u0435\u043a\u0442\u043e\u0440\u0438\u0438 \u043f\u0440\u043e\u0435\u043a\u0442\u0430 (\u043d\u0435 \u0438\u0437 \u0434\u0438\u0440\u0435\u043a\u0442\u043e\u0440\u0438\u0438 common):<\/p>\n<pre><code class=\"bash\">.\/gradlew registerSchemaTask<\/code><\/pre>\n<p>\u0414\u043b\u044f \u0440\u0430\u0431\u043e\u0442\u044b \u0441 \u0437\u0430\u043f\u0438\u0441\u044f\u043c\u0438 \u0432 \u0442\u043e\u043f\u0438\u043a\u0430\u0445 \u043f\u0440\u0438\u043b\u043e\u0436\u0435\u043d\u0438\u0435 Kafka Streams \u0434\u043e\u043b\u0436\u043d\u043e \u0437\u043d\u0430\u0442\u044c \u043e \u0444\u043e\u0440\u043c\u0430\u0442\u0435 \u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u0438 \u0434\u0430\u043d\u043d\u044b\u0445. \u0414\u043b\u044f \u044d\u0442\u043e\u0433\u043e \u043d\u0430\u0441\u0442\u0440\u0430\u0438\u0432\u0430\u044e\u0442\u0441\u044f \u0441\u043f\u0435\u0446\u0438\u0430\u043b\u044c\u043d\u044b\u0435 <code>org.apache.kafka.common.serialization.Serde<\/code> \u043a\u043b\u0430\u0441\u0441\u044b, \u043a\u043e\u0442\u043e\u0440\u044b\u0435 \u0437\u043d\u0430\u044e\u0442 \u043a\u0430\u043a \u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u043e\u0432\u0430\u0442\u044c\/\u0434\u0435\u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u043e\u0432\u0430\u0442\u044c \u043a\u043b\u044e\u0447\/\u0437\u043d\u0430\u0447\u0435\u043d\u0438\u0435. \u0414\u043b\u044f \u0443\u0434\u043e\u0431\u0441\u0442\u0432\u0430 \u0432\u0441\u0435 \u043d\u0430\u0441\u0442\u0440\u043e\u0439\u043a\u0438 \u043f\u043e\u043c\u0435\u0449\u0435\u043d\u044b \u0432 \u043e\u0442\u0434\u0435\u043b\u044c\u043d\u044b\u0439 \u043a\u043b\u0430\u0441\u0441 <a href=\"https:\/\/github.com\/aasmc\/event-driven-microservices-kafka\/blob\/master\/common\/src\/main\/kotlin\/ru\/aasmc\/eventdriven\/common\/schemas\/Schemas.kt\" rel=\"noopener noreferrer nofollow\">Schemas<\/a>. \u0414\u043b\u044f \u0441\u0433\u0435\u043d\u0435\u0440\u0438\u0440\u043e\u0432\u0430\u043d\u043d\u044b\u0445 AVRO \u043a\u043b\u0430\u0441\u0441\u043e\u0432 \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u0435\u0442\u0441\u044f <code>io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde<\/code>, \u043f\u0440\u0438 \u043d\u0430\u0441\u0442\u0440\u043e\u0439\u043a\u0435 \u043a\u043e\u0442\u043e\u0440\u043e\u0433\u043e \u0443\u043a\u0430\u0437\u044b\u0432\u0430\u0435\u0442\u0441\u044f \u0430\u0434\u0440\u0435\u0441 Schema Registry &#8212; \u0441\u0435\u0440\u0432\u0435\u0440\u0430, \u0445\u0440\u0430\u043d\u044f\u0449\u0435\u0433\u043e \u0432\u0441\u0435, \u0432 \u043d\u0430\u0448\u0435\u043c \u0441\u043b\u0443\u0447\u0430\u0435 AVRO, \u0441\u0445\u0435\u043c\u044b.<\/p>\n<pre><code class=\"kotlin\">fun configureSerdes() {     val config = hashMapOf&lt;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&lt;*>, config: Map&lt;String, Any>, isKey: Boolean) {   if (serde is SpecificAvroSerde) {     serde.configure(config, isKey)   } }<\/code><\/pre>\n<h3>Orders Service<\/h3>\n<p>\u0412 Spring Boot \u043d\u0435\u0442 \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e\u0441\u0442\u0438 \u0432\u0440\u0443\u0447\u043d\u0443\u044e \u0443\u043f\u0440\u0430\u0432\u043b\u044f\u0442\u044c \u0436\u0438\u0437\u043d\u0435\u043d\u043d\u044b\u043c \u0446\u0438\u043a\u043b\u043e\u043c \u0441\u0443\u0449\u043d\u043e\u0441\u0442\u0438 KafkaStreams. \u0414\u043b\u044f \u044d\u0442\u043e\u0433\u043e \u0435\u0441\u0442\u044c \u0430\u0431\u0441\u0442\u0440\u0430\u043a\u0446\u0438\u044f <code>org.springframework.kafka.config.StreamsBuilderFactoryBean<\/code>, \u043a\u043e\u0442\u043e\u0440\u0430\u044f \u0441\u043e\u0437\u0434\u0430\u0435\u0442 \u0438 \u0443\u043f\u0440\u0430\u0432\u043b\u044f\u0435\u0442 <code>org.apache.kafka.streams.KafkaStreams<\/code> \u0438 <code>org.apache.kafka.streams.StreamsBuilder<\/code>. \u0422\u0435\u043c \u043d\u0435 \u043c\u0435\u043d\u0435\u0435, \u043d\u0430\u043c \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e \u0441\u043a\u043e\u043d\u0444\u0438\u0433\u0443\u0440\u0438\u0440\u043e\u0432\u0430\u0442\u044c \u0431\u0438\u043d <code>org.springframework.kafka.config.KafkaStreamsConfiguration<\/code>:<\/p>\n<pre><code class=\"kotlin\">@EnableKafkaStreams @Configuration class KafkaConfig(     private val kafkaProps: KafkaProps,     private val serviceUtils: ServiceUtils,     private val topicProps: TopicsProps ) {      @Bean     fun outstandingRequests(): MutableMap&lt;String, FilteredResponse&lt;String, Order, OrderDto>> {         return ConcurrentHashMap()     }      @Bean(name = [KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME])     fun kStreamsConfig(): KafkaStreamsConfiguration {         val props = hashMapOf&lt;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,             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)     } }<\/code><\/pre>\n<p>\u0414\u043b\u044f \u0442\u043e\u0433\u043e, \u0447\u0442\u043e\u0431\u044b \u0440\u0430\u0431\u043e\u0442\u0430\u043b\u0438 Interactive Queries (\u0435\u0441\u043b\u0438 \u043c\u044b \u0445\u043e\u0442\u0438\u043c \u0437\u0430\u043f\u0440\u0430\u0448\u0438\u0432\u0430\u0442\u044c \u0434\u0430\u043d\u043d\u044b\u0435 \u043d\u0435 \u0442\u043e\u043b\u044c\u043a\u043e<br \/> \u0438\u0437 \u043b\u043e\u043a\u0430\u043b\u044c\u043d\u043e\u0433\u043e state store, \u043d\u043e \u0438 \u0441 \u0434\u0440\u0443\u0433\u0438\u0445 \u0438\u043d\u0441\u0442\u0430\u043d\u0441\u043e\u0432), \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e \u0432 \u043a\u043e\u043d\u0444\u0438\u0433\u0443\u0440\u0430\u0446\u0438\u044f\u0445 \u0443\u043a\u0430\u0437\u0430\u0442\u044c<br \/> <code>StreamsConfig.APPLICATION_SERVER_CONFIG<\/code> &#8212; \u0430\u0434\u0440\u0435\u0441 \u0442\u0435\u043a\u0443\u0449\u0435\u0433\u043e \u0438\u043d\u0441\u0442\u0430\u043d\u0441\u0430 \u0432 \u0444\u043e\u0440\u043c\u0430\u0442\u0435 <code>host:port<\/code>. \u0422\u0430\u043a\u0436\u0435 \u043c\u044b \u0443\u043a\u0430\u0437\u044b\u0432\u0430\u0435\u043c \u0430\u0434\u0440\u0435\u0441 Schema Registry, \u043f\u0443\u0442\u044c \u0434\u043e \u043a\u0430\u0442\u0430\u043b\u043e\u0433\u0430, \u0433\u0434\u0435 \u0431\u0443\u0434\u0435\u0442 \u0445\u0440\u0430\u043d\u0438\u0442\u0441\u044f state store \u0438 \u0442.\u0434. \u041f\u043e \u0443\u043c\u043e\u043b\u0447\u0430\u043d\u0438\u044e Kafka Streams \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u0435\u0442 \u0432\u0441\u0442\u0440\u043e\u0435\u043d\u043d\u043e\u0435 key-value \u0445\u0440\u0430\u043d\u0438\u043b\u0438\u0449\u0435 RocksDB \u0432 \u043a\u0430\u0447\u0435\u0441\u0442\u0432\u0435 state store.<\/p>\n<p>\u041f\u0440\u0438 \u0441\u0442\u0430\u0440\u0442\u0435 \u043f\u0440\u0438\u043b\u043e\u0436\u0435\u043d\u0438\u044f Orders Service \u043d\u0430 \u043e\u0441\u043d\u043e\u0432\u0435 \u0442\u043e\u043f\u0438\u043a\u0430 Kafka <code>orders.v1<\/code> \u0441\u043e\u0437\u0434\u0430\u0435\u0442\u0441\u044f KTable &#8212; \u0430\u0431\u0441\u0442\u0440\u0430\u043a\u0446\u0438\u044f Kafka Streams, \u043a\u043e\u0442\u043e\u0440\u0430\u044f \u043f\u0440\u0435\u0434\u0441\u0442\u0430\u0432\u043b\u044f\u0435\u0442 \u0438\u0437 \u0441\u0435\u0431\u044f snapshot \u0441\u043e\u0441\u0442\u043e\u044f\u043d\u0438\u044f (\u043c\u043e\u0436\u043d\u043e \u043f\u0440\u043e\u0432\u0435\u0441\u0442\u0438 \u0430\u043d\u0430\u043b\u043e\u0433\u0438\u044e \u0441 compacted topic) &#8212; \u0445\u0440\u0430\u043d\u044f\u0442\u0441\u044f \u043f\u043e\u0441\u043b\u0435\u0434\u043d\u0438\u0435 \u0437\u0430\u043f\u0438\u0441\u0438 \u043f\u043e \u043a\u043b\u044e\u0447\u0443 \u0432 \u0442\u043e\u043f\u0438\u043a\u0435. \u041f\u0440\u0438 \u043f\u043e\u0441\u0442\u0443\u043f\u043b\u0435\u043d\u0438\u0438 \u043d\u043e\u0432\u044b\u0445 \u0437\u0430\u043f\u0438\u0441\u0435\u0439 \u0434\u0430\u043d\u043d\u044b\u0435 \u0432 KTable \u043e\u0431\u043d\u043e\u0432\u043b\u044f\u044e\u0442\u0441\u044f, \u0444\u0430\u043a\u0442\u0438\u0447\u0435\u0441\u043a\u0438 \u043f\u0440\u043e\u0438\u0437\u0432\u043e\u0434\u0438\u0442\u0441\u044f \u043e\u043f\u0435\u0440\u0430\u0446\u0438\u044f <code>UPSERT<\/code> &#8212; \u0435\u0441\u043b\u0438 \u043a\u043b\u044e\u0447 \u0435\u0441\u0442\u044c \u0432 \u0442\u0430\u0431\u043b\u0438\u0446\u0435, \u0434\u0430\u043d\u043d\u044b\u0435 \u043e\u0431\u043d\u043e\u0432\u043b\u044f\u044e\u0442\u0441\u044f, \u0435\u0441\u043b\u0438 \u043d\u0435\u0442 &#8212; \u0441\u043e\u0437\u0434\u0430\u044e\u0442\u0441\u044f, \u0435\u0441\u043b\u0438 \u043f\u0440\u0438\u0445\u043e\u0434\u0438\u0442 \u043a\u043b\u044e\u0447 &#8212; <code>null<\/code> &#8212; \u0442\u043e \u044d\u0442\u043e \u043e\u0437\u043d\u0430\u0447\u0430\u0435\u0442 \u0443\u0434\u0430\u043b\u0435\u043d\u0438\u0435 \u0434\u0430\u043d\u043d\u044b\u0445, \u0431\u043e\u043b\u0435\u0435 \u043e\u043d\u0438 \u043d\u0435 \u0431\u0443\u0434\u0443\u0442 \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u044c\u0441\u044f \u043f\u0440\u0438 \u0432\u044b\u0431\u043e\u0440\u043a\u0435 \u043f\u043e \u043a\u043b\u044e\u0447\u0443 \u0438\u043b\u0438 \u0441\u043e\u0435\u0434\u0438\u043d\u0435\u043d\u0438\u0438 \u0442\u0430\u0431\u043b\u0438\u0446\u044b \u0441<\/p>\n<\/div>\n<\/div>\n<\/div>\n<\/div>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[],"tags":[],"class_list":["post-360983","post","type-post","status-publish","format-standard","hentry"],"_links":{"self":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/posts\/360983","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=360983"}],"version-history":[{"count":0,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/posts\/360983\/revisions"}],"wp:attachment":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=360983"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=360983"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=360983"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}