Manually assign topics and partitions to this consumer.

Get the set of partitions currently assigned to this consumer.

Provides two ways to close things: a default one with 'close [thing]'
and the one with the specified timeout.

Commit offsets returned by the last poll for all subscribed topics and partitions, or
manually specify offsets to commit.

offsets (optional) - commit the specified offsets for the specified list of topics
and partitions to Kafka. A seq of offset maps, as below:

e.g. {:topic "foo"
      :partition 1
      :offset 42}

optionally provide metadata:

e.g. {:topic "bar"
      :partition 0
      :offset 17
      :metadata "Dude, that's so meta."}

Commit offsets returned by the last poll for all subscribed topics and partitions,
or manually specify offsets to commit.

This is an asynchronous call and will not block. Any errors encountered are either
passed to the callback (if provided) or discarded.

offsets (optional) - commit the specified offsets for the specified list of topics
and partitions to Kafka. A seq of offset maps, as below:

e.g. {:topic "foo"
      :partition 1
      :offset 42}

optionally provide metadata:

e.g. {:topic "bar"
      :partition 0
      :offset 17
      :metadata "Dude, that's so meta."}

The committed offset should be the next message your application will consume,
i.e. lastProcessedMessageOffset + 1.

Return OffsetAndMetadata of the last committed offset for the given partition. This
offset will be used as the position for the consumer in the event of a failure.

If no offsets have been committed, return nil.

Return a KafkaConsumer.

Args:
  servers: comma-separated host:port strs or list of strs as bootstrap servers
  group-id: str that identifies the consumer group this consumer belongs to
  topics: an optional list of topics to which the consumer will be dynamically
          subscribed.
  config: an optional map of str to str containing additional consumer
          configuration. More info on optional config is available here:
          http://kafka.apache.org/documentation.html#newconsumerconfigs

The StringDeserializer class is the default for both key.deserializer and
value.deserializer.

Create a topic.

Args:
  zk-config: A map with Zookeeper connection details as expected by
             with-zookeeper.
  topic: The name of the topic to create.
  An unnamed configuration map. Valid keys are as follows:

    :partitions         (optional) The number of ways to partition the topic.
                                   Defaults to 1.
    :replication-factor (optional) The replication factor for the topic.
                                   Defaults to 1.
    :config             (optional) A map of configuration options for the
                                   topic.
    :rack-aware-mode    (optional) Control how rack aware replica assignment
                                   is done. Valid values are :disabled,
                                   :enforced, :safe. Default is :safe.

Delete a topic.

Args:
  zk-config: A map with Zookeeper connection details as expected by
             with-zookeeper.
  topic: The name of the topic to delete.

Invoking this method makes all buffered records immediately available to send (even if
linger.ms is greater than 0) and blocks on the completion of the requests associated
with these records.

Metadata for when an offset is committed.

Suspend fetching for a seq of topic name, partition number pairs.

Return a seq of consumer records currently available to the consumer (via a single poll).
Fetches sequetially from the last consumed offset.

A consumer record is represented as a clojure map with corresponding keys :value, :key,
:partition, :topic, :offset

timeout - the time, in milliseconds, spent waiting in poll if data is not
available. If 0, returns immediately with any records that are available now.
Must not be negative.

Return the offset of the next record that will be fetched (if a record with that
offset exists).

Return a KafkaProducer.

The producer is thread safe and sharing a single producer instance across
threads will generally be faster than having multiple instances.

Args:
 servers: comma-separated host:port strs or list of strs as bootstrap servers
 config: an optional map of str to str containing additional producer
         configuration. More info on optional config is available here:
         http://kafka.apache.org/documentation.html#producerconfigs

The StringSerializer class is the default for both key.serializer and value.serializer

Return a lazy sequence of sequences of consumer-records by polling the consumer.

Each element in the returned sequence is the seq of consumer records returned from a
poll by the consumer. The consumer fetches sequetially from the last consumed offset.

A consumer record is represented as a clojure map with corresponding keys :value, :key,
:partition, :topic, :offset

timeout - the time, in milliseconds, spent waiting in poll if data is not
available. If 0, returns immediately with any records that are available now.
Must not be negative.

Resume specified partitions which have been paused.

Overrides the fetch offsets that the consumer will use on the next poll.

Seek to the :beginning or :end offset for each of the given partitions.

Asynchronously send a record to a topic, providing at least a topic and value.

Asynchronously send a record to a topic, providing at least a topic and value, and
invoke the provided callback when the send has been acknowledged.

The callback function should take 2 args:
  - a metadata map: the metadata for the record that was sent. Keys
    are :topic, :partition, :offset.
  - a java.lang.Exception object: the exception thrown during processing of this
    record.

Subscribe to the given list of topics to get dynamically assigned partitions. Topic
subscriptions are not incremental. This list will replace the current assignment (if
there is one). It is not possible to combine topic subscription with group management
with manual partition assignment through assign(List). If the given list of topics is
empty, it is treated the same as unsubscribe.

topics-or-regex can be a list of topic names or a java.util.regex.Pattern object to
subscribe to all topics matching a specified pattern.

the optional functions are a callback interface to trigger custom actions when the set
of partitions assigned to the consumer changes.

Get the current subscription for this consumer.

Query whether or not a topic exists.

Args:
  zk-config: A map with Zookeeper connection details as expected by
             with-zookeeper.
  topic: The name of the topic to check for.

A topic name and partition number.

Query existing topics.

Args:
  zk-config: A map with Zookeeper connection details as expected by
             with-zookeeper.

Unsubscribe from topics currently subscribed with subscribe. This also clears any
partitions directly assigned through assign.

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.

A utility macro for interacting with Zookeeper.

Args:
  zk-config: A map with Zookeeper connection details. This will be validated
             using validate-zookeeper-config before use.
  zookeeper: This will be bound to an instance of ZkUtils while the body is
             executed. The connection to Zookeeper will be cleaned up when
             the body exits.