NAME

Kafka::Consumer::Avro - Avro message consumer for Apache Kafka.

SYNOPSIS

  use Kafka qw/DEFAULT_MAX_BYTES/;
  use Kafka::Connection;
  use Kafka::Consumer::Avro;
  use Confluent::SchemaRegistry;
  
  my $connection = Kafka::Connection->new( host => 'localhost' );
  
  my $consumer = Kafka::Consumer::Avro->new( Connection => $connection , SchemaRegistry => Confluent::SchemaRegistry->new() );
  
  # Consuming messages
  my $messages = $consumer->fetch(
        'mytopic',            # topic
        0,                    # partition
        0,                    # offset
        $DEFAULT_MAX_BYTES    # Maximum size of MESSAGE(s) to receive
  );
  
  if ($messages) {
        foreach my $message (@$messages) {
                if ( $message->valid ) {
                        say 'payload    : ', $message->payload;
                        say 'key        : ', $message->key;
                        say 'offset     : ', $message->offset;
                        say 'next_offset: ', $message->next_offset;
                }
                else {
                        say 'error      : ', $message->error;
                }
        }
  }
  
  # Closes the consumer and cleans up
  undef $consumer;
  $connection->close;
  undef $connection;

DESCRIPTION

Kafka::Consumer::Avro main feature is to provide object-oriented API to consume messages according to Confluent SchemaRegistry and Avro serialization.

Kafka::Consumer::Avro inerhits from and extends Kafka::Consumer.

INSTALL

Installation of Kafka::Consumer::Avro is a canonical:

  perl Makefile.PL
  make
  make test
  make install

TEST NOTES

Tests are focused on verifying Avro-formatted messages and theirs interactions with Confluent Schema Registry and are intended to extend Kafka::Consumer test suite.

They expect that in the target machine are available Kafka and Schema Registry listening on localhost and default ports, otherwise most of the test are skipped.

USAGE

CONSTRUCTOR

new

Creates new consumer client object.

new() takes arguments in key-value pairs as described in Kafka::Consumer from which it inherits.

In addition, takes in the following arguments:

SchemaRegistry => $schema_registry (mandatory)

Is a Confluent::SchemaRegistry instance.

METHODS

The following methods are defined for the Kafka::Avro::Consumer class:

schema_registry()

Returns the Confluent::SchemaRegistry instance supplied to the construcor.

get_error()

Returns a string containing last error message.

fetch( %params )

Gets messages froma a Kafka topic.

Please, see Kafka::Consumer for more details.

AUTHOR

Alvaro Livraghi, <alvarol@cpan.org>

CONTRIBUTE

https://github.com/alivraghi/Kafka-Consumer-Avro

BUGS

Please use GitHub project link above to report problems or contact authors.

COPYRIGHT AND LICENSE

Copyright 2018 by Alvaro Livraghi

This program is free software; you can redistribute it and/or modify it under the same terms as Perl itself.

cpanm Kafka::Consumer::Avro

perl -MCPAN -e shell
install Kafka::Consumer::Avro