Net::Kafka(3pm)

BUILD FROM CPAN

    cpanm install Net::Kafka

BUILD FROM SOURCE

Sources are available on Github: <https://github.com/bookingcom/perl-Net-Kafka>.

    perl Makefile.pl
    make
    make test
    make install

METHODS

new()

    my $producer = Net::Kafka::Producer->new(
        'bootstrap.servers' => 'localhost:9092'
    );

Create an instance of Net::Kafka::Producer. Accept hash where keys are equal to property names of librdkafka (see <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md>). Note that only "error_cb" and "stats_cb" callbacks are supported for Producer. Message delivery reports are served automatically through "Promise" based "produce" method (see below).

produce()

    my $promise = $producer->produce(
        payload     => "my_message",
        topic       => "my_topic",
        key         => "my_key",    # optional
        timestamp   => 1234567,   # optional, if not specified current local timestamp will be used
        partition   => 0          # optional, if not specified internal librdkafka partitioner will be used
        headers     => $headers,  # Optional, see Net::Kafka::Headers
    )->then(sub {
        my $delivery_report = shift;
        print "Message is sent with offset " . $delivery_report->{offset};
    })->catch(sub {
        my $error = shift;
        print $error->{error} . "\n";
    });

Sends a message to Kafka. Accepts hash with parameters.

Returns back an instance of "Promise" that will be resolved/rejected later. In case message is successfully send "resolve" callback will receive a delievry report in the form of the hash that contains "offset", "partition" and "timestamp". If message delivery has failed "reject" callback will receive a hash that contains "error" (a human readable error description) and (optionally) "error_code" that is equal to librdkafka's error code. All error codes are mapped and exported by "Net::Kafka" module as constants (e.g. "Net::Kafka::RD_KAFKA_RESP_ERR__PREV_IN_PROGRESS") for simplicity.

partitions_for()

    my $partitions = $producer->partitions_for("my_topic", $timeout_ms);

Returns an "ARRAYREF" that contains partition metadata information about the given topic (leader, replicas, ISR replicas);

close()

    $producer->close();

Explicitly closees "Net::Kafka::Producer" instance and underlying librdkafka handles.

METHODS

new()

    my $consumer = Net::Kafka::Consumer->new(
        'bootstrap.servers'  => 'localhost:9092',
        'group.id'           => "my_consumer_group",
        'enable.auto.commit' => "true",
    );

Create an instance of Net::Kafka::Consumer. Accept hash where keys are equal to property names of librdkafka (see <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md>). Note that not all callbacks are supported at the moment. Supported ones are: "error_cb", "rebalance_cb", "commit_cb" and "stats_cb".

subscribe()

    $consumer->subscribe([ 'my_topic' ]);

Subscribe to topic set using balanced consumer groups. The main entry-point for "distributed" consumer mode - partitions will be assigned automatically using Kafka's GroupApi semantics. Wildcard/regex topics are supported so matching topics will be added to the subscription list.

unsubscribe()

    $consumer->unsubscribe();

Unsubscribe from the current subscription set.

assign()

    # manually assign partitions 0 and 1 to be consumed
    my $tp_list = Net::Kafka::TopicPartitionList->new();
    $tp_list->add("my_topic", 0);
    $tp_list->add("my_topic", 1);
    $consumer->assign($tp_list);

Atomic assignment of partitions to consume. The main entry-point for "simple" consumer mode - partitions are assigned manually.

poll()

    my $message = $consumer->poll($timeout_ms);

Poll the consumer for messages or events. Returns instance of "Net::Kafka::Message". Will block for at most "timeout_ms" milliseconds. An application should make sure to call "poll" at regular intervals.

committed()

    my $tp_list = Net::Kafka::TopicPartitionList->new();
    $tp_list->add("my_topic", 0);
    $consumer->committed($tp_list);
    my $offset = $tp_list->offset("my_topic_, 0);

Retrieve committed offsets for topics+partitions.

offsets_for_times()

    my $tp_list = Net::Kafka::TopicPartitionList->new();
    $tp_list->add("my_topic", 0);
    $tp_list->set_offset("my_topic", 0, 958349923); # timestamp if passed through offset field
    $consumer->offsets_for_times($tp_list);
    my $offset = $tp_list->offset("my_topic");

Look up the offsets for the given partitions by timestamp.

pause()

    my $tp_list = Net::Kafka::TopicPartitionList->new();
    $tp_list->add("my_topic", 0);
    $consumer->pause($tp_list); # pauses consumption of partition 0 of "my_topic"

Pause consumption for the provided list of partitions.

resume()

    my $tp_list = Net::Kafka::TopicPartitionList->new();
    $tp_list->add("my_topic", 0);
    $consumer->resume($tp_list); # resumes consumption of partition 0 of "my_topic"

Resume consumption for the provided list of partitions.

subscription()

    my $topics = $consumer->subscription();

Returns the current topic subscription

partitions_for()

    my $partitions = $producer->partitions_for("my_topic");

Returns an "ARRAYREF" that contains partition metadata information about the given topic (leader, replicas, ISR replicas);

commit()

    $consumer->commit(); # commit current partition assignment (blocking call)
    $consumer->commit(1); # commit current partition assignment (non-blocking call)
    my $tp_list = Net::Kafka::TopicPartitionList->new();
    $tp_list->add("my_topic", 0);
    $tp_list->set_offset("my_topic", 0, 12345);
    $consumer->commit(0, $tp_list); # commit $tp_list assignment (blocking call);

Commit offsets on broker for the provided list of partitions. If no partitions provided current assignment is committed instead.

commit_message();

    my $message = $consumer->poll(1000);
    $consumer->commit_message(0, $message); # commit message (blocking call);
    $consumer->commit_message(1, $message); # commit message (non-blocking call);

Commit message's offset on broker for the message's partition.

position()

    my $position_list = Net::Kafka::TopicPartitionList->new();
    $position_list->add("my_topic", 0);
    $consumer->position($position_list);
    my $position = $position_list->offset("my_topic", 0);

Retrieve current positions (offsets) for topics+partitions. The \p offset field of each requested partition will be set to the offset of the last consumed message + 1, or RD_KAFKA_OFFSET_INVALID in case there was no previous message.

Note: in this context the last consumed message is the offset consumed by the current librdkafka instance and, in case of rebalancing, not necessarily the last message fetched from the partition.

seek()

    $consumer->seek("my_topic", 0, 12345); # seek partition 0 of "my_topic" to offset "12345"
    $consumer->seek("my_topic", 0, RD_KAFKA_OFFSET_BEGINNING); # seek to the beginning of "my_topic" partition 0
    $consumer->seek("my_topic", 0, RD_KAFKA_OFFSET_END); # seek to the end of "my_topic" partition 0

Seek consumer for topic+partition to offset which is either an absolute or logical offset.

query_watermark_offsets()

    my ($low, $high) = $consumer->query_watermark_offsets("my_topic", 0);

Queries Kafka Broker for lowest and highest watermark offsets in the given topic-partition.

close()

    $consumer->close();

Close all consumer handles. Make sure to call it before destroying your application to make sure that all outstanding requests to be flushed.