uperl / kafka-librd Goto Github PK

perl bindings to librdkafka

Perl 55.82% XS 29.84% C 14.12% Shell 0.23%

kafka-librd's Introduction

NAME
    Kafka::Librd - bindings for librdkafka

SYNOPSIS
        use Kafka::Librd;

        my $kafka = Kafka::Librd->new(
            Kafka::Librd::RD_KAFKA_CONSUMER,
            {
                "group.id" => 'consumer_id',
            },
        );
        $kafka->brokers_add('server1:9092,server2:9092');
        $kafka->subscribe( \@topics );
        while (1) {
            my $msg = $kafka->consumer_poll(1000);
            if ($msg) {
                if ( $msg->err ) {
                    say "Error: ", Kafka::Librd::Error::to_string($err);
                }
                else {
                    say $msg->payload;
                }
            }
        }

DESCRIPTION
    This module provides perl bindings for librdkafka.

METHODS
  new
        $kafka = $class->new($type, \%config)

    Create a new instance. $type can be either "RD_KAFKA_CONSUMER" or
    "RD_KAFKA_PRODUCER". Config is a hash with configuration parameters as
    described in
    <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md>,
    additionally it may include "default_topic_config" key, with a hash
    containing default topic configuration properties.

  brokers_add
        $cnt = $kafka->brokers_add($brokers)

    add one or more brokers to the list of initial bootstrap brokers.
    *$brokers* is a comma separated list of brokers in the format
    "[proto://]host[:port]".

  subscribe
        $err = $kafka->subscribe(\@topics)

    subscribe to the list of topics using balanced consumer groups.

  unsubscribe
        $err = $kafka->unsubscribe

    unsubscribe from the current subscription set

  subscription
        $tplist = $kafka->subscription

    return current subscriptions. Subscription returned as a reference to
    array of hashes with the following fields: "topic", "partition",
    "offset", "metadata".

  assign
        $err = $kafka->assign(\@tplist)

    assign partitions to consume. @tplist is an array of hashes with "topic"
    and "partition" fields set.

  assignment
        $tplist = $kafka->assignment

    return current assignment. Result returned in the same way as for
    "subscription".

  consumer_poll
        $msg = $kafka->consumer_poll($timeout_ms)

    poll for messages or events. If any message or event received, returns
    "Kafka::Librd::Message" object. If "<$msg-"err>> for returned object is
    zero (RD_KAFKA_RESP_ERR_NO_ERROR), then it is a proper message,
    otherwise it is an event or an error.

  commit
        $err = $kafka->commit(\@tplist, $async)

    commit offsets to the broker. @tplist is an array of hashes with the
    following keys: "topic", "partition", "offset", "metadata". If
    @topic_partition_list is missing or undef, then current partition
    assignment is used instead. If $async is 1, then method returns
    immediately, if it is 0 or missing then method blocks until offsets are
    committed.

  commit_message
        $err = $kafka->commit_message($msg, $async)

    commit message's offset for the message's partition. $async same as for
    "commit".

  committed
        $tplist = $kafka->committed(\@tplist, $timeout_ms)

    retrieve committed offsets for topics and partitions specified in
    @tplist, which is an array of hashes with "topic" and "partition"
    fields. Returned $tplist contains a copy of the input list with added
    "offset" fields.

  position
        $tplist = $kafka->position(\@tplist)

    retrieve current offsets for topics and partitions specified in @tplist,
    which is an array of hashes with "topic" and "partition" fields.
    Returned $tplist contains a copy of the input list with added "offset"
    fields.

  consumer_close
        $err = $kafka->consumer_close

    close down the consumer

  topic
        $topic = $kafka->topic($name, \%config)

    Return a topic object, that can be used to produce messages.

    If an error occurs during creation of the topic, "undef" is returned. In
    such case use "Kafka::Librd::Error::last_error" to obtain the
    corresponding error code!

  outq_len
        $len = $kafka->outq_len

    return the current out queue length.

  flush
        $kafka->flush($timeout_ms)

    wait until all outstanding produce requests, et.al, are completed.

  destroy
        $kafka->destroy

    destroy kafka handle

  dump
        $kafka->dump

    dump internal state of kafka handle to stdout, only useful for debugging

Kafka::Librd::Topic
    This class maps to "rd_kafka_topic_t" structure from librdkafka and
    represents topic. It should be created with "topic" method of
    Kafka::Librd object. It provides the following method:

  produce
        $status = $topic->produce($partition, $msgflags, $payload, $key)

    produce a message for the topic. *$msgflags* can be RD_KAFKA_MSG_F_BLOCK
    in the future, but currently it should be set to 0, RD_KAFKA_MSG_F_COPY
    and RD_KAFKA_MSG_F_FREE must not be used, internally RD_KAFKA_MSG_F_COPY
    is always set.

    The returned status is -1 in case of an error, otherwise 0. The error
    code can be retrieved using the "Kafka::Librd::Error::last_error"
    function.

  destroy
        $topic->destroy

    destroy topic handle

Kafka::Librd::Message
    This class maps to "rd_kafka_message_t" structure from librdkafka and
    represents message or 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.

  key
    return message key

  payload
    return message payload

  timestamp(\$tstype)
    return message timestamp (milliseconds since UNIX epoch)

    The $tstype argument is optional, and if present, it should be a scalar
    reference. It will be filled with one of the following values:

    *   "Kafka::Librd::RD_KAFKA_TIMESTAMP_NOT_AVAILABLE"

    *   "Kafka::Librd::RD_KAFKA_TIMESTAMP_CREATE_TIME"

    *   "Kafka::Librd::RD_KAFKA_TIMESTAMP_LOG_APPEND_TIME"

Kafka::Librd::Error
  Kafka::Librd::Error::to_string
       my $error_message =  Kafka::Librd::Error::to_string($err)

    Convert an error code into a human-readable error description. Use this
    for error codes returned by "Kafka::Librd::Error::last_error" and
    Kafka::Librd::Message::err.

  Kafka::Librd::Error::last_error
        my $err = Kafka::Librd::Error::last_error

    Retrieve the last error state set by function calls "topic" and
    "produce". This function should be called immediately after those
    functions, since they store error information globally.

CAVEATS
    Message offset is truncated to 32 bit if perl compiled without support
    for 64 bit integers.

SEE ALSO
    <https://github.com/edenhill/librdkafka>

BUGS
    Please report any bugs or feature requests via GitHub bug tracker at
    <http://github.com/uperl/Kafka-Librd/issues>.

AUTHOR
    Original author Pavel Shaydo "<zwon at cpan.org>"

    Current maintainer Graham Ollis "<[email protected]>"

LICENSE AND COPYRIGHT
    Copyright (C) 2016, 2017 Pavel Shaydo

    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.

kafka-librd's People

Contributors

Stargazers

Watchers

Forkers

trueneu mikey-austin idealo miketonks mrdvt92 harmathy eserte ibiscum

kafka-librd's Issues

Error handling of produce method

As of now the produce method only returns -1 in case of an error. But there is no way to handle different errors appropriately or get an error description.

Implement access to all methods of High-level KafkaConsumer C API

