Async consumer & manual offset control

[vielmath]

I had some trouble trying to manually commit offsets.

After initializing a consumer and registering it to a topic I start a simple loop like this one:

(go-loop []
    (when-let [{:keys [topic offset partition] :as event} (<! (first consumer))]
      (when (= :record (:type event))
        ;; SOME PROCESSING
        (put! (second consumer) {:op :commit
                                 :topic-offsets [{:topic topic
                                                  :partition partition
                                                  :offset (inc offset)
                                                  :metadata ""}]}))
      (recur)))

A few messages are processed and then it stops processing. The loop seems dead.

After a bit of investigation I found that it fails here https://github.com/pyr/kinsky/blob/master/src/kinsky/async.clj#L108

The commit operations are so close that a wake-up exception is thrown while processing a commit operation. There is no try catch here so the loop dies...

Even with a try catch, poller-ctl only processes one event so if multiple operations are submitted "simultaneously" only the first will be applied.

[fmjrey]

@vielmath I wonder if the changes I made in the fmjrey branch on my fork would resolve your issue:
https://github.com/fmjrey/kinsky/tree/fmjrey

[pyr]

@vielmath I released 0.1.23, can you verify with this version and see if you are still running into the same issue?

[vielmath]

@pyr 0.1.23 doesn't seem to fix my problem.
@fmjrey your branch doesn't do the trick either.

Thanks for trying ;)

[pyr]

@vielmath the async facade has been deprecated, sorry about this

[fmjrey]

The referenced line has sinced changed, I believe this is where the exception was raised:

kinsky/src/kinsky/async.clj

Line 108 in fdcb203

(client/commit! driver topic-offsets)

I've not encountered this issue, a good test case would help BTW.

Still, my guess is it's related to the control thread calling wakeup whenever there is a message on the control channel (source).

This makes me wonder why would we need a second thread for control? I imagine this was because of a consumer blocking indefinitely. However KIP-266: Fix consumer indefinite blocking behavior was released in 2.1, meaning most if not all consumer operations can now timeout, including commit.

So I wonder now, do we still need that second control thread which only purpose is to call wakeup on any input message?

@pyr @vielmath
Edit: I suggest we continue this discussion on fmjrey/kinsky-async#1