The London Perl and Raku Workshop takes place on 26th Oct 2024. If your company depends on Perl, please consider sponsoring and/or attending.
 / 
Kafka-0.800_3
River stage one • 2 direct dependents • 2 total dependents
11 ++
/ Kafka::Connection

NAME

Kafka::Connection - object interface to connect to a kafka cluster.

VERSION

This documentation refers to Kafka::Connection version 0.800_3 .

SYNOPSIS

    use 5.010;
    use strict;
    use warnings;

    # A simple example of Kafka::Connection usage:
    use Kafka::Connection;

    # connect to local cluster with the defaults
    my $connect = Kafka::Connection->new( host => 'localhost' );

    # decoding of the error
    say STDERR 'last error: ', $connect->last_error
        unless $connect->last_errorcode;

    # Closes the connection and cleans up
    undef $connect;

DESCRIPTION

The main features of the Kafka::Connection class are:

  • Provides API for communication with Kafka 0.8 cluster.

  • Coding and decoding of requests and responses, auto-selection of a server from Kafka cluster. =item *

    Allows for getting information about Kafka cluster.

CONSTRUCTOR

new

Creates Kafka::Connection object for interaction with Kafka cluster. Returns created Kafka::Connection object.

Depending on the value of RaiseError attribute, an error causes program to halt or the constructor returns Kafka::Connection object.

Use methods "last_errorcode" and "last_error" of the Kafka::Connection object to get information about the error.

new() takes arguments in key-value pairs. The following arguments are currently recognized:

host => $host

$host is an any Apache Kafka cluster host to connect to. It can be a hostname or the IP-address in the "xx.xx.xx.xx" form.

Optional. Either host or broker_list must be supplied.

port => $port

Optional, default = $KAFKA_SERVER_PORT.

$port is the attribute denoting the port number of the service we want to access (Apache Kafka service). $port should be an integer number.

$KAFKA_SERVER_PORT is the default Apache Kafka server port that can be imported from the Kafka module and = 9092.

broker_list => $broker_list

Optional, $broker_list is a reference to array of the host:port strings, defining the list of Kafka servers. This list will be used to locate the new master in case server specified via host => $host and port => $port arguments is unavailable. Either host or broker_list must be supplied.

timeout => $timeout

Optional, default = $REQUEST_TIMEOUT.

$timeout specifies how long we wait for the remote server to respond before IO object disconnects and creates an internal exception. $timeout is in second, could be positive integer or floating-point type.

$REQUEST_TIMEOUT is the default timeout that can be imported from the Kafka module.

RaiseError => $mode

Optional, default = 0.

An error will cause the program to halt if "RaiseError" is set to true: confess if the argument is not valid or die in the other error case (this can always be trapped with eval).

You should always check for errors, when not establishing the RaiseError mode to true.

CorrelationId => $correlation_id

Optional, default = undef .

Correlation is a user-supplied integer. It will be passed back with the response by the server, unmodified. The $correlation_id should be an integer number.

An error will be thrown if CorrelationId from response will not match one supplied in request.

If CorrelationId is not set, its value is assigned as random negative integer.

SEND_MAX_RETRIES => $retries

Optional, default = $SEND_MAX_RETRIES .

$SEND_MAX_RETRIES is the default number of retries that can be imported from the Kafka module and = 3 .

The leader may be unavailable transiently, which can fail the sending of a message. This property specifies the number of retries when such failures occur. The $retries should be an integer number.

RETRY_BACKOFF => $backoff

Optional, default = $RETRY_BACKOFF .

$RETRY_BACKOFF is the default timeout that can be imported from the Kafka module and = 100 ms.

This property specifies ms before each retry, the producer refreshes the metadata of relevant topics. Since leader election takes a bit of time, this property specifies the amount of time that the producer waits before refreshing the metadata. The $backoff should be an integer number.

METHODS

The following methods are defined for the Kafka::Producer class:

get_known_servers

Returns the list of known Kafka servers (in host:port format).

is_server_known( $server )

Returns true, if $server (host:port) is known in cluster.

is_server_alive( $server )

Returns true, if successful connection is established with $server (host:port).

receive_response_to_request( $request )

$request is a reference to the hash representing the structure of the request.

This method encodes $request, pass it to the leader of cluster, receives reply and returns it in a form of hash reference.

WARNING:

  • This method is not designed to be use by the end user.

  • In order to achieve better performance, this method do not perform arguments validation.

close_connection( $server )

Closes connection with $server (defined as host:port).

close

Closes connection with all known Kafka servers.

RaiseError

This method returns current value showing how errors are handled within Kafka module. If set to true, die() is dispatched when error during communication is detected.

last_errorcode and last_error are diagnostic methods and can be used to get detailed error codes and messages for various cases: when server or the resource is not available, access to the resource was denied, etc.

last_errorcode

Returns code of the last error.

last_error

Returns an error message that contains information about the encountered failure.

DIAGNOSTICS

Review documentation of the "RaiseError" method for additional information about possible errors.

It's advised to always check "last_errorcode" and more descriptive "last_error" when "RaiseError" is not set.

Invalid argument

Invalid argument was provided to new constructor or to other method.

Can't send

Message can't be sent to Kafka.

Can't recv

Message can't be received from Kafka.

Can't bind

A successful TCP connection can't be established on given host and port.

Can't get metadata

Error detected during parsing of response from Kafka.

Leader not found

Failed to locate leader of Kafka cluster.

Mismatch CorrelationId

Mismatch of CorrelationId of request and response.

There are no known brokers

Failed to locate cluster broker.

Can't get metadata

Received meta data is incorrect or missing.

Use "last_error" method from Kafka::Connection object to obtain detailed description of an error.

SEE ALSO

The basic operation of the Kafka package modules:

Kafka - constants and messages used by the Kafka package modules.

Kafka::Connection - interface to connect to a Kafka cluster.

Kafka::Producer - interface for producing client.

Kafka::Consumer - interface for consuming client.

Kafka::Message - interface to access Kafka message properties.

Kafka::Int64 - functions to work with 64 bit elements of the protocol on 32 bit systems.

Kafka::Protocol - functions to process messages in the Apache Kafka's Protocol.

Kafka::IO - low level interface for communication with Kafka server.

Kafka::Internals - Internal constants and functions used by several package modules.

A wealth of detail about the Apache Kafka and the Kafka Protocol:

Main page at http://kafka.apache.org/

Kafka Protocol at https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol

AUTHOR

Sergey Gladkov, <sgladkov@trackingsoft.com>

CONTRIBUTORS

Alexander Solovey

Jeremy Jordan

Vlad Marchenko

COPYRIGHT AND LICENSE

Copyright (C) 2012-2013 by TrackingSoft LLC.

This package is free software; you can redistribute it and/or modify it under the same terms as Perl itself. See perlartistic at http://dev.perl.org/licenses/artistic.html.

This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.