Batch Processing

Hello

Do you plan to implement a batch processing?

RD_EXPORT
ssize_t rd_kafka_consume_batch(rd_kafka_topic_t *rkt, int32_t partition,
int timeout_ms,
rd_kafka_message_t **rkmessages,
size_t rkmessages_size);

thnx in advance

Implement rd_kafka_conf_set_log_cb

The default behavior of librdkafka is to log events like connection errors to stderr. In order to have the ability to change this behavior and set a custom logging function Kafka::Librd should implement rd_kafka_conf_set_log_cb.

Better test coverage

It would be nice to further increase test coverage. To do so I think we have to run tests against a real kafka server. There's some work in progress to implement this: https://github.com/idealo/Kafka-Librd/blob/more-testing-with-broker/t/fulltest.t
This test script starts zookeeper and kafka on the fly with temporary directories and ports, and runs some Kafka::Librd commands on this.

However a lot of polishing is still needed:

probably this test script should only run in some environments (e.g. in some travis-ci configurations, and if the user explicitly requests it)
kafka should be made available, either downloaded by the test script or outside using .travis.yml
zookeeper should be made available, either downloaded by the test script or outside using .travis.yml; installation could also be done via package manager at least on debian-based systems
there are some sleep() calls in the test script which should be removed
more test cases have to be written

Producer not working

I have a perl producer that looks like the following:

my $kafka_producer = Kafka::Librd->new( Kafka::Librd::RD_KAFKA_PRODUCER, {'debug' => 'all'} );
$kafka_producer->brokers_add("$ENV{KAFKA_HOST}:9092");

my $kafka_topic = $kafka_producer->topic( "$ENV{KAFKA_TOPIC}", {} );

# <stuff>

    eval {
        my $response = $kafka_topic->produce(
            $partition,        # partition (zero based)
            0,                 # msgflags
            $json,             # single email + metadata
            "$msg_hash"        # key
        );
    };
    if ( $@ ) {
        die "can't send to kafka: $@";
    }

