|
37 | 37 | - key.serializer |
38 | 38 | - value.serializer |
39 | 39 | - max.in.flight.requests.per.connection |
40 | | - - enable.idempotencecd |
| 40 | + - enable.idempotence |
41 | 41 |
|
42 | 42 | Please see [producer configs](http://kafka.apache.org/documentation.html#producerconfigs) |
43 | 43 | for a complete list of all producer configs available in Kafka." |
44 | | - |
45 | | - (:require [ziggurat.config :refer [ziggurat-config]] |
46 | | - [clojure.tools.logging :as log] |
| 44 | + (:refer-clojure :exclude [send]) |
| 45 | + (:require [clojure.tools.logging :as log] |
47 | 46 | [mount.core :refer [defstate]] |
48 | | - [camel-snake-kebab.core :as csk] |
| 47 | + [ziggurat.config :refer [build-producer-config-properties ziggurat-config]] |
49 | 48 | [ziggurat.tracer :refer [tracer]] |
50 | | - [ziggurat.util.java-util :refer [get-key]] |
51 | | - [schema.core :as s]) |
52 | | - (:import (org.apache.kafka.clients.producer KafkaProducer ProducerRecord ProducerConfig) |
53 | | - (java.util Properties) |
54 | | - (io.opentracing.contrib.kafka TracingKafkaProducer)) |
| 49 | + [ziggurat.util.java-util :refer [get-key]]) |
| 50 | + (:import (io.opentracing.contrib.kafka TracingKafkaProducer) |
| 51 | + (org.apache.kafka.clients.producer KafkaProducer ProducerRecord)) |
55 | 52 | (:gen-class |
56 | | - :name tech.gojek.ziggurat.internal.Producer |
57 | 53 | :methods [^{:static true} [send [String String Object Object] java.util.concurrent.Future] |
58 | | - ^{:static true} [send [String String int Object Object] java.util.concurrent.Future]])) |
59 | | - |
60 | | -(defn *implements-serializer?* [serializer-class] |
61 | | - (contains? (set (.getInterfaces (Class/forName serializer-class))) |
62 | | - (Class/forName "org.apache.kafka.common.serialization.Serializer"))) |
63 | | - |
64 | | -(def implements-serializer? (s/pred *implements-serializer?* 'implements-serializer?)) |
65 | | - |
66 | | -(s/defschema ProducerConfigSchema {(s/required-key :bootstrap-servers) s/Any |
67 | | - (s/optional-key :key-serializer-class) implements-serializer? |
68 | | - (s/optional-key :value-serializer-class) implements-serializer? |
69 | | - (s/optional-key :key-serializer) implements-serializer? |
70 | | - (s/optional-key :value-serializer) implements-serializer? |
71 | | - (s/optional-key :retries-config) s/Any |
72 | | - (s/optional-key :metadata-max-age) s/Any |
73 | | - (s/optional-key :reconnect-backoff-ms) s/Any |
74 | | - (s/optional-key :client-id) s/Any |
75 | | - (s/optional-key :metric-num-samples) s/Any |
76 | | - (s/optional-key :transaction-timeout) s/Any |
77 | | - (s/optional-key :retries) s/Any |
78 | | - (s/optional-key :retry-backoff-ms) s/Any |
79 | | - (s/optional-key :receive-buffer) s/Any |
80 | | - (s/optional-key :partitioner-class) s/Any |
81 | | - (s/optional-key :max-block-ms) s/Any |
82 | | - (s/optional-key :metrics-reporter-classes) s/Any |
83 | | - (s/optional-key :compression-type) s/Any |
84 | | - (s/optional-key :max-request-size) s/Any |
85 | | - (s/optional-key :delivery-timeout-ms) s/Any |
86 | | - (s/optional-key :metrics-sample-window-ms) s/Any |
87 | | - (s/optional-key :request-timeout-ms) s/Any |
88 | | - (s/optional-key :buffer-memory) s/Any |
89 | | - (s/optional-key :interceptor-classes) s/Any |
90 | | - (s/optional-key :linger-ms) s/Any |
91 | | - (s/optional-key :connections-max-idle-ms) s/Any |
92 | | - (s/optional-key :acks) s/Any |
93 | | - (s/optional-key :enable-idempotence) s/Any |
94 | | - (s/optional-key :metrics-recording-level) s/Any |
95 | | - (s/optional-key :transactional-id) s/Any |
96 | | - (s/optional-key :reconnect-backoff-max-ms) s/Any |
97 | | - (s/optional-key :client-dns-lookup) s/Any |
98 | | - (s/optional-key :max-in-flight-requests-per-connection) s/Any |
99 | | - (s/optional-key :send-buffer) s/Any |
100 | | - (s/optional-key :batch-size) s/Any}) |
101 | | - |
102 | | -(def valid-configs? (partial s/validate ProducerConfigSchema)) |
103 | | - |
104 | | -(def explain-str (partial s/explain ProducerConfigSchema)) |
105 | | - |
106 | | -(defn property->fn [field-name] |
107 | | - (let [raw-field-name (condp = field-name |
108 | | - :max-in-flight-requests-per-connection "org.apache.kafka.clients.producer.ProducerConfig/%s" |
109 | | - :retries-config "org.apache.kafka.clients.producer.ProducerConfig/RETRIES_CONFIG" |
110 | | - :key-serializer "org.apache.kafka.clients.producer.ProducerConfig/KEY_SERIALIZER_CLASS_CONFIG" |
111 | | - :value-serializer "org.apache.kafka.clients.producer.ProducerConfig/VALUE_SERIALIZER_CLASS_CONFIG" |
112 | | - "org.apache.kafka.clients.producer.ProducerConfig/%s_CONFIG")] |
113 | | - (->> field-name |
114 | | - csk/->SCREAMING_SNAKE_CASE_STRING |
115 | | - (format raw-field-name) |
116 | | - (read-string)))) |
117 | | - |
118 | | -(defn producer-properties [config-map] |
119 | | - (if (valid-configs? config-map) |
120 | | - (let [props (Properties.)] |
121 | | - (doseq [config-key (keys config-map)] |
122 | | - (.setProperty props |
123 | | - (eval (property->fn config-key)) |
124 | | - (str (get config-map config-key)))) |
125 | | - props) |
126 | | - (throw (ex-info (explain-str config-map) config-map)))) |
| 54 | + ^{:static true} [send [String String int Object Object] java.util.concurrent.Future]] |
| 55 | + :name tech.gojek.ziggurat.internal.Producer)) |
127 | 56 |
|
128 | 57 | (defn producer-properties-map [] |
129 | 58 | (reduce (fn [producer-map [stream-config-key stream-config]] |
130 | 59 | (let [producer-config (:producer stream-config)] |
131 | 60 | (if (some? producer-config) |
132 | | - (assoc producer-map stream-config-key (producer-properties producer-config)) |
| 61 | + (assoc producer-map stream-config-key (build-producer-config-properties producer-config)) |
133 | 62 | producer-map))) |
134 | 63 | {} |
135 | 64 | (seq (:stream-router (ziggurat-config))))) |
136 | 65 |
|
| 66 | +(declare kafka-producers) |
| 67 | + |
137 | 68 | (defstate kafka-producers |
138 | 69 | :start (if (not-empty (producer-properties-map)) |
139 | 70 | (do (log/info "Starting Kafka producers ...") |
140 | 71 | (reduce (fn [producers [stream-config-key properties]] |
141 | | - (do (log/debug "Constructing Kafka producer associated with [" stream-config-key "] ") |
142 | | - (let [_ (println properties) |
143 | | - kp (KafkaProducer. properties) |
144 | | - _ (println kp) |
145 | | - tkp (TracingKafkaProducer. kp tracer)] |
146 | | - (assoc producers stream-config-key tkp)))) |
| 72 | + (log/debug "Constructing Kafka producer associated with [" stream-config-key "] ") |
| 73 | + (let [kp (KafkaProducer. properties) |
| 74 | + tkp (TracingKafkaProducer. kp tracer)] |
| 75 | + (assoc producers stream-config-key tkp))) |
147 | 76 | {} |
148 | 77 | (seq (producer-properties-map)))) |
149 | 78 | (log/info "No producers found. Can not initiate start.")) |
|
0 commit comments