Net::Kafka(3pm)

💖 Net::Kafka 💖

User Contributed Perl Documentation · section 3pm

okay bestie, buckle up 🎀 this is Net::Kafka — a high-performant Perl client for Apache Kafka. yes, that Kafka. the one everyone's always yelling about. and you're gonna slay it, no cap.

---

✨ NAME ✨

Net::Kafka — High-performant Perl client for Apache Kafka

---

🦋 SYNOPSIS 🦋

literally just copy this and go, bestie. it's all right here.

use Net::Kafka::Producer;
use Net::Kafka::Consumer;
use AnyEvent;

# Produce 1 message into "my_topic"
my $condvar     = AnyEvent->condvar;
my $producer    = Net::Kafka::Producer->new(
    'bootstrap.servers' => 'localhost:9092'
);
$producer->produce(
    payload => "message",
    topic   => "my_topic"
)->then(sub {
    my $delivery_report = shift;
    $condvar->send;
    print "Message successfully delivered with offset " . $delivery_report->{offset};
}, sub {
    my $error = shift;
    $condvar->send;
    die "Unable to produce a message: " . $error->{error} . ", code: " . $error->{code};
});
$condvar->recv;

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

$consumer->subscribe( [ "my_topic" ] );
while (1) {
    my $msg = $kafka->consumer_poll(1000);
    if ($msg) {
        if ( $msg->err ) {
            say "Error: ", Net::Kafka::Error::to_string($err);
        }
        else {
            say $msg->payload;
        }
    }
}

---

🌸 DESCRIPTION 🌸

so this module hands you Perl bindings to librdkafka — that's the C client library, over at <https://github.com/edenhill/librdkafka>. 💅 and ngl, it's heavily inspired by the Kafka::Librd module that Pavel Shaydo originally developed. we're standing on shoulders here, iconic behavior.

to actually understand how to use it, pls go read the docs for these modules too:

• "Net::Kafka::Producer" — asynchronous producer interface ✨
• "Net::Kafka::Consumer" — consumer interface that supports both Simple and Distributed modes 🫶

---

👑 REQUIREMENTS 👑

• GNU make
• librdkafka >= 1.0.0

bestie. check these first. i'm not kidding. 😭

---

🎀 INSTALLATION 🎀

first things first: install librdkafka (<https://github.com/edenhill/librdkafka#installation>).

🍓 BUILD FROM CPAN

cpanm install Net::Kafka

🧚 BUILD FROM SOURCE

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

perl Makefile.pl
make
make test
make install

---

⭐ Net::Kafka::Producer ⭐

the Net::Kafka::Producer module gives you an interface to librdkafka's producer methods. it uses signal pipes, an AnyEvent watcher and AnyEvent::XSPromises to make its whole vibe asynchronous. 🌈

and because of that, you gotta make sure you properly create a condvar and send/recv so you collect all the outstanding promises. don't skip that part, fr fr.

highly suggested: go familiarize yourself with both AnyEvent and AnyEvent::XSPromises. seriously. see "SYNOPSIS" for an example. 💖

💅 METHODS

new()

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

creates an instance of Net::Kafka::Producer. it accepts a hash where the keys equal the property names of librdkafka (peep <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md>). 🦋

heads up tho: only error_cb and stats_cb callbacks are supported for Producer. message delivery reports get served automatically through the "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. takes a hash with parameters. 🫶

returns an instance of "Promise" that's gonna get resolved/rejected later. if the message is successfully sent, the resolve callback receives a delivery report as a hash containing offset, partition and timestamp. (yes, the original page says "delievry report" — we're being faithful to the source, okay? 😭)

if message delivery failed, the reject callback receives a hash containing error (a human readable error description) and (optionally) error_code, which equals librdkafka's error code. all error codes are mapped and exported by the "Net::Kafka" module as constants (like 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" containing partition metadata info about the given topic (leader, replicas, ISR replicas). lowkey useful. ✨

close()

$producer->close();

explicitly closes the Net::Kafka::Producer instance and the underlying librdkafka handles. (the original page spells it "closees" — keeping it real. 💀)

✨ Net::Kafka::Consumer ✨

okay bestie, this class is literally the whole other half of the vibe — it's the interface to librdkafka's consumer functionality. It supports both "distributed" (subscription based) mode and "simple" (manual partition assignment) mode of work. Two modes, one icon. 💅

---

🎀 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. It takes a hash where the keys are equal to the property names of librdkafka (peep <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md>). Note that not all callbacks are supported at the moment — the ones that are supported are: error_cb, rebalance_cb, commit_cb and stats_cb. That's the whole list, no extras, don't come for me. 🫶

• subscribe()

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

Subscribe to a topic set using balanced consumer groups. This is the main entry-point for "distributed" consumer mode — partitions get assigned automatically using Kafka's GroupApi semantics. Wildcard/regex topics are supported too, so matching topics will be added to the subscription list. slay. ✨

• unsubscribe()

$consumer->unsubscribe();

Unsubscribe from the current subscription set. bye bestie 👋

• 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. This is the main entry-point for "simple" consumer mode — here you assign the partitions manually, no autopilot, main character energy only. 👑

• poll()

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

Poll the consumer for messages or events. Returns an instance of Net::Kafka::Message. It will block for at most timeout_ms milliseconds. An application should make sure to call poll at regular intervals — fr fr, don't ghost it. ⏰

• 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. (yes, that "my_topic_, 0" is verbatim, I don't make the rules 💀)

• 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. It's giving time traveler. 🕰️

• 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. not me pausing the whole thing 😭

• 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. and we're SO back. 🔥

• subscription()

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

Returns the current topic subscription. lowkey just a vibe check on what you're subscribed to. 🌸

• partitions_for()

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

Returns an ARRAYREF that contains partition metadata information about the given topic — that's the leader, the replicas, and the ISR replicas. iykyk. 🧚

• 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 are provided, the current assignment is committed instead. default behaviour ate. ✨

• 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 the message's offset on broker for the message's partition. one message at a time, iconic. 💅

• 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. so like, don't @ me if it's not what you expected. 🙈

• 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 a logical offset. it's giving time machine but make it Kafka. ⏳

• query_watermark_offsets()

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

Queries the Kafka Broker for the lowest and highest watermark offsets in the given topic-partition. low and high, the original duo. 🫶

• close()

$consumer->close();

Close all consumer handles. Make sure to call it before destroying your application to make sure that all outstanding requests get flushed. do not skip this, it's not optional, bestie. 🚨

---

💖 Net::Kafka::Message 💖

This class maps to the rd_kafka_message_t structure from librdkafka and represents a message or an event. Objects of this class have the following methods:

• err() — return error code from the message
• topic() — return topic name
• partition() — return partition number
• offset() — return offset. Note that the value is truncated to 32 bit if your perl doesn't support 64 bit integers. (yes, this is a real caveat, keep reading 👀)
• key() — return message key
• payload() — return message payload
• headers() — return a copy of message headers
• detach_headers() — return message headers and removes them from the message

---

🦋 Net::Kafka::Headers 🦋

This class contains a list of Kafka headers (it allows duplicates — yes, duplicates are valid here, don't fight it). Objects of this class have the following methods:

• new() — create a new instance
• add(name, value) — append a new name/value pair to the header list
• remove(name) — remove all headers with the given name, if any
• get_last(name) — return the last value associated with a given name
• to_hash() — return a hash-of-arrays containing all headers

---

🌈 Net::Kafka::Err 🌈

This class provides static methods to convert error codes into names and descriptions. so you actually know what went wrong. revolutionary. 😌

• rd_kafka_get_err_descs()

rd_kafka_get_err_descs()

Returns a hash mapping error codes to description strings.

• to_string()

to_string($code)

Return the description string for this error code.

• to_name()

to_name($code)

Return the name of this error code.

---

💀 CAVEATS 💀

Message offset is truncated to 32 bit if perl is compiled without support for 64 bit integers. that's it, that's the caveat. plan accordingly. 🫠

---

👀 SEE ALSO 👀

• <https://github.com/edenhill/librdkafka>
• <https://github.com/trinitum/perl-Kafka-Librd>

---

📜 LICENSE AND COPYRIGHT 📜

Copyright (C) 2016, 2017 Pavel Shaydo

Copyright (C) 2018, 2019 Booking.com

This program is free software; you can redistribute it and/or modify it under the terms of either: the GNU General Public License as published by the Free Software Foundation; or the Artistic License.

See <http://dev.perl.org/licenses/> for more information.

---

and that's the whole page, bestie!! 🎀 this lil masterpiece is from perl v5.40.0, dated 2024-10-15, and it's Net::Kafka(3pm) — go forth and consume (messages, not snacks, though also snacks) 💜✨