The process exits successfully, and says that it produced the messages.
However, the messages never show up in kafka (I've tested both with my own consumer, and a kafkacat consumer).

The output with the debug flag is:

%7|1510013279.519|BRKMAIN|rdkafka#producer-1| [thrd::0/internal]: :0/internal: Enter main broker thread
%7|1510013279.519|STATE|rdkafka#producer-1| [thrd::0/internal]: :0/internal: Broker changed state INIT -> UP
%7|1510013279.519|BROADCAST|rdkafka#producer-1| [thrd::0/internal]: Broadcasting state change
%7|1510013279.519|WAKEUPFD|rdkafka#producer-1| [thrd:app]: kafka:9092/bootstrap: Enabled low-latency ops queue wake-ups
%7|1510013279.519|BROKER|rdkafka#producer-1| [thrd:app]: kafka:9092/bootstrap: Added new broker with NodeId -1
%7|1510013279.519|TOPIC|rdkafka#producer-1| [thrd:app]: New local topic: inbox-monitor-ref-emails_dev
%7|1510013279.519|BRKMAIN|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap: Enter main broker thread
%7|1510013279.519|CONNECT|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap: broker in state INIT connecting
%7|1510013279.519|TOPPARNEW|rdkafka#producer-1| [thrd:app]: NEW inbox-monitor-ref-emails_dev [-1] 0x18d8c10 (at rd_kafka_topic_new0:282)
%7|1510013279.519|METADATA|rdkafka#producer-1| [thrd:app]: Skipping metadata refresh of 1 topic(s): no usable brokers
%7|1510013279.522|CONNECT|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap: Connecting to ipv4#172.18.0.2:9092 (plaintext) with socket 7
%7|1510013279.523|STATE|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap: Broker changed state INIT -> CONNECT
%7|1510013279.523|BROADCAST|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: Broadcasting state change
Creating message for: xxxx
%7|1510013279.533|CONNECT|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap: Connected to ipv4#172.18.0.2:9092
%7|1510013279.533|CONNECTED|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap: Connected (#1)
%7|1510013279.533|APIVERSION|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap: Using (configuration fallback) 0.9.0 protocol features
%7|1510013279.533|APIVERSION|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap:  Feature MsgVer1: Produce (2..2) NOT supported by broker
%7|1510013279.533|APIVERSION|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap:  Feature MsgVer1: Fetch (2..2) NOT supported by broker
%7|1510013279.533|APIVERSION|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap: Disabling feature MsgVer1
%7|1510013279.533|APIVERSION|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap:  Feature ApiVersion: ApiVersion (0..0) NOT supported by broker
%7|1510013279.533|APIVERSION|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap: Disabling feature ApiVersion
%7|1510013279.533|APIVERSION|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap:  Feature BrokerGroupCoordinator: GroupCoordinator (0..0) supported by broker
%7|1510013279.533|APIVERSION|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap: Enabling feature BrokerGroupCoordinator
%7|1510013279.533|APIVERSION|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap:  Feature BrokerBalancedConsumer: GroupCoordinator (0..0) supported by broker
%7|1510013279.533|APIVERSION|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap:  Feature BrokerBalancedConsumer: OffsetCommit (1..2) supported by broker
%7|1510013279.533|APIVERSION|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap:  Feature BrokerBalancedConsumer: OffsetFetch (1..1) supported by broker
%7|1510013279.533|APIVERSION|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap:  Feature BrokerBalancedConsumer: JoinGroup (0..0) supported by broker
%7|1510013279.533|APIVERSION|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap:  Feature BrokerBalancedConsumer: SyncGroup (0..0) supported by broker
%7|1510013279.533|APIVERSION|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap:  Feature BrokerBalancedConsumer: Heartbeat (0..0) supported by broker
%7|1510013279.533|APIVERSION|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap:  Feature BrokerBalancedConsumer: LeaveGroup (0..0) supported by broker
%7|1510013279.533|APIVERSION|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap: Enabling feature BrokerBalancedConsumer
%7|1510013279.533|APIVERSION|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap:  Feature ThrottleTime: Produce (1..2) supported by broker
%7|1510013279.533|APIVERSION|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap:  Feature ThrottleTime: Fetch (1..2) supported by broker
%7|1510013279.533|APIVERSION|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap: Enabling feature ThrottleTime
%7|1510013279.533|APIVERSION|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap:  Feature Sasl: JoinGroup (0..0) supported by broker
%7|1510013279.533|APIVERSION|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap: Enabling feature Sasl
%7|1510013279.533|APIVERSION|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap:  Feature SaslHandshake: SaslHandshake (0..0) NOT supported by broker
%7|1510013279.533|APIVERSION|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap: Disabling feature SaslHandshake
%7|1510013279.533|APIVERSION|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap:  Feature LZ4: GroupCoordinator (0..0) supported by broker
%7|1510013279.533|APIVERSION|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap: Enabling feature LZ4
%7|1510013279.533|APIVERSION|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap:  Feature : Offset (1..1) NOT supported by broker
%7|1510013279.533|APIVERSION|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap: Disabling feature
%7|1510013279.533|FEATURE|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap: Updated enabled protocol features to BrokerBalancedConsumer,ThrottleTime,Sasl,BrokerGroupCoordinator,LZ4
%7|1510013279.533|STATE|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap: Broker changed state CONNECT -> UP
%7|1510013279.533|BROADCAST|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: Broadcasting state change
%7|1510013279.533|METADATA|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: Hinted cache of 1/1 topic(s) being queried
%7|1510013279.533|METADATA|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: Requesting metadata for 1/1 topics: connected
%7|1510013279.533|METADATA|rdkafka#producer-1| [thrd:kafka:9092/bootstrap]: kafka:9092/bootstrap: Request metadata for 1 topic(s): connected

Your docs have good examples for how to create a consumer (which worked for me, thanks!), but there aren't any examples of producers.
Is there something I'm missing?
thank you...

Build on CentOS7

I had the hardest time building this package on CentOS 7 and so I added my spec to my fork and the required updates to drop Alien that is not needed under package managed systems.

https://github.com/mrdvt92/Kafka-Librd

$ git clone https://github.com/mrdvt92/Kafka-Librd.git
$ cd Kafka-Librd/
$ perl Makefile.PL
$ make
$ make test
$ make dist
$ rpmbuild -ta Kafka-Librd-0.14.tar.gz

I'm not exactly sure how to package the deltas as a patch so that you could just merge this fork into your master. In any event, this is done for any others out there.

Build warning: label ‘CROAK’ defined but not used

This is just a warning and I don't know C well enough to determine why. The tests pass an there does not appear to be any operational impact.

My Kafka versions on CentOS Linux release 7.7.1908 (Core)
librdkafka-devel-0.11.5-1.el7.x86_64
librdkafka-0.11.5-1.el7.x86_64

`
gcc -c -D_REENTRANT -D_GNU_SOURCE -fno-strict-aliasing -pipe -fstack-protector -I/usr/local/include -D_LARGEFILE_SOURCE -D_FILE_OFFSET_BITS=64 -O2 -g -pipe -Wall -Wp,-D_FORTIFY_SOURCE=2 -fexceptions -fstack-protector-strong --param=ssp-buffer-size=4 -grecord-gcc-switches -m64 -mtune=generic -DVERSION="0.14" -DXS_VERSION="0.14" -fPIC "-I/usr/lib64/perl5/CORE" rdkafkaxs.c

rdkafkaxs.c: In function ‘krd_expand_topic_partition_list’:

rdkafkaxs.c:82:1: warning: label ‘CROAK’ defined but not used [-Wunused-label]

CROAK:

Topic with SSL

Hi, the topic that I am trying to connect requires the client to connect with SSL. I was given a jks file; how do I pass the jks file location to the module? Is there any example?

Support for KAFKA-4208, Record Headers

Hello,

I would like to use Kafka Record Headers (KEFKA-4208) to store some message metadata, but it seems that Kafka::Librd doesn't expose the support for it that already exists in librdkafka.

Is someone working on this?

Thanks,

Failed to create thread (fedora only)

t/basic.t fails on my fedora 31 smokers (perls compiled without threads, if that matters). No errors on my other systems (centos, debian, ubuntu, freebsd):

Failed to create thread: No such file or directory (2) at /home/cpansand/.cpan/build/2020031700/Kafka-Librd-0.15-1/blib/lib/Kafka/Librd.pm line 66.
t/basic.t ...... 
Dubious, test returned 2 (wstat 512, 0x200)
No subtests run

(And probably $! isn't set here, the error message looks dubious)

Move commit_message to Message object

Is it possible to copy the commit_message method to the Kafka::Librd::Message object

$err = $kafka->commit_message($msg, $async);

should be

$err = $msg->commit;
$err = $msg->commit($async);

Possible thread race condition

Hello,
We've been trying to get version 0.08 Kafka::Librd to work through Alien-Librdkafka-v0.9.5 to a 6-node Kafka cluster. An apparent race condition in the librdkafka library emerges with broker lists > 1 in length is used. We happen to be using perl 5.16.3, but verified the problem in other versions as well, in both 32-bit and 64-bit versions. Sometime perl dies with a segmentation violation, or it may produce a "perl: double free or corruption" with a stack dump (see below). The race condition appears to occur between the Kafka::Librd->new() and $kafka_librd->brokers_add() calls. Putting a sleep(10) between the calls, for example, seems to mitigate the problem somewhat. Increasing the number of brokers increases the probability of a crash.
The reason we're contacting you instead of Alien::Librdkafka is that we could not reproduce the problem with their examples/rdkafka_example program, which basically is doing the same things as your module. We could not capture the problem with the gdb debugger, as the debugger must be doing something to control the operation of the threads.
We were wondering if you're already aware of the issue and are addressing it. If not, we may try to rework the Rdkafka.xs to see if we can make it match the behavior of the rdkafka_example c program, which works flawlessly.

Thanks for your efforts!
D Weber
Ticketmaster.com

Stack trace from perl:
*** glibc detected *** perl: double free or � �� *** glibc detected *** perl: ======= Backtrace: =========
/lib64/libc.so.6[0x2ba50d8964af]
/lib64/libc.so.6(cfree+0x4b)[0x2ba50d89a7ab]
/lib64/libc.so.6(fclose+0x14b)[0x2ba50d884d5b]
/lib64/libnss_files.so.2(_nss_files_gethostbyname2_r /lib64/libc.so.6[0x2ba50d8e176b]
/lib64/libc.so.6(getaddrinfo+0x21a)[0x2ba50d8e39ba]< /app/shared/lib/cpan64/lib/perl5/auto/share/dist/Ali /app/shared/lib/cpan64/lib/perl5/auto/share/dist/Ali /app/shared/lib/cpan64/lib/perl5/auto/share/dist/Ali /lib64/libpthread.so.0[0x2ba51269d83d]
/lib64/libc.so.6[0x2ba50d8e1bbc]
/lib64/libc.so.6(getaddrinfo+0x21a)[0x2ba50d8e39ba]< /lib64/libc.so.6(clone+0x6d)[0x2ba50d8fa18d]
======= Memory map: ========
00400000-00402000 r-xp 00000000 00:15 35346834 00601000-00602000 rw-p 00001000 00:15 35346834 0296e000-03054000 rw-p 0296e000 00:00 0
2ba50c7b4000-2ba50c7d0000 r-xp 00000000 fd:00 713155 2ba50c7d0000-2ba50c7d2000 rw-p 2ba50c7d0000 2ba50c9d0000-2ba50c9d1000 r--p 0001c000 fd:00 713155 2ba50c9d1000-2ba50c9d2000 rw-p 0001d000 fd:00 713155 2ba50c9d2000-2ba50cb33000 r-xp 00000000 00:15 103262105 2ba50cb33000-2ba50cd33000 ---p 00161000 00:15 103262105 2ba50cd33000-2ba50cd3e000 rw-p 00161000 00:15 103262105 2ba50cd49000-2ba50cd5e000 r-xp 00000000 fd:00 713172 2ba50cd5e000-2ba50cf5d000 ---p 00015000 fd:00 713172 2ba50cf5d000-2ba50cf5e000 r--p 00014000 fd:00 713172 2ba50cf5e000-2ba50cf5f000 rw-p 00015000 fd:00 713172 2ba50cf5f000-2ba50cf62000 rw-p 2ba50cf5f000 2ba50cf62000-2ba50cf64000 r-xp 00000000 fd:00 713168 2ba50cf64000-2ba50d164000 ---p 00002000 fd:00 713168 2ba50d164000-2ba50d165000 r--p 00002000 fd:00 713168 2ba50d165000-2ba50d166000 rw-p 00003000 fd:00 713168 2ba50d166000-2ba50d1e8000 r-xp 00000000 fd:00 713170 2ba50d1e8000-2ba50d3e7000 ---p 00082000 fd:00 713170 2ba50d3e7000-2ba50d3e8000 r--p 00081000 fd:00 713170 2ba50d3e8000-2ba50d3e9000 rw-p 00082000 fd:00 713170 2ba50d3e9000-2ba50d3f2000 r-xp 00000000 fd:00 713166 2ba50d3f2000-2ba50d5f1000 ---p 00009000 fd:00 713166 2ba50d5f1000-2ba50d5f2000 r--p 00008000 fd:00 713166 2ba50d5f2000-2ba50d5f3000 rw-p 00009000 fd:00 713166 2ba50d5f3000-2ba50d622000 rw-p 2ba50d5f3000 2ba50d622000-2ba50d624000 r-xp 00000000 fd:00 713194 2ba50d624000-2ba50d823000 ---p 00002000 fd:00 713194 2ba50d823000-2ba50d824000 r--p 00001000 fd:00 713194 2ba50d824000-2ba50d825000 rw-p 00002000 fd:00 713194 2ba50d825000-2ba50d974000 r-xp 00000000 fd:00 713162 2ba50d974000-2ba50db74000 ---p 0014f000 fd:00 713162 2ba50db74000-2ba50db78000 r--p 0014f000 fd:00 713162 2ba50db78000-2ba50db79000 rw-p 00153000 fd:00 713162 2ba50db79000-2ba50db80000 rw-p 2ba50db79000 2ba50db80000-2ba51114f000 r--p 00000000 fd:00 1337999 2ba51114f000-2ba511153000 r-xp 00000000 00:15 19647616 2ba511153000-2ba511352000 ---p 00004000 00:15 19647616 2ba511352000-2ba511353000 rw-p 00003000 00:15 19647616 2ba511353000-2ba51135a000 r-xp 00000000 00:15 103619381 2ba51135a000-2ba511559000 ---p 00007000 00:15 103619381 2ba511559000-2ba51155b000 rw-p 00006000 00:15 103619381 2ba51155b000-2ba511568000 r-xp 00000000 00:15 133225988 2ba511568000-2ba511767000 ---p 0000d000 00:15 133225988 2ba511767000-2ba511768000 rw-p 0000c000 00:15 133225988 2ba511768000-2ba51176a000 r-xp 00000000 00:15 16627297 2ba51176a000-2ba511969000 ---p 00002000 00:15 16627297 2ba511969000-2ba51196a000 rw-p 00001000 00:15 16627297 2ba51196a000-2ba5119b9000 r-xp 00000000 00:15 19550974 2ba5119b9000-2ba511bb9000 ---p 0004f000 00:15 19550974 2ba511bb9000-2ba511bba000 rw-p 0004f000 00:15 19550974 2ba511bba000-2ba511bc0000 r-xp 00000000 00:15 38262731 2ba511bc0000-2ba511dbf000 ---p 00006000 00:15 38262731 2ba511dbf000-2ba511dc0000 rw-p 00005000 00:15 38262731 2ba511dc0000-2ba511dc8000 r-xp 00000000 00:15 24110830 2ba511dc8000-2ba511fc7000 ---p 00008000 00:15 24110830 2ba511fc7000-2ba511fc8000 rw-p 00007000 00:15 24110830 2ba511fc8000-2ba511fd0000 r-xp 00000000 00:15 8079130 2ba511fd0000-2ba5121cf000 ---p 00008000 00:15 8079130 2ba5121cf000-2ba5121d0000 rw-p 00007000 00:15 8079130 2ba5121d0000-2ba512260000 r-xp 00000000 00:15 3604540 2ba512260000-2ba512460000 ---p 00090000 00:15 3604540 2ba512460000-2ba51246d000 rw-p 00090000 00:15 3604540 2ba51246d000-2ba512489000 r-xp 00000000 00:15 3604536 2ba512489000-2ba512688000 ---p 0001c000 00:15 3604536 2ba512688000-2ba51268c000 rw-p 0001b000 00:15 3604536 2ba512697000-2ba5126ad000 r-xp 00000000 fd:00 713186 2ba5126ad000-2ba5128ad000 ---p 00016000 fd:00 713186 2ba5128ad000-2ba5128ae000 r--p 00016000 fd:00 713186 2ba5128ae000-2ba5128af000 rw-p 00017000 fd:00 713186 2ba5128af000-2ba5128b3000 rw-p 2ba5128af000 2ba5128b3000-2ba5128c7000 r-xp 00000000 fd:00 713278 2ba5128c7000-2ba512ac6000 ---p 00014000 fd:00 713278 2ba512ac6000-2ba512ac7000 rw-p 00013000 fd:00 713278 2ba512ac7000-2ba512bf4000 r-xp 00000000 fd:00 713277 2ba512bf4000-2ba512df3000 ---p 0012d000 fd:00 713277 2ba512df3000-2ba512e14000 rw-p 0012c000 fd:00 713277 2ba512e14000-2ba512e18000 rw-p 2ba512e14000 2ba512e18000-2ba512e60000 r-xp 00000000 fd:00 713279 2ba512e60000-2ba513060000 ---p 00048000 fd:00 713279 2ba513060000-2ba513066000 rw-p 00048000 fd:00 713279 2ba513066000-2ba51306d000 r-xp 00000000 fd:00 713190 2ba51306d000-2ba51326d000 ---p 00007000 fd:00 713190 2ba51326d000-2ba51326e000 r--p 00007000 fd:00 713190 2ba51326e000-2ba51326f000 rw-p 00008000 fd:00 713190 2ba51326f000-2ba513355000 r-xp 00000000 fd:00 1333707 2ba513355000-2ba513554000 ---p 000e6000 fd:00 1333707 2ba513554000-2ba51355a000 r--p 000e5000 fd:00 1333707 2ba51355a000-2ba51355d000 rw-p 000eb000 fd:00 1333707 2ba51355d000-2ba51356f000 rw-p 2ba51355d000 2ba51356f000-2ba51357c000 r-xp 00000000 fd:00 713154 2ba51357c000-2ba51377c000 ---p 0000d000 fd:00 713154 2ba51377c000-2ba51377d000 rw-p 0000d000 fd:00 713154 2ba51377d000-2ba5137a9000 r-xp 00000000 fd:00 1334565 2ba5137a9000-2ba5139a9000 ---p 0002c000 fd:00 1334565 2ba5139a9000-2ba5139ab000 rw-p 0002c000 fd:00 1334565 2ba5139ab000-2ba513a3c000 r-xp 00000000 fd:00 1333319 2ba513a3c000-2ba513c3c000 ---p 00091000 fd:00 1333319 2ba513c3c000-2ba513c40000 rw-p 00091000 fd:00 1333319 2ba513c40000-2ba513c42000 r-xp 00000000 fd:00 713223 2ba513c42000-2ba513e41000 ---p 00002000 fd:00 713223 2ba513e41000-2ba513e42000 rw-p 00001000 fd:00 713223 2ba513e42000-2ba513e66000 r-xp 00000000 fd:00 1334799 2ba513e66000-2ba514065000 ---p 00024000 fd:00 1334799 2ba514065000-2ba514067000 rw-p 00023000 fd:00 1334799 2ba514067000-2ba51406f000 r-xp 00000000 fd:00 1335417 2ba51406f000-2ba51426e000 ---p 00008000 fd:00 1335417 2ba51426e000-2ba51426f000 rw-p 00007000 fd:00 1335417 2ba51426f000-2ba514271000 r-xp 00000000 fd:00 713373 2ba514271000-2ba514470000 ---p 00002000 fd:00 713373 2ba514470000-2ba514471000 rw-p 00001000 fd:00 713373 2ba514471000-2ba514482000 r-xp 00000000 fd:00 713188 2ba514482000-2ba514682000 ---p 00011000 fd:00 713188 2ba514682000-2ba514683000 r--p 00011000 fd:00 713188 2ba514683000-2ba514684000 rw-p 00012000 fd:00 713188 2ba514684000-2ba514686000 rw-p 2ba514684000 2ba514686000-2ba51469b000 r-xp 00000000 fd:00 713271 2ba51469b000-2ba51489b000 ---p 00015000 fd:00 713271 2ba51489b000-2ba51489d000 rw-p 00015000 fd:00 713271 2ba51489d000-2ba51489e000 rw-p 2ba51489d000 2ba51489e000-2ba5148d9000 r-xp 00000000 fd:00 713215 2ba5148d9000-2ba514ad9000 ---p 0003b000 fd:00 713215 2ba514ad9000-2ba514ada000 rw-p 0003b000 fd:00 713215 2ba514ada000-2ba514ae4000 rw-p 2ba514ada000 2ba514ae4000-2ba514ae7000 r-xp 00000000 00:15 47521075 2ba514ae7000-2ba514ce7000 ---p 00003000 00:15 47521075 2ba514ce7000-2ba514ce8000 rw-p 00003000 00:15 47521075 2ba514ce8000-2ba514cf9000 r-xp 00000000 00:15 133980523 2ba514cf9000-2ba514ef9000 ---p 00011000 00:15 133980523 2ba514ef9000-2ba514efc000 rw-p 00011000 00:15 133980523 2ba514efc000-2ba514f03000 r-xp 00000000 00:15 17896113 2ba514f03000-2ba515103000 ---p 00007000 00:15 17896113 2ba515103000-2ba515104000 rw-p 00007000 00:15 17896113 2ba515104000-2ba515105000 ---p 2ba515104000 2ba515105000-2ba515b05000 rw-p 2ba515105000 2ba515b05000-2ba515b06000 ---p 2ba515b05000 2ba515b06000-2ba516506000 rw-p 2ba515b06000 2ba516506000-2ba51650d000 r--s 00000000 fd:00 1394145 2ba51650d000-2ba51650e000 rw-p 2ba51650d000 2ba51650e000-2ba51650f000 ---p 2ba51650e000 2ba51650f000-2ba516f0f000 rw-p 2ba51650f000 2ba516f0f000-2ba516f10000 ---p 2ba516f0f000 2ba516f10000-2ba517910000 rw-p 2ba516f10000 2ba517910000-2ba517911000 ---p 2ba517910000 2ba517911000-2ba518311000 rw-p 2ba517911000 2ba518311000-2ba518312000 ---p 2ba518311000 2ba518312000-2ba518d12000 rw-p 2ba518312000 2ba518d12000-2ba518d13000 ---p 2ba518d12000 2ba518d13000-2ba519714000 rw-p 2ba518d13000 2ba51971e000-2ba519728000 r-xp 00000000 fd:00 713178 2ba519728000-2ba519927000 ---p 0000a000 fd:00 713178 2ba519927000-2ba519928000 r--p 00009000 fd:00 713178 2ba519928000-2ba519929000 rw-p 0000a000 fd:00 713178 2ba519934000-2ba519938000 r-xp 00000000 fd:00 713176 2ba519938000-2ba519b37000 ---p 00004000 fd:00 713176 2ba519b37000-2ba519b38000 r--p 00003000 fd:00 713176 2ba519b38000-2ba519b39000 rw-p 00004000 fd:00 713176 2ba51c000000-2ba51c021000 rw-p 2ba51c000000 2ba51c021000-2ba520000000 ---p 2ba51c021000 7fffd9756000-7fffd976b000 rw-p 7ffffffe9000 00:00 0 7fffd97fd000-7fffd9800000 r-xp 7fffd97fd000 00:00 0 ffffffffff600000-ffffffffffe00000 ---p 00000000 00:00 0 corruption (!prev): 0x0000000003048ed0 ***
corrupted double-linked list: 0x0000000003048ec0 ***
+0x190)[0x2ba519721bd0]
br> en-Librdkafka/lib/librdkafka.so.1[0x2ba512226ef4]
en-Librdkafka/lib/librdkafka.so.1[0x2ba5121f6204]
en-Librdkafka/lib/librdkafka.so.1[0x2ba512227ed0]
br> /software/x64/perl/5.16.3/bin/perl
/software/x64/perl/5.16.3/bin/perl
/lib64/ld-2.5.so
00:00 0
/lib64/ld-2.5.so
/lib64/ld-2.5.so
/software/x64/perl/5.16.3/lib/5.16.3/x86_64-linux/CORE/libperl.so
/software/x64/perl/5.16.3/lib/5.16.3/x86_64-linux/CORE/libperl.so
/software/x64/perl/5.16.3/lib/5.16.3/x86_64-linux/CORE/libperl.so
/lib64/libnsl-2.5.so
/lib64/libnsl-2.5.so
/lib64/libnsl-2.5.so
/lib64/libnsl-2.5.so
00:00 0
/lib64/libdl-2.5.so
/lib64/libdl-2.5.so
/lib64/libdl-2.5.so
/lib64/libdl-2.5.so
/lib64/libm-2.5.so
/lib64/libm-2.5.so
/lib64/libm-2.5.so
/lib64/libm-2.5.so
/lib64/libcrypt-2.5.so
/lib64/libcrypt-2.5.so
/lib64/libcrypt-2.5.so
/lib64/libcrypt-2.5.so
00:00 0
/lib64/libutil-2.5.so
/lib64/libutil-2.5.so
/lib64/libutil-2.5.so
/lib64/libutil-2.5.so
/lib64/libc-2.5.so
/lib64/libc-2.5.so
/lib64/libc-2.5.so
/lib64/libc-2.5.so
00:00 0
/usr/lib/locale/locale-archive
/software/x64/perl/5.16.3/lib/5.16.3/x86_64-linux/auto/IO/IO.so
/software/x64/perl/5.16.3/lib/5.16.3/x86_64-linux/auto/IO/IO.so
/software/x64/perl/5.16.3/lib/5.16.3/x86_64-linux/auto/IO/IO.so
/software/x64/perl/5.16.3/lib/5.16.3/x86_64-linux/auto/Socket/Socket.so
/software/x64/perl/5.16.3/lib/5.16.3/x86_64-linux/auto/Socket/Socket.so
/software/x64/perl/5.16.3/lib/5.16.3/x86_64-linux/auto/Socket/Socket.so
/app/shared/lib/cpan64/lib/perl5/x86_64-linux/auto/JSON/XS/XS.so
/app/shared/lib/cpan64/lib/perl5/x86_64-linux/auto/JSON/XS/XS.so
/app/shared/lib/cpan64/lib/perl5/x86_64-linux/auto/JSON/XS/XS.so
/software/x64/perl/5.16.3/lib/5.16.3/x86_64-linux/auto/attributes/attributes.so
/software/x64/perl/5.16.3/lib/5.16.3/x86_64-linux/auto/attributes/attributes.so
/software/x64/perl/5.16.3/lib/5.16.3/x86_64-linux/auto/attributes/attributes.so
/software/x64/perl/5.16.3/lib/5.16.3/x86_64-linux/auto/re/re.so
/software/x64/perl/5.16.3/lib/5.16.3/x86_64-linux/auto/re/re.so
/software/x64/perl/5.16.3/lib/5.16.3/x86_64-linux/auto/re/re.so
/software/x64/perl/5.16.3/lib/5.16.3/x86_64-linux/auto/List/Util/Util.so
/software/x64/perl/5.16.3/lib/5.16.3/x86_64-linux/auto/List/Util/Util.so
/software/x64/perl/5.16.3/lib/5.16.3/x86_64-linux/auto/List/Util/Util.so
/software/x64/perl/5.16.3/lib/5.16.3/x86_64-linux/auto/Encode/Encode.so
/software/x64/perl/5.16.3/lib/5.16.3/x86_64-linux/auto/Encode/Encode.so
/software/x64/perl/5.16.3/lib/5.16.3/x86_64-linux/auto/Encode/Encode.so
/app/shared/lib/cpan64/lib/perl5/x86_64-linux/auto/Kafka/Librd/Librd.so
/app/shared/lib/cpan64/lib/perl5/x86_64-linux/auto/Kafka/Librd/Librd.so
/app/shared/lib/cpan64/lib/perl5/x86_64-linux/auto/Kafka/Librd/Librd.so
/app/shared/lib/cpan64/lib/perl5/auto/share/dist/Alien-Librdkafka/lib/librdkafka.so.1
/app/shared/lib/cpan64/lib/perl5/auto/share/dist/Alien-Librdkafka/lib/librdkafka.so.1
/app/shared/lib/cpan64/lib/perl5/auto/share/dist/Alien-Librdkafka/lib/librdkafka.so.1
/app/shared/lib/cpan64/lib/perl5/auto/share/dist/Alien-Librdkafka/lib/librdkafka++.so.1
/app/shared/lib/cpan64/lib/perl5/auto/share/dist/Alien-Librdkafka/lib/librdkafka++.so.1
/app/shared/lib/cpan64/lib/perl5/auto/share/dist/Alien-Librdkafka/lib/librdkafka++.so.1
/lib64/libpthread-2.5.so
/lib64/libpthread-2.5.so
/lib64/libpthread-2.5.so
/lib64/libpthread-2.5.so
00:00 0
/lib64/libz.so.1.2.3
/lib64/libz.so.1.2.3
/lib64/libz.so.1.2.3
/lib64/libcrypto.so.0.9.8e
/lib64/libcrypto.so.0.9.8e
/lib64/libcrypto.so.0.9.8e
00:00 0
/lib64/libssl.so.0.9.8e
/lib64/libssl.so.0.9.8e
/lib64/libssl.so.0.9.8e
/lib64/librt-2.5.so
/lib64/librt-2.5.so
/lib64/librt-2.5.so
/lib64/librt-2.5.so
/usr/lib64/libstdc++.so.6.0.8
/usr/lib64/libstdc++.so.6.0.8
/usr/lib64/libstdc++.so.6.0.8
/usr/lib64/libstdc++.so.6.0.8
00:00 0
/lib64/libgcc_s-4.1.2-20080825.so.1
/lib64/libgcc_s-4.1.2-20080825.so.1
/lib64/libgcc_s-4.1.2-20080825.so.1
/usr/lib64/libgssapi_krb5.so.2.2
/usr/lib64/libgssapi_krb5.so.2.2
/usr/lib64/libgssapi_krb5.so.2.2
/usr/lib64/libkrb5.so.3.3
/usr/lib64/libkrb5.so.3.3
/usr/lib64/libkrb5.so.3.3
/lib64/libcom_err.so.2.1
/lib64/libcom_err.so.2.1
/lib64/libcom_err.so.2.1
/usr/lib64/libk5crypto.so.3.1
/usr/lib64/libk5crypto.so.3.1
/usr/lib64/libk5crypto.so.3.1
/usr/lib64/libkrb5support.so.0.1
/usr/lib64/libkrb5support.so.0.1
/usr/lib64/libkrb5support.so.0.1
/lib64/libkeyutils-1.2.so
/lib64/libkeyutils-1.2.so
/lib64/libkeyutils-1.2.so
/lib64/libresolv-2.5.so
/lib64/libresolv-2.5.so
/lib64/libresolv-2.5.so
/lib64/libresolv-2.5.so
00:00 0
/lib64/libselinux.so.1
/lib64/libselinux.so.1
/lib64/libselinux.so.1
00:00 0
/lib64/libsepol.so.1
/lib64/libsepol.so.1
/lib64/libsepol.so.1
00:00 0
/software/x64/perl/5.16.3/lib/5.16.3/x86_64-linux/auto/Fcntl/Fcntl.so
/software/x64/perl/5.16.3/lib/5.16.3/x86_64-linux/auto/Fcntl/Fcntl.so
/software/x64/perl/5.16.3/lib/5.16.3/x86_64-linux/auto/Fcntl/Fcntl.so
/software/x64/perl/5.16.3/lib/5.16.3/x86_64-linux/auto/POSIX/POSIX.so
/software/x64/perl/5.16.3/lib/5.16.3/x86_64-linux/auto/POSIX/POSIX.so
/software/x64/perl/5.16.3/lib/5.16.3/x86_64-linux/auto/POSIX/POSIX.so
/software/x64/perl/5.16.3/lib/5.16.3/x86_64-linux/auto/Data/Dumper/Dumper.so
/software/x64/perl/5.16.3/lib/5.16.3/x86_64-linux/auto/Data/Dumper/Dumper.so
/software/x64/perl/5.16.3/lib/5.16.3/x86_64-linux/auto/Data/Dumper/Dumper.so
00:00 0
00:00 0
00:00 0
00:00 0
/usr/lib64/gconv/gconv-modules.cache
00:00 0
00:00 0
00:00 0
00:00 0
00:00 0
00:00 0
00:00 0
00:00 0
00:00 0
00:00 0
00:00 0
/lib64/libnss_files-2.5.so
/lib64/libnss_files-2.5.so
/lib64/libnss_files-2.5.so
/lib64/libnss_files-2.5.so
/lib64/libnss_dns-2.5.so
/lib64/libnss_dns-2.5.so
/lib64/libnss_dns-2.5.so
/lib64/libnss_dns-2.5.so
00:00 0
00:00 0
[stack]
[vdso]
[vsyscall]

Feature Request: Implement rd_kafka_message_timestamp

Hi Pavel,

Looked into it, landed here confluentinc/librdkafka#1016 and it seems it came somewhere 0.10 into Kafka. What is you take on adding this into this module ?

PS: thumbs up for this module by the way. It just works, thats it.

How to use $key for internal Kafka partitioning logic on producer

I'm not a pro at C but it appears that we can pass a -1 as the partition and then the $key will go through the crc32 logic for automatic partitioning. Is that correct?

If so, can that be documented. Also supporting undef() would make for a better perl-ish API.

$err = $topic->produce($partition, $msgflags, $payload, $key);
$err = $topic->produce(-1, $msgflags, $payload, $key); #auto partition based on $key
$err = $topic->produce(undef(), $msgflags, $payload, $key); #future auto partition based on $key

Can't build on thread-enabled perl.

Hi,
I tried to install Kafka::Librd 0.05 on linux perl-5.20.3( useithreads=define ).
but I got the following error messages.

   ZWON/Kafka-Librd-0.05.tar.gz
  /home/aero/perl5/perlbrew/perls/perl-5.20/bin/perl Makefile.PL -- OK
Running make for Z/ZW/ZWON/Kafka-Librd-0.05.tar.gz
cp lib/Kafka/Librd.pm blib/lib/Kafka/Librd.pm
Running Mkbootstrap for Librd ()
chmod 644 "Librd.bs"
"/home/aero/perl5/perlbrew/perls/perl-5.20/bin/perl" -MExtUtils::Command::MM -e 'cp_nonempty' -- Librd.bs blib/arch/auto/Kafka/Librd/Librd.bs 644
/home/aero/perl5/perlbrew/perls/perl-5.20/bin/perl utils/generate_const.pl
"/home/aero/perl5/perlbrew/perls/perl-5.20/bin/perl" "/home/aero/perl5/perlbrew/perls/perl-5.20/lib/site_perl/5.20.3/ExtUtils/xsubpp"  -typemap '/home/aero/perl5/perlbrew/perls/perl-5.20/lib/5.20.3/ExtUtils/typemap' -typemap '/home/aero/.cpan/build/Kafka-Librd-0.05-ALFgeR/typemap'  Rdkafka.xs > Rdkafka.xsc
mv Rdkafka.xsc Rdkafka.c
cc -c   -I/home/aero/perl5/perlbrew/perls/perl-5.20/lib/site_perl/5.20.3/auto/share/dist/Alien-Librdkafka/include -I/home/aero/perl5/perlbrew/perls/perl-5.20/lib/site_perl/5.20.3/auto/share/dist/Alien-Librdkafka/include -O2   -DVERSION=\"0.05\" -DXS_VERSION=\"0.05\" -fPIC "-I/home/aero/perl5/perlbrew/perls/perl-5.20/lib/5.20.3/x86_64-linux-thread-multi/CORE"   Rdkafka.c
cc -c   -I/home/aero/perl5/perlbrew/perls/perl-5.20/lib/site_perl/5.20.3/auto/share/dist/Alien-Librdkafka/include -I/home/aero/perl5/perlbrew/perls/perl-5.20/lib/site_perl/5.20.3/auto/share/dist/Alien-Librdkafka/include -O2   -DVERSION=\"0.05\" -DXS_VERSION=\"0.05\" -fPIC "-I/home/aero/perl5/perlbrew/perls/perl-5.20/lib/5.20.3/x86_64-linux-thread-multi/CORE"   rdkafkaxs.c
In file included from rdkafkaxs.h:3:0,
                 from rdkafkaxs.c:1:
/home/aero/perl5/perlbrew/perls/perl-5.20/lib/5.20.3/x86_64-linux-thread-multi/CORE/perl.h:2396:22: error: unknown type name ‘off64_t’
 #       define Off_t off64_t
                      ^
/home/aero/perl5/perlbrew/perls/perl-5.20/lib/5.20.3/x86_64-linux-thread-multi/CORE/perlio.h:277:15: note: in expansion of macro ‘Off_t’
 PERL_EXPORT_C Off_t PerlIO_tell(PerlIO *);
               ^
/home/aero/perl5/perlbrew/perls/perl-5.20/lib/5.20.3/x86_64-linux-thread-multi/CORE/perl.h:2396:22: error: unknown type name ‘off64_t’
 #       define Off_t off64_t
                      ^
/home/aero/perl5/perlbrew/perls/perl-5.20/lib/5.20.3/x86_64-linux-thread-multi/CORE/perlio.h:280:41: note: in expansion of macro ‘Off_t’
 PERL_EXPORT_C int PerlIO_seek(PerlIO *, Off_t, int);
                                         ^
In file included from rdkafkaxs.h:3:0,
                 from rdkafkaxs.c:1:
/home/aero/perl5/perlbrew/perls/perl-5.20/lib/5.20.3/x86_64-linux-thread-multi/CORE/perl.h:2396:22: error: unknown type name ‘off64_t’
 #       define Off_t off64_t
                      ^
/home/aero/perl5/perlbrew/perls/perl-5.20/lib/5.20.3/x86_64-linux-thread-multi/CORE/proto.h:975:47: note: in expansion of macro ‘Off_t’
 PERL_CALLCONV bool Perl_do_seek(pTHX_ GV* gv, Off_t pos, int whence);
                                               ^
/home/aero/perl5/perlbrew/perls/perl-5.20/lib/5.20.3/x86_64-linux-thread-multi/CORE/perl.h:2396:22: error: unknown type name ‘off64_t’
 #       define Off_t off64_t
                      ^
/home/aero/perl5/perlbrew/perls/perl-5.20/lib/5.20.3/x86_64-linux-thread-multi/CORE/proto.h:987:15: note: in expansion of macro ‘Off_t’
 PERL_CALLCONV Off_t Perl_do_sysseek(pTHX_ GV* gv, Off_t pos, int whence)
               ^
/home/aero/perl5/perlbrew/perls/perl-5.20/lib/5.20.3/x86_64-linux-thread-multi/CORE/perl.h:2396:22: error: unknown type name ‘off64_t’
 #       define Off_t off64_t
                      ^
/home/aero/perl5/perlbrew/perls/perl-5.20/lib/5.20.3/x86_64-linux-thread-multi/CORE/proto.h:987:51: note: in expansion of macro ‘Off_t’
 PERL_CALLCONV Off_t Perl_do_sysseek(pTHX_ GV* gv, Off_t pos, int whence)
                                                   ^
/home/aero/perl5/perlbrew/perls/perl-5.20/lib/5.20.3/x86_64-linux-thread-multi/CORE/perl.h:2396:22: error: unknown type name ‘off64_t’
 #       define Off_t off64_t
                      ^
/home/aero/perl5/perlbrew/perls/perl-5.20/lib/5.20.3/x86_64-linux-thread-multi/CORE/proto.h:992:15: note: in expansion of macro ‘Off_t’
 PERL_CALLCONV Off_t Perl_do_tell(pTHX_ GV* gv)
               ^
/home/aero/perl5/perlbrew/perls/perl-5.20/lib/5.20.3/x86_64-linux-thread-multi/CORE/perl.h:2396:22: error: unknown type name ‘off64_t’
 #       define Off_t off64_t
                      ^
/home/aero/perl5/perlbrew/perls/perl-5.20/lib/5.20.3/x86_64-linux-thread-multi/CORE/proto.h:8073:53: note: in expansion of macro ‘Off_t’
 PERL_CALLCONV int Perl_PerlIO_seek(pTHX_ PerlIO *f, Off_t offset, int whence);
                                                     ^
/home/aero/perl5/perlbrew/perls/perl-5.20/lib/5.20.3/x86_64-linux-thread-multi/CORE/perl.h:2396:22: error: unknown type name ‘off64_t’
 #       define Off_t off64_t
                      ^
/home/aero/perl5/perlbrew/perls/perl-5.20/lib/5.20.3/x86_64-linux-thread-multi/CORE/proto.h:8086:15: note: in expansion of macro ‘Off_t’
 PERL_CALLCONV Off_t Perl_PerlIO_tell(pTHX_ PerlIO *f);
               ^
In file included from rdkafkaxs.h:3:0,
                 from rdkafkaxs.c:1:
rdkafkaxs.c: In function ‘krd_parse_config’:
/home/aero/perl5/perlbrew/perls/perl-5.20/lib/5.20.3/x86_64-linux-thread-multi/CORE/perl.h:155:16: error: ‘my_perl’ undeclared (first use in this function)
 #  define aTHX my_perl
                ^
/home/aero/perl5/perlbrew/perls/perl-5.20/lib/5.20.3/x86_64-linux-thread-multi/CORE/perl.h:168:18: note: in expansion of macro ‘aTHX’
 #  define aTHX_  aTHX,
                  ^
/home/aero/perl5/perlbrew/perls/perl-5.20/lib/5.20.3/x86_64-linux-thread-multi/CORE/embed.h:217:42: note: in expansion of macro ‘aTHX_’
 #define hv_iterinit(a)  Perl_hv_iterinit(aTHX_ a)
                                          ^
rdkafkaxs.c:13:5: note: in expansion of macro ‘hv_iterinit’
     hv_iterinit(params);
     ^
/home/aero/perl5/perlbrew/perls/perl-5.20/lib/5.20.3/x86_64-linux-thread-multi/CORE/perl.h:155:16: note: each undeclared identifier is reported only once for each function it appears in
 #  define aTHX my_perl
                ^
/home/aero/perl5/perlbrew/perls/perl-5.20/lib/5.20.3/x86_64-linux-thread-multi/CORE/perl.h:168:18: note: in expansion of macro ‘aTHX’
 #  define aTHX_  aTHX,
                  ^
/home/aero/perl5/perlbrew/perls/perl-5.20/lib/5.20.3/x86_64-linux-thread-multi/CORE/embed.h:217:42: note: in expansion of macro ‘aTHX_’
 #define hv_iterinit(a)  Perl_hv_iterinit(aTHX_ a)
                                          ^
rdkafkaxs.c:13:5: note: in expansion of macro ‘hv_iterinit’
     hv_iterinit(params);
     ^
rdkafkaxs.c: In function ‘krd_parse_topic_config’:
/home/aero/perl5/perlbrew/perls/perl-5.20/lib/5.20.3/x86_64-linux-thread-multi/CORE/perl.h:155:16: error: ‘my_perl’ undeclared (first use in this function)
 #  define aTHX my_perl
                ^
/home/aero/perl5/perlbrew/perls/perl-5.20/lib/5.20.3/x86_64-linux-thread-multi/CORE/perl.h:168:18: note: in expansion of macro ‘aTHX’
 #  define aTHX_  aTHX,
                  ^
/home/aero/perl5/perlbrew/perls/perl-5.20/lib/5.20.3/x86_64-linux-thread-multi/CORE/embed.h:217:42: note: in expansion of macro ‘aTHX_’
 #define hv_iterinit(a)  Perl_hv_iterinit(aTHX_ a)
                                          ^
rdkafkaxs.c:53:5: note: in expansion of macro ‘hv_iterinit’
     hv_iterinit(params);
     ^
make: *** [rdkafkaxs.o] Error 1

So I also tried to installl Kafka::LIbrd on linux perl-5.20.3 ( useithreads=undef ).
and then I could install without problem.

I searched the similar error message cases on the internet.

http://stackoverflow.com/questions/10469821/why-a-not-threaded-perl-is-not-using-off64-t-type-compared-to-a-threads-enabled
xslate/p5-Mouse#53
http://blogs.perl.org/users/nick_wellnhofer/2015/03/writing-xs-like-a-pro---perl-no-get-context-and-static-functions.html

I guess somthing in Kafka::Librd is not compatible with thread-enabled perl.

Thanks in advance.

Undeclared dependency on Test::Most

Hi Pavel, as per subject:)

Sample fail report:

http://www.cpantesters.org/cpan/report/d0fbd032-986e-11e6-89f7-2b35f9c3459a

Greets,

Need access to methods to get metadata about brokers and groups

rd_kafka_list_groups
rd_kafka_metadata

Ability to seek in topic

I'd like to be able to seek to an offset within the topic. At the moment, it's useful when testing a Kafka consumer, as it makes it easier to iterate over the same messages with different versions of the consumer.

I believe this implies calling rd_kafka_seek, which doesn't seem to be exposed by Kafka::Librd.

Compilation failure on debian/jessie

On debian/jessie systems with installed package librdkafka-dev (version 0.8.5-2) the compilation fails:

cc -c   -fwrapv -fno-strict-aliasing -pipe -fstack-protector-strong -I/usr/local/include -D_LARGEFILE_SOURCE -D_FILE_OFFSET_BITS=64 -D_FORTIFY_SOURCE=2 -O2   -DVERSION=\"0.01\" -DXS_VERSION=\"0.01\" -fPIC "-I/opt/perl-5.22.2/lib/5.22.2/x86_64-linux/CORE"   Rdkafka.c
Rdkafka.xs: In function 'XS_Kafka__Librd_subscribe':
Rdkafka.xs:64:9: error: unknown type name 'rd_kafka_topic_partition_list_t'
         rd_kafka_topic_partition_list_t* topic_list;
         ^
...
Rdkafka.xs: In function 'XS_Kafka__Librd__Error_rd_kafka_get_err_descs':
Rdkafka.xs:198:13: error: invalid use of undefined type 'struct rd_kafka_err_desc'
             if (descs[i].name != NULL) {
             ^
...