felice.consumer — com.oscaro/felice 3.2.0-1.5

assignment clj

(assignment consumer)

returns a set of topic-partition map currently assigned to this consumer.

close! clj

(close! consumer)

(close! consumer timeout)

Tries to close the consumer cleanly within the specified timeout in ms (defaults to 30 secs).

This method waits up to timeout for the consumer to complete pending commits and leave the group.

If auto-commit is enabled, this will commit the current offsets if possible within the timeout.

If the consumer is unable to complete offset commits and gracefully leave the group before the timeout expires, the consumer is force closed.

commit-message-offset clj

(commit-message-offset consumer {:keys [partition topic offset] :as record})

Commit a specific record

consumer must be a KafkaConsumer object

record must be a map with :partition :topic and :offset

commit-sync clj

(commit-sync consumer)

Commit offsets returned on the last poll() for all the subscribed list of topics and partitions.

consumer must be a KafkaConsumer object

CONF-COERCERS clj

consumer clj

(consumer conf)

(consumer conf topics)

(consumer conf key-deserializer value-deserializer)

(consumer conf key-deserializer value-deserializer topics)

create a consumer

conf is a map {:keyword value} See: https://kafka.apache.org/documentation/#consumerconfigs for all possibilities

key and value serializer can be one of keys defined in felice.serializer namespace with the 1 argument arity, :key.deserializer and :value.deserializer must be provided in conf

you can optionaly provide a list of topics to subscribe to

consumer-record->map clj

(consumer-record->map record)

transforms a ConsumerRecord to a clojure map containing: :key``:value :offset :topic :partition :timestamp :timestamp-type and :header

metrics clj

(metrics consumer)

returns a list of mtrics mapkept by the consumer

poll clj

(poll consumer timeout)

Fetch data for the topics or partitions specified using one of the subscribe/assign APIs.

This method returns immediately if there are records available. Otherwise, it will await the timeout ms.

If the timeout expires, an empty record set will be returned.

poll->all-records clj

(poll->all-records records)

takes the return off a poll (see ConsumerRecords) returns a lazy seq of records as clojure maps

poll->records-by-topic clj

(poll->records-by-topic records)

takes the return of a poll (see ConsumerRecords) returns a map {topic records-seq}

poll-and-process clj

(poll-and-process consumer timeout process-fn commit-policy)

Poll records and run process-fn on each of them (presumably for side effects)

poll-loop clj

(poll-loop consumer-conf process-record-fn)

(poll-loop consumer-conf process-record-fn opts)

Start a consumer loop, calling a callback for each record, and returning a function to stop the loop.

Parameters

         consumer: consumer config (see consumer)
process-record-fn: function to call with each record polled
          options: {:poll-timeout 2000 ; duration of a polling without events (ms)
                    :on-error-fn  (fn [ex] ...); called on exception
                    :commit-policy :never ; #{:never :poll :record}}

commit policy

• :never : does nothing (use it if you enabled client auto commit)

• :poll : commit last read offset after processing all the items of a poll

• :record : commit the offset of every processed record

  if you want to commit messages yourself, set commit policy to :never and use commit-message-offset or commit-sync

Returns

          stop-fn: callback function to stop the loop

poll-loop* clj

(poll-loop* consumer
            process-record-fn
            {:keys [poll-timeout on-error-fn commit-policy close-timeout-ms]
             :or {poll-timeout 2000 close-timeout-ms 5000}})

poll-loops clj

(poll-loops consumer-conf process-record-fn)

(poll-loops consumer-conf process-record-fn {:as opts})

(poll-loops consumer-conf
            process-record-fn
            topics
            {:keys [threads threads-by-topic] :as opts})

Start consumer loops, calling a callback for each record, and returning a function to stop the loops.

Parameters

         consumer: consumer config (see consumer)
process-record-fn: function to call with each record polled
           topics: topics you want to subscribe to
          options: {:poll-timeout 2000 ; duration of a polling without events (ms)
                    :on-error-fn  (fn [ex] ...); called on exception
                    :commit-policy :never ; #{:never :poll :record}
                    :threads-by-topic 1 ; number of spawned consumers for each topic
                    :threads 1 ; number of spawned consumers}

commit policy

• :never : does nothing (use it if you enabled client auto commit)

• :poll : commit last read offset after processing all the items of a poll

• :record : commit the offset of every processed record

  if you want to commit messages yourself, set commit policy to :never and use commit-message-offset or commit-sync

Multi-threading

You can set either :threads-by-topic or :threads option (if both are set, :threads-by-topic will win)

• :threads : spawn N threads total (each thread listening all registered topic)
• :threads-by-topic : spawn N threads for each registered topic
• you can also provide a map {:topic :threads} instead of a list of topics

Returns

          stop-fn: callback function to stop the loop

poll-loops* clj

(poll-loops* consumer-conf process-record-fn topics opts threads)

subscribe clj

(subscribe consumer & topics)

subscribe the consumer to one or more topics automaticly resubscribes previous subscriptions returns the consumer

subscription clj

(subscription consumer)

returns the set of currenctly subscribed topics

topic-partition->map clj

(topic-partition->map topic-partition)

converts a TopicPartition object to a clojure map containing :topic and :partition

unsubscribe clj

(unsubscribe consumer)

Unsubscribe from all topics currently subscribed returns the consumer

wakeup clj

(wakeup consumer)

Wakeup the consumer. This method is thread-safe and is useful in particular to abort a long poll.

The thread which is blocking in an operation will throw WakeupException.

If no thread is blocking in a method which can throw WakeupException, the next call to such a method will raise it instead.