Not able to pause/resume consumtion #13
Comments
|
One possible solution is to add support for a custom function that would be called from within the poll loop, allowing clients of the library to add custom logic (which will also enable my use case of pause/resume). I have tested it locally and it seems to work. Sample code of suggested versionChanges in poll loop inside (defn- source-existing-consumer [^Consumer consumer out-chan opts]
'...
(let [custom-poll-fn (:ketu.source/on-poll-fn opts)]
(while [@should-poll?]
(let [records (poll!)]
(when custom-poll-fn
(custom-poll-fn {:consumer consumer}))
(run! put! records))))
'...)Client code: (let [pause-requested? (atom false)
on-poll-fn (let [paused? (atom false)]
(fn [{:keys [consumer]}]
(when (and @pause-requested? (not @paused?))
(.pause consumer (.assignment consumer))
(reset! paused? true)
(println (str (java.time.LocalDateTime/now)) :paused))
(when (and (not @pause-requested?) @paused?)
(.resume consumer (.paused consumer))
(reset! paused? false)
(println (str (java.time.LocalDateTime/now)) :resumed))))
consumer-source (source/source
(chan 10)
{:name "test-consumer"
:brokers "localhost:9092"
:topic "test1"
:group-id "group1"
:value-type :string
:shape :value
:on-poll-fn on-poll-fn})]
(Thread/sleep 1000) ; To allow consumer to start polling
(reset! pause-requested? true)
(Thread/sleep 1000)
(reset! pause-requested? false)
(Thread/sleep 1000)
(source/stop! consumer-source))Output: |
|
Another possible solution is to use a "commands" channel in which the client of the library will be given a channel to which it can push custom functions that will be executed from within the poll loop. Sample code of suggested versionChanges in poll loop inside source-existing-consumer fn (take item from commands chan if available and invoke it from the poll loop): (defn- source-existing-consumer [^Consumer consumer out-chan opts]
'...
(while @should-poll?
(let [records (poll!)]
(when-let [command (async/poll! commands-chan)]
(command {:consumer consumer}))
(run! put! records)))
'...)Client code: (let [consumer-source (source/source
(chan 10)
{:name "test-consumer"
:brokers "localhost:9092"
:topic "test1"
:group-id "group1"
:value-type :string
:shape :value})
commands-chan (:ketu.source/commands-chan consumer-source)]
(Thread/sleep 1000) ; To allow consumer to start polling
(>!! commands-chan (fn [{:keys [consumer]}]
(.pause consumer (.assignment consumer))
(println (str (java.time.LocalDateTime/now)) :paused)))
(Thread/sleep 1000)
(>!! commands-chan (fn [{:keys [consumer]}]
(.resume consumer (.paused consumer))
(println (str (java.time.LocalDateTime/now)) :resumed)))
(Thread/sleep 1000)
(source/stop! consumer-source))Output: |
|
@yaronthurm Thank you for the suggestions. I prefer the commands channel solution. Please submit a PR. My suggestions:
|
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Hello,
I was looking for a way to pause/resume consumption using the library but it seems that this is not supported.
I tried to achive this "manually" by using the underlying kafka consumer and the raw java API, but since the consumer is not thread safe, an
ConcurrentModificationExceptionexception is thrown when I attempt to call .pause().It seems that the only way to achieve this is by calling pause/resume from within the poll loop but this loop is handled by the library itself and cannot be modified.
Sample Code
Output:
The text was updated successfully, but these errors were encountered: