Skip to content

Commit 5749006

Browse files
yblankeyblanke
committed
Enhance error handling with custom catch function in polling mechanism
1 parent 0068246 commit 5749006

1 file changed

Lines changed: 109 additions & 91 deletions

File tree

‎src/ketu/async/source.clj‎

Lines changed: 109 additions & 91 deletions
Original file line numberDiff line numberDiff line change
@@ -25,26 +25,26 @@
2525
"Returns opts with final :ketu.apache.consumer/config entry"
2626
[opts]
2727
(let [original-config (:ketu.apache.consumer/config opts)
28-
config (util/set-ketu-to-apache-opts original-config opts)]
28+
config (util/set-ketu-to-apache-opts original-config opts)]
2929
(assoc opts :ketu.apache.consumer/config config)))
3030

3131
(defn default-consumer-supplier [opts]
32-
(let [key-type (:ketu.source/key-type opts)
33-
key-deserializer (when (instance? Deserializer key-type) key-type)
34-
value-type (:ketu.source/value-type opts)
32+
(let [key-type (:ketu.source/key-type opts)
33+
key-deserializer (when (instance? Deserializer key-type) key-type)
34+
value-type (:ketu.source/value-type opts)
3535
value-deserializer (when (instance? Deserializer value-type) value-type)]
3636
(consumer/consumer (:ketu.apache.consumer/config opts) key-deserializer value-deserializer)))
3737

3838
(defn default-opts []
39-
{:ketu.source/consumer-supplier default-consumer-supplier
40-
:ketu.source/key-type :byte-array
41-
:ketu.source/value-type :byte-array
42-
:ketu.source/poll-timeout-ms 100
43-
:ketu.source/done-putting-timeout-ms 60000
44-
:ketu.source/consumer-close-timeout-ms 60000
39+
{:ketu.source/consumer-supplier default-consumer-supplier
40+
:ketu.source/key-type :byte-array
41+
:ketu.source/value-type :byte-array
42+
:ketu.source/poll-timeout-ms 100
43+
:ketu.source/done-putting-timeout-ms 60000
44+
:ketu.source/consumer-close-timeout-ms 60000
4545
:ketu.source/consumer-thread-timeout-ms 60000
46-
:ketu.source/close-out-chan? true
47-
:ketu.source/close-consumer? true})
46+
:ketu.source/close-out-chan? true
47+
:ketu.source/close-consumer? true})
4848

4949
(defn- finalize-opts [opts]
5050
(-> (default-opts)
@@ -54,13 +54,13 @@
5454
(defn- subscribe-fn
5555
"Returns a function that takes a consumer and subscribes to either a topic list or a pattern according to opts."
5656
[opts]
57-
(let [subscribe (cond
58-
(:ketu.source/topic opts) consumer/subscribe-to-topic!
59-
(:ketu.source/topic-list opts) consumer/subscribe-to-list!
60-
(:ketu.source/topic-pattern opts) consumer/subscribe-to-pattern!)
61-
topic (or (:ketu.source/topic opts)
62-
(:ketu.source/topic-list opts)
63-
(:ketu.source/topic-pattern opts))
57+
(let [subscribe (cond
58+
(:ketu.source/topic opts) consumer/subscribe-to-topic!
59+
(:ketu.source/topic-list opts) consumer/subscribe-to-list!
60+
(:ketu.source/topic-pattern opts) consumer/subscribe-to-pattern!)
61+
topic (or (:ketu.source/topic opts)
62+
(:ketu.source/topic-list opts)
63+
(:ketu.source/topic-pattern opts))
6464
create-listener (:ketu.source/create-rebalance-listener-obj opts)]
6565
(if create-listener
6666
(fn [consumer] (subscribe consumer topic (create-listener {:ketu.source/consumer consumer})))
@@ -70,7 +70,7 @@
7070
"Returns a function that takes a consumer and assigns specific partitions according to opts."
7171
[opts]
7272
(when-let [assign-tps (:ketu.source/assign-single-topic-partitions opts)]
73-
(let [topic (:ketu.source.assign/topic assign-tps)
73+
(let [topic (:ketu.source.assign/topic assign-tps)
7474
partitions (:ketu.source.assign/partition-nums assign-tps)]
7575
(fn [consumer]
7676
(consumer/assign! consumer (consumer/topic-partitions topic partitions))))))
@@ -96,7 +96,7 @@
9696
([^Consumer consumer source-name opts records-to-skip]
9797
(try
9898
(let [assigned-partitions (consumer/assignment consumer)
99-
records-to-skip (or records-to-skip (get-max-poll-records-from-opts opts))]
99+
records-to-skip (or records-to-skip (get-max-poll-records-from-opts opts))]
100100
(doseq [^TopicPartition partition assigned-partitions]
101101
(try
102102
(let [current-position (consumer/position consumer partition)
@@ -111,16 +111,34 @@
111111
(log/error logger "[source={}] Failed to get assigned partitions for offset increment"
112112
source-name e)))))
113113

114-
(defn default-poll-error-handler [consumer opts]
114+
(defn- default-poll-error-handler [consumer opts]
115115
(let [records-to-skip (:ketu.source/error-skip-offset-amount opts)
116-
source-name (:ketu/name opts)]
116+
source-name (:ketu/name opts)]
117117
(increment-offsets-for-assigned-partitions! consumer source-name opts records-to-skip)
118118
[]))
119119

120+
(defn- get-custom-error-handler [opts]
121+
(let [provided-catch-fn (:ketu.source/custom-catch-fn opts)
122+
custom-catch-fn
123+
(cond
124+
(nil? provided-catch-fn)
125+
nil
126+
127+
(fn? provided-catch-fn)
128+
provided-catch-fn
129+
130+
:else
131+
(do
132+
(log/error logger "[source={}] Invalid :ketu.source/custom-catch-fn (must be fn [consumer opts] -> coll), got: %s. Using default error handler."
133+
(type provided-catch-fn))
134+
nil))]
135+
custom-catch-fn))
136+
120137
(defn- poll-fn [^Consumer consumer should-poll? opts]
121138
(when @should-poll?
122139
(let [source-name (:ketu/name opts)
123-
error-handler (:ketu.source/poll-error-handler opts (default-poll-error-handler consumer opts))
140+
custom-error-handler (get-custom-error-handler opts)
141+
error-handler (or custom-error-handler (default-poll-error-handler consumer opts))
124142
poll-timeout-duration (Duration/ofMillis (:ketu.source/poll-timeout-ms opts))]
125143
(fn []
126144
(try
@@ -139,80 +157,80 @@
139157

140158
(defn- source-existing-consumer
141159
[^Consumer consumer out-chan opts]
142-
(let [source-name (:ketu/name opts)
143-
^String thread-name (str "ketu-source-" source-name)
144-
close-out-chan? (:ketu.source/close-out-chan? opts)
145-
^long close-consumer? (:ketu.source/close-consumer? opts)
160+
(let [source-name (:ketu/name opts)
161+
^String thread-name (str "ketu-source-" source-name)
162+
close-out-chan? (:ketu.source/close-out-chan? opts)
163+
^long close-consumer? (:ketu.source/close-consumer? opts)
146164
consumer-close-timeout-ms (:ketu.source/consumer-close-timeout-ms opts)
147-
should-poll? (volatile! true)
148-
abort-pending-put (async/chan)
149-
done-putting (async/chan)
150-
subscribe! (or (subscribe-fn opts) (assign-fn opts))
151-
poll-impl (poll-fn consumer should-poll? opts)
152-
poll! (if (some? (:ketu.source/consumer-decorator opts))
153-
(consumer-decorator/decorate-poll-fn {:ketu.source/consumer consumer} poll-impl opts)
154-
poll-impl)
155-
->data (->data-fn opts)
156-
put! (fn [record] (put-or-abort-pending! out-chan (->data record) abort-pending-put))
165+
should-poll? (volatile! true)
166+
abort-pending-put (async/chan)
167+
done-putting (async/chan)
168+
subscribe! (or (subscribe-fn opts) (assign-fn opts))
169+
poll-impl (poll-fn consumer should-poll? opts)
170+
poll! (if (some? (:ketu.source/consumer-decorator opts))
171+
(consumer-decorator/decorate-poll-fn {:ketu.source/consumer consumer} poll-impl opts)
172+
poll-impl)
173+
->data (->data-fn opts)
174+
put! (fn [record] (put-or-abort-pending! out-chan (->data record) abort-pending-put))
157175

158176
consumer-thread
159-
(async/thread
160-
(try
161-
(.info logger "[source={}] Start consumer thread" source-name)
162-
(.setName (Thread/currentThread) thread-name)
163-
164-
(subscribe! consumer)
165-
166-
(loop []
167-
(when-let [records (poll!)]
168-
(run! put! records)
169-
(recur)))
170-
171-
(catch WakeupException e
172-
; We wakeup the consumer on graceful shutdown after should-poll? is false.
173-
; If it's not false, somebody else woke the consumer up unexpectedly.
174-
(when @should-poll?
175-
(log/error logger "[source={}] Unexpected consumer wakeup" source-name e)))
176-
(catch Exception e
177-
(log/error logger "[source={}] Unrecoverable consumer error" source-name e))
178-
(finally
179-
(log/info logger "[source={}] Done consuming" source-name)
180-
(async/close! done-putting)
181-
(when close-out-chan?
182-
(log/info logger "[source={}] Close out channel" source-name)
183-
(async/close! out-chan))
184-
(when close-consumer?
185-
(log/info logger "[source={}] Close consumer" source-name)
186-
(consumer/close! consumer consumer-close-timeout-ms))
187-
(log/info logger "[source={}] Exit consumer thread" source-name))))
188-
189-
190-
state {:ketu/source-opts opts
191-
:ketu.source/out-chan out-chan
192-
:ketu.source/consumer consumer
193-
:ketu.source/should-poll? should-poll?
194-
:ketu.source/abort-pending-put abort-pending-put
195-
:ketu.source/done-putting done-putting
196-
:ketu.source/consumer-thread consumer-thread}]
177+
(async/thread
178+
(try
179+
(.info logger "[source={}] Start consumer thread" source-name)
180+
(.setName (Thread/currentThread) thread-name)
181+
182+
(subscribe! consumer)
183+
184+
(loop []
185+
(when-let [records (poll!)]
186+
(run! put! records)
187+
(recur)))
188+
189+
(catch WakeupException e
190+
; We wakeup the consumer on graceful shutdown after should-poll? is false.
191+
; If it's not false, somebody else woke the consumer up unexpectedly.
192+
(when @should-poll?
193+
(log/error logger "[source={}] Unexpected consumer wakeup" source-name e)))
194+
(catch Exception e
195+
(log/error logger "[source={}] Unrecoverable consumer error" source-name e))
196+
(finally
197+
(log/info logger "[source={}] Done consuming" source-name)
198+
(async/close! done-putting)
199+
(when close-out-chan?
200+
(log/info logger "[source={}] Close out channel" source-name)
201+
(async/close! out-chan))
202+
(when close-consumer?
203+
(log/info logger "[source={}] Close consumer" source-name)
204+
(consumer/close! consumer consumer-close-timeout-ms))
205+
(log/info logger "[source={}] Exit consumer thread" source-name))))
206+
207+
208+
state {:ketu/source-opts opts
209+
:ketu.source/out-chan out-chan
210+
:ketu.source/consumer consumer
211+
:ketu.source/should-poll? should-poll?
212+
:ketu.source/abort-pending-put abort-pending-put
213+
:ketu.source/done-putting done-putting
214+
:ketu.source/consumer-thread consumer-thread}]
197215

198216
state))
199217

200218
(defn source
201219
"Starts consuming into a channel.
202220
Returns a state map including out-chan. Pass it to the `stop!` function when done."
203221
[ch opts]
204-
(let [opts (ketu.spec/assert-and-conform :ketu/public-source-opts opts)
205-
opts (finalize-opts opts)
206-
source-name (:ketu/name opts)
222+
(let [opts (ketu.spec/assert-and-conform :ketu/public-source-opts opts)
223+
opts (finalize-opts opts)
224+
source-name (:ketu/name opts)
207225
consumer-supplier (:ketu.source/consumer-supplier opts)
208-
consumer (try
209-
(consumer-supplier opts)
210-
(catch Exception e
211-
(log/error logger "[source={}] Error creating consumer-source" source-name e)
212-
(when (opts :ketu.source/close-out-chan?)
213-
(log/warn logger "[source={}] Close consumer channel" source-name)
214-
(async/close! ch))
215-
(throw e)))]
226+
consumer (try
227+
(consumer-supplier opts)
228+
(catch Exception e
229+
(log/error logger "[source={}] Error creating consumer-source" source-name e)
230+
(when (opts :ketu.source/close-out-chan?)
231+
(log/warn logger "[source={}] Close consumer channel" source-name)
232+
(async/close! ch))
233+
(throw e)))]
216234
(try
217235
(source-existing-consumer consumer ch opts)
218236
(catch Exception e
@@ -234,9 +252,9 @@
234252
(consumer/wakeup! consumer)))
235253

236254
(defn- wait-for-put! [state]
237-
(let [done-putting (:ketu.source/done-putting state)
255+
(let [done-putting (:ketu.source/done-putting state)
238256
done-putting-timeout-ms (-> state :ketu/source-opts :ketu.source/done-putting-timeout-ms)
239-
timeout (async/timeout done-putting-timeout-ms)]
257+
timeout (async/timeout done-putting-timeout-ms)]
240258
(async/alts!! [done-putting timeout] :priority true)))
241259

242260
(defn- abort-pending-put! [state]
@@ -252,9 +270,9 @@
252270
(:ketu.source/consumer-thread state))
253271

254272
(defn- wait-for-the-thread! [state]
255-
(let [consumer-thread (:ketu.source/consumer-thread state)
273+
(let [consumer-thread (:ketu.source/consumer-thread state)
256274
consumer-thread-timeout-ms (-> state :ketu/source-opts :ketu.source/consumer-thread-timeout-ms)
257-
timeout (async/timeout consumer-thread-timeout-ms)]
275+
timeout (async/timeout consumer-thread-timeout-ms)]
258276
(async/alts!! [consumer-thread timeout] :priority true)))
259277

260278
(defn stop! [state]

0 commit comments

Comments
 (0)