KafkaConsumer

High-level KafkaConsumer (for brokers 0.9 and later)

Note: Requires Apache Kafka >= 0.9.0 brokers

Currently supports the \c range and \c roundrobin partition assignment strategies (see \c partition.assignment.strategy)

class KafkaConsumer : Handle {

this(GlobalConf conf);

ErrorCode assignment(TopicPartition[] partitions);

ErrorCode subscribe(const(char)*[] topics);

auto unsubscribe();

auto consume(Message msg, int timeout_ms);

ErrorCode assign(TopicPartition[] partitions);

ErrorCode unassign();

ErrorCode commitSync();

ErrorCode commitAsync();

ErrorCode commitSync(Message message);

ErrorCode commitAsync(Message message);

ErrorCode commitSync(TopicPartition[] offsets);

ErrorCode commitAsync(TopicPartition[] offsets);

ErrorCode committed(TopicPartition[] partitions, int timeout_ms);

ErrorCode position(TopicPartition[] partitions);

ErrorCode close();

}

Constructors

this this(GlobalConf conf): Creates a KafkaConsumer.

Members

Functions

assign ErrorCode assign(TopicPartition[] partitions): Update the assignment set to \p partitions.
assignment ErrorCode assignment(TopicPartition[] partitions): Returns the current partition assignment as set by * assign()
close ErrorCode close(): Close and shut down the proper.
commitAsync ErrorCode commitAsync(): Asynchronous version of CommitSync()
commitAsync ErrorCode commitAsync(Message message): Commit offset for a single topic+partition based on \p message
commitAsync ErrorCode commitAsync(TopicPartition[] offsets): Commit offset for the provided list of partitions.
commitSync ErrorCode commitSync(): Commit offsets for the current assignment.
commitSync ErrorCode commitSync(Message message): Commit offset for a single topic+partition based on \p message
commitSync ErrorCode commitSync(TopicPartition[] offsets): Commit offsets for the provided list of partitions.
committed ErrorCode committed(TopicPartition[] partitions, int timeout_ms): Retrieve committed offsets for topics+partitions.
consume auto consume(Message msg, int timeout_ms)
position ErrorCode position(TopicPartition[] partitions): Retrieve current positions (offsets) for topics+partitions.
subscribe ErrorCode subscribe(const(char)*[] topics): Update the subscription set to \p topics.
unassign ErrorCode unassign(): Stop consumption and remove the current assignment.
unsubscribe auto unsubscribe(): Unsubscribe from the current subscription set.

Meta

Source

See Implementation

rdkafkad handler kafka_consumer classes

KafkaConsumer