NAME

Kafka::Producer::Avro - Avro message producer for Apache Kafka.

SYNOPSIS

  use Kafka::Connection;
  use Kafka::Producer::Avro;
  
  my $connection = Kafka::Connection->new( host => 'localhost' );
  
  my $producer = Kafka::Producer::Avro->new( Connection => $connection , SchemaRegistry => Confluent::SchemaRegistry->new() );
  
  # Set Avro schema for message key (valid JSON-string)
  my $key_schema = <<KEY_SCHEMA;
        {
                "type": "long",
                "name": "_id"
        }
  KEY_SCHEMA
  # Set Avro schema for message value (payload) (valid JSON-string)
  my $value_schema = <<VALUE_SCHEMA;
        {
                "type": "record",
                "name": "myrecord",
                "fields": [
                        {
                                "name": "f1",
                                "type": "string"
                        }
                ]
        }
  VALUE_SCHEMA
  
  # Sending a single message
  my $response = $producer->send(
        'mytopic',          # topic
        0,                  # partition
        'Single message',   # message
        undef,              # key
        undef,              # compression_codec
        undef,              # timestamps
        $key_schema,        # key_schema
        $value_schema       # value_schema
  );
  
  # Sending a series of messages
  $response = $producer->send(
        'mytopic',          # topic
        0,                  # partition
        [                   # messages
                'The first message',
                'The second message',
                'The third message',
        ],
        undef,              # key(s)
        undef,              # compression_codec
        undef,              # timestamp(s)
        $key_schema,        # key_schema
        $value_schema       # value_schema
  );
  
  # ...or use named parameters
  
  $producer->send(
        topic             => $topic,
        partition         => $partition,
        messages          => $messages,
        keys              => $keys,
        compression_codec => $compression_codec,
        timestamps        => $timestamps,
        key_schema        => $key_schema,
        value_schema      => $value_schema
  );    
  
  # Closes the producer and cleans up
  undef $producer;
  $connection->close;
  undef $connection;

DESCRIPTION

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

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

INSTALL

Installation of Kafka::Producer::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::Producer test suite.

They expect that in the target are listening Apache Kafka and Schema Registry services, respectively listening on localhost:9092 and http://localhost:8081.

You can alternatively set a different URLs by exporting the following environment variable:

KAFKA_HOST

KAFKA_PORT

CONFLUENT_SCHEMA_REGISTY_URL

For example:

  export KAFKA_HOST=my-kafka-host.my-domain.org
  export FALFA_PORT=9092
  export CONFLUENT_SCHEMA_REGISTY_URL=http://my-schema-registry-host.my-domain.org

USAGE

CONSTRUCTOR

new

Creates new producer client object.

new() takes arguments in key-value pairs as described in Kafka::Producer 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::Producer class:

schema_registry()

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

get_error()

Returns a string containing last error message.

send( $topic, $partition, $messages, $keys, $compression_codec, $timestamps, $key_schema, $value_schema )

send( %named_params )

Sends Avro-formatted messages on a Kafka::Connection object.

Returns a non-blank value (a reference to a hash with server response description) if the message is successfully sent.

In order to handle Avro format, Kafka::Producer|Kafka::Producer send() method is extended with two more positional arguments, $key_schema and $value_schema:

  $producer->send(
        $topic,             # scalar 
        $partition,         # scalar
        $messages,          # scalar | array
        $keys,              # (optional) undef | scalar | array
        $compression_codec, # (optional) undef | scalar
        $timestamps,        # (optional) undef | scalar | array
        $key_schema,        # (optional) undef | JSON-string
        $value_schema       # (optional) undef | JSON-string
  );

Both $key_schema and $value_schema parameters are optional and must provide JSON strings that represent Avro schemas to use to validate and serialize key(s) and value(s).

These schemas are validated against $schema_registry and, if compliant, they are added to the registry under the $topic+'key' or $topic+'value' Schema Registry's subjects.

If an expected schema isn't provided, latest version from Schema Registry is used accordingly to the (topic + key/value) subject.

Alternatively, for ease of use, the send() method may be also used by suggesting named parameters:

  $producer->send(
        topic             => $topic,             # scalar 
        partition         => $partition,         # scalar
        messages          => $messages,          # scalar | array
        keys              => $keys,              # (optional) undef | scalar | array
        compression_codec => $compression_codec, # (optional) undef | scalar
        timestamps        => $timestamps,        # (optional) undef | scalar | array
        key_schema        => $key_schema,        # (optional) undef | JSON-string
        value_schema      => $value_schema       # (optional) undef | JSON-string
  );    

bulk_send( %params )

Similar to send but uses bulks to avoid memory leaking.

Extra named parameters are expected:

size => $size

The size of the bulk

on_before_send_bulk => sub {...} (optional)

A code block that will be executed before the sending of each bulk.

The block will receive the following positional parameters:

$bulk_num the number of the bulk

$bulk_messages the number of messages in the bulk

$bulk_keys the number of keys in the bulk

$index_from the absolute index of the first message in the bulk

$index_to the absolute index of the last message in the bulk

on_after_send_bulk => sub {...} (optional)

A code block that will be executed after the sending of each bulk.

The block will receive the following positional parameters:

$sent the number of sent messages in the bulk

$total_sent the total number of messages sent

on_init => sub {...} (optional)

A code block that will be executed only once before at the beginning of the cycle.

The block will receive the following positional parameters:

$to_send the total number of messages to send

$bulk_size the size of the bulk

on_complete => sub {...} (optional)

A code block that will be executed only once after the end of the cycle.

The block will receive the following positional parameters:

$to_send the total number of messages to send

$total_sent the total number of messages sent

$errors the number bulks sent with errors

on_send_error => sub {...} (optional)

A code block that will be executed when a bulk registers an error.

AUTHOR

Alvaro Livraghi, <alvarol@cpan.org>

CONTRIBUTE

https://github.com/alivraghi/Kafka-Producer-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::Producer::Avro

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