From 9f36d28ce29031f89bdbbbdd37c72245a85e39ea Mon Sep 17 00:00:00 2001 From: didier amyot Date: Mon, 2 Jul 2018 14:25:30 -0400 Subject: [PATCH] started to play with kafka. hit a solid wall --- docker-compose.yml | 152 ++++++++++++++++++++++ project.clj | 1 + resources/config.edn | 2 +- src/skirmish/core.clj | 4 +- src/skirmish/kafka/book.clj | 1 + src/skirmish/kafka/exchange_simulator.clj | 22 ++++ src/skirmish/kafka_exchange.clj | 41 ++++++ src/skirmish/rest/core.clj | 8 +- src/skirmish/rest/exchange.clj | 12 +- src/skirmish/rest/marketdata.clj | 14 +- 10 files changed, 237 insertions(+), 20 deletions(-) create mode 100644 docker-compose.yml create mode 100644 src/skirmish/kafka/book.clj create mode 100644 src/skirmish/kafka/exchange_simulator.clj create mode 100644 src/skirmish/kafka_exchange.clj diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..4985c43 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,152 @@ +# A theft from https://github.com/simplesteph/kafka-stack-docker-compose/blob/master/full-stack.yml +version: '2' + +services: + zoo1: + image: zookeeper:3.4.9 + restart: unless-stopped + hostname: zoo1 + ports: + - "2181:2181" + environment: + ZOO_MY_ID: 1 + ZOO_PORT: 2181 + ZOO_SERVERS: server.1=zoo1:2888:3888 + volumes: + - ./full-stack/zoo1/data:/data + - ./full-stack/zoo1/datalog:/datalog + + + kafka1: + image: confluentinc/cp-kafka:4.1.0 + hostname: kafka1 + ports: + - "9092:9092" + environment: + # add the entry "127.0.0.1 kafka1" to your /etc/hosts file + KAFKA_ADVERTISED_LISTENERS: "PLAINTEXT://kafka1:9092" + KAFKA_ZOOKEEPER_CONNECT: "zoo1:2181" + KAFKA_BROKER_ID: 1 + KAFKA_LOG4J_LOGGERS: "kafka.controller=INFO,kafka.producer.async.DefaultEventHandler=INFO,state.change.logger=INFO" + KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 + volumes: + - ./full-stack/kafka1/data:/var/lib/kafka/data + depends_on: + - zoo1 + + kafka-schema-registry: + image: confluentinc/cp-schema-registry:4.1.0 + hostname: kafka-schema-registry + restart: unless-stopped + ports: + - "8081:8081" + environment: + SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: zoo1:2181 + SCHEMA_REGISTRY_HOST_NAME: kafka-schema-registry + SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081 + depends_on: + - zoo1 + - kafka1 + + schema-registry-ui: + image: landoop/schema-registry-ui:0.9.4 + hostname: kafka-schema-registry-ui + ports: + - "8001:8000" + environment: + SCHEMAREGISTRY_URL: http://kafka-schema-registry:8081/ + PROXY: "true" + depends_on: + - kafka-schema-registry + + kafka-rest-proxy: + image: confluentinc/cp-kafka-rest:4.1.0 + hostname: kafka-rest-proxy + ports: + - "8082:8082" + environment: + KAFKA_REST_ZOOKEEPER_CONNECT: zoo1:2181 + KAFKA_REST_LISTENERS: http://0.0.0.0:8082/ + KAFKA_REST_SCHEMA_REGISTRY_URL: http://kafka-schema-registry:8081/ + KAFKA_REST_HOST_NAME: kafka-rest-proxy + KAFKA_REST_BOOTSTRAP_SERVERS: kafka1:9092 + depends_on: + - zoo1 + - kafka1 + - kafka-schema-registry + + kafka-topics-ui: + image: landoop/kafka-topics-ui:0.9.3 + hostname: kafka-topics-ui + ports: + - "8000:8000" + environment: + KAFKA_REST_PROXY_URL: "http://kafka-rest-proxy:8082/" + PROXY: "true" + depends_on: + - zoo1 + - kafka1 + - kafka-schema-registry + - kafka-rest-proxy + + kafka-connect: + image: confluentinc/cp-kafka-connect:4.1.0 + hostname: kafka-connect + ports: + - "8083:8083" + environment: + CONNECT_BOOTSTRAP_SERVERS: "kafka1:9092" + CONNECT_REST_PORT: 8083 + CONNECT_GROUP_ID: compose-connect-group + CONNECT_CONFIG_STORAGE_TOPIC: docker-connect-configs + CONNECT_OFFSET_STORAGE_TOPIC: docker-connect-offsets + CONNECT_STATUS_STORAGE_TOPIC: docker-connect-status + CONNECT_KEY_CONVERTER: io.confluent.connect.avro.AvroConverter + CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL: 'http://kafka-schema-registry:8081' + CONNECT_VALUE_CONVERTER: io.confluent.connect.avro.AvroConverter + CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: 'http://kafka-schema-registry:8081' + CONNECT_INTERNAL_KEY_CONVERTER: "org.apache.kafka.connect.json.JsonConverter" + CONNECT_INTERNAL_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter" + CONNECT_REST_ADVERTISED_HOST_NAME: "kafka-connect" + CONNECT_LOG4J_ROOT_LOGLEVEL: "INFO" + CONNECT_LOG4J_LOGGERS: "org.apache.kafka.connect.runtime.rest=WARN,org.reflections=ERROR" + CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: "1" + CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: "1" + CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: "1" + depends_on: + - zoo1 + - kafka1 + - kafka-schema-registry + - kafka-rest-proxy + + kafka-connect-ui: + image: landoop/kafka-connect-ui:0.9.4 + hostname: kafka-connect-ui + ports: + - "8003:8000" + environment: + CONNECT_URL: "http://kafka-connect:8083/" + PROXY: "true" + depends_on: + - kafka-connect + + zoonavigator-web: + image: elkozmon/zoonavigator-web:0.4.0 + ports: + - "8004:8000" + environment: + API_HOST: "zoonavigator-api" + API_PORT: 9000 + links: + - zoonavigator-api + depends_on: + - zoonavigator-api + restart: unless-stopped + + zoonavigator-api: + image: elkozmon/zoonavigator-api:0.4.0 + environment: + SERVER_HTTP_PORT: 9000 + restart: unless-stopped + depends_on: + - zoo1 diff --git a/project.clj b/project.clj index 3f09a9d..7a6090a 100644 --- a/project.clj +++ b/project.clj @@ -18,6 +18,7 @@ ;;utils [slingshot "0.12.2"] [spootnik/kinsky "0.1.22"] + [org.clojars.fossi/kafka-streams-clojure "0.1.0-SNAPSHOT"] ;; JEAH! [cheshire "5.8.0"]] :java-source-paths ["src/java"] :main ^:skip-aot skirmish.core diff --git a/resources/config.edn b/resources/config.edn index cf200e0..e465f16 100644 --- a/resources/config.edn +++ b/resources/config.edn @@ -1,6 +1,6 @@ { :kafka - {:bootstrap.servers ["localhost:9092"] + {:bootstrap.servers "localhost:9092" :topics-prefix "skirmish-01-"} diff --git a/src/skirmish/core.clj b/src/skirmish/core.clj index 6322dae..e1be0fa 100644 --- a/src/skirmish/core.clj +++ b/src/skirmish/core.clj @@ -2,11 +2,13 @@ (:require [skirmish.cli.core] [skirmish.rest.core] [skirmish.cli.benchmark] + [skirmish.kafka.exchange-simulator] [aero.core :refer (read-config)]) (:gen-class)) (def apps {:webserver skirmish.rest.core/-main - :benchmark skirmish.cli.benchmark/-main}) + :benchmark skirmish.cli.benchmark/-main + :kafka-exchange-simulator skirmish.kafka.exchange-simulator/-main}) (defn -main [& argv] (let [config (-> "config.edn" clojure.java.io/resource read-config) diff --git a/src/skirmish/kafka/book.clj b/src/skirmish/kafka/book.clj new file mode 100644 index 0000000..bc62856 --- /dev/null +++ b/src/skirmish/kafka/book.clj @@ -0,0 +1 @@ +(ns skirmish.kafka.book) diff --git a/src/skirmish/kafka/exchange_simulator.clj b/src/skirmish/kafka/exchange_simulator.clj new file mode 100644 index 0000000..8632d26 --- /dev/null +++ b/src/skirmish/kafka/exchange_simulator.clj @@ -0,0 +1,22 @@ +(ns skirmish.kafka.exchange-simulator + (:require [kinsky.client :as client])) + +(def mock-kafka-messages + [[:sell-1 {:action :received :order {:type :sell :quantity 10 :price 10 :book.order/id :sell-1}}] + [:sell-1 {:action :inserted-in-book :order {:type :sell :quantity 10 :price 10 :book.order/id :sell-1}}] + [:sell-2 {:action :received :order {:type :sell :quantity 5 :price 9 :book.order/id :sell-2}}] + [:sell-2 {:action :inserted-in-book :order {:type :sell :quantity 5 :price 9 :book.order/id :sell-2}}] + [:buy-1 {:action :received :order {:type :buy :quantity 7 :price 9 :book.order/id :buy-1}}] + [:buy-1 {:action :executed-before :order {:type :buy :quantity 7 :price 9 :book.order/id :buy-1}}]]) + + + + +(defn -main [config] + (let [p (client/producer {:bootstrap.servers (-> config :kafka :bootstrap.servers)} + (client/keyword-serializer) + (client/edn-serializer))] + (doseq [[key val] mock-kafka-messages] + @(client/send! p "account" key val) + (println "<<>>" key "--" val))) + (print "finished")) \ No newline at end of file diff --git a/src/skirmish/kafka_exchange.clj b/src/skirmish/kafka_exchange.clj new file mode 100644 index 0000000..f3bac69 --- /dev/null +++ b/src/skirmish/kafka_exchange.clj @@ -0,0 +1,41 @@ +(ns skirmish.kafka-exchange + (:require [kafka-streams-clojure.api :as api]) + (:import [org.apache.kafka.clients.producer KafkaProducer ProducerRecord] + [org.apache.kafka.streams StreamsConfig KafkaStreams] + [org.apache.kafka.streams.kstream KStreamBuilder])) + + +(defn -main [] + (def xform (comp (filter (fn [[k v]] (string? v))) + (map (fn [[k v]] [v k])) + (filter (fn [[k v]] (= "foo" v))))) + + (def builder (KStreamBuilder.)) + + (def kstream (-> builder + (.stream (into-array String ["tset"])) + (api/transduce-kstream xform) + (.to "test"))) + + (def kafka-streams + (KafkaStreams. builder + (StreamsConfig. {StreamsConfig/APPLICATION_ID_CONFIG "test-app-id" + StreamsConfig/BOOTSTRAP_SERVERS_CONFIG "localhost:9092" + StreamsConfig/KEY_SERDE_CLASS_CONFIG org.apache.kafka.common.serialization.Serdes$StringSerde + StreamsConfig/VALUE_SERDE_CLASS_CONFIG org.apache.kafka.common.serialization.Serdes$StringSerde}))) + (.start kafka-streams) + + (def producer (KafkaProducer. {"bootstrap.servers" "localhost:9092" + "acks" "all" + "retries" "0" + "key.serializer" "org.apache.kafka.common.serialization.StringSerializer" + "value.serializer" "org.apache.kafka.common.serialization.StringSerializer"})) + + @(.send producer (ProducerRecord. "tset" "foo" "bar")) + ;; Observe message come across topic "test" via kafka-console-consumer + + @(.send producer (ProducerRecord. "tset" "baz" "quux")) + ;; Observe message does not come across topic "test" via kafka-console-consumer + + (.close producer) + (.close kafka-streams)) \ No newline at end of file diff --git a/src/skirmish/rest/core.clj b/src/skirmish/rest/core.clj index 125985a..8df51a7 100644 --- a/src/skirmish/rest/core.clj +++ b/src/skirmish/rest/core.clj @@ -1,11 +1,17 @@ (ns skirmish.rest.core (:require [org.httpkit.server :refer [run-server]] [compojure.handler :refer [api]] + [kinsky.client] [skirmish.rest.exchange :refer [create-exchanges-api-for create-routes]])) (defn -main [config] (let [webserver-config (:webserver config) exchange-apis (create-exchanges-api-for (:securities config)) - api-routes (create-routes exchange-apis)] + kafka-producer (kinsky.client/producer {:bootstrap.servers (get-in config [:kafka :bootstrap.servers])} + (kinsky.client/keyword-serializer) + (kinsky.client/edn-serializer)) + api-routes (create-routes exchange-apis kafka-producer)] + (println "Running server with bellow config") + (println webserver-config) (run-server (api api-routes) webserver-config))) \ No newline at end of file diff --git a/src/skirmish/rest/exchange.clj b/src/skirmish/rest/exchange.clj index 7b1d243..3379345 100644 --- a/src/skirmish/rest/exchange.clj +++ b/src/skirmish/rest/exchange.clj @@ -8,6 +8,7 @@ [skirmish.exchange :refer [create-exchange-for-security]] [skirmish.rest.util :refer [build-response build-response-body]] [clojure.spec.alpha :as s] + [kinsky.client] [slingshot.slingshot :refer [try+ throw+]] [clojure.core.async :refer [chan go go-loop close! !! sub]] [clojure.edn :as edn])) @@ -97,14 +98,17 @@ (throw+ {:error-type :4XX :status 404 :body {:error "Security not available."}}))) -(defn create-order [exchange-apis] +(defn create-order [exchange-apis kafka-producer] (fn [{{sec-id-str :sec-id} :route-params body :body :as request}] (try+ (let [asked-order (get-body body) exchange-api (get-exchange-api (keyword sec-id-str) exchange-apis) incoming-order-channel (:incoming-order-channel exchange-api) order (validate-order asked-order (:sec-id exchange-api))] - (>!! incoming-order-channel order) + @(kinsky.client/send! kafka-producer + (str "new-orders-" sec-id-str) + (:book.order/id order) + order) (build-response request order)) (catch [:error-type :4XX] e (build-response request e)) (catch [:error-type :5XX] e (build-response request e))))) @@ -129,13 +133,13 @@ (let [order-channel (get-new-trades-channel exchange-apis (keyword sec-id-str))] (chan-to-websocket request order-channel ws-chan))))) -(defn create-routes [exchange-apis] +(defn create-routes [exchange-apis kafka-producer] (routes (context "/api/1" [] (GET "/" [] api-description) (GET "/securities" [] (securities exchange-apis)) (GET "/securities/:sec-id" [] (get-security exchange-apis)) - (POST "/securities/:sec-id" [] (create-order exchange-apis)) + (POST "/securities/:sec-id" [] (create-order exchange-apis kafka-producer)) (GET "/securities/:sec-id/trades" [] (listen-to-trades exchange-apis)) (GET "/securities/:sec-id/orders" [] (listen-to-orders exchange-apis))) (not-found "not found"))) diff --git a/src/skirmish/rest/marketdata.clj b/src/skirmish/rest/marketdata.clj index 48de8d5..cb203ea 100644 --- a/src/skirmish/rest/marketdata.clj +++ b/src/skirmish/rest/marketdata.clj @@ -1,13 +1 @@ -(ns skirmish.rest.marketdata - (:require [compojure.core :refer :all])) - - -(defn create-routes [exchange-apis] - (routes - (context "/api/1" [] - (GET "/securities" [] ) - (GET "/securities/:sec-id" [] (get-security exchange-apis)) - (POST "/securities/:sec-id" [] (create-order exchange-apis)) - (GET "/securities/:sec-id/trades" [] (listen-to-trades exchange-apis)) - (GET "/securities/:sec-id/orders" [] (listen-to-orders exchange-apis))) - (not-found "not found"))) +(ns skirmish.rest.marketdata)