view release on metacpan or  search on metacpan

README  view on Meta::CPAN

  client_id($id)
    Set the client identifier.

  tls($enable, [$ca_file, $skip_verify])
    Configure TLS.

  sasl($mechanism, [$username, $password])
    Configure SASL authentication.

  auto_reconnect($enable, [$delay_ms])
    Enable automatic reconnection. $delay_ms (default 1000) is the base
    delay: the first retry happens after exactly $delay_ms, later retries
    use capped exponential backoff (doubling each attempt, up to 30s, with
    +/-25% jitter), resetting once a connection succeeds. Reconnects reuse
    the timeout from the original connect() call.

  leave_group($group_id, $member_id, $cb)
    Send LeaveGroup to coordinator for fast partition rebalance.

  create_topics(\@topics, $timeout_ms, $cb)
    Create topics. Each element: "{name, num_partitions,
    replication_factor}".

        $conn->create_topics(
            [{ name => 'new-topic', num_partitions => 3, replication_factor => 1 }],
            5000, sub { my ($res, $err) = @_ }
        );

  delete_topics(\@topic_names, $timeout_ms, $cb)
    Delete topics by name.

  init_producer_id($transactional_id, $txn_timeout_ms, $cb)
    Initialize a producer ID for idempotent/transactional produce. Pass
    "undef" for non-transactional idempotent producer.

  add_partitions_to_txn($txn_id, $producer_id, $epoch, \@topics, $cb)
    Register partitions with the transaction coordinator.

  end_txn($txn_id, $producer_id, $epoch, $committed, $cb)
    Commit ("$committed=1") or abort ("$committed=0") a transaction.

  txn_offset_commit($txn_id, $group_id, $producer_id, $epoch, $generation, $member_id, \@offsets, $cb)
    Commit consumer offsets within a transaction (API 28).

  pending
    Number of requests awaiting broker response.

  state
    Connection state as integer (0=disconnected, 6=ready).

  Object lifetime
    The connection is torn down when its last reference drops, or when
    "DESTROY" is called explicitly; explicit destruction is idempotent.
    Pending request callbacks are invoked with a 'destroyed' error during
    teardown. Once destroyed, the object is inert: every subsequent method
    call on it (from any copy of the reference) croaks with
    "EV::Kafka::Conn: method called on destroyed connection" -- including
    calls made from callbacks running during the teardown itself (those
    exceptions are caught by the callback dispatcher and reported via
    warn()). A conn passed a custom "EV::Loop" holds a reference on it, so
    the loop cannot be destroyed while the conn is alive.

UTILITY FUNCTIONS
  EV::Kafka::_murmur2($key)
    Kafka-compatible murmur2 hash. Returns a non-negative 31-bit integer.

  EV::Kafka::_crc32c($data)
    CRC32C checksum (Castagnoli). Used internally for RecordBatch integrity.

  EV::Kafka::_error_name($code)
    Convert Kafka error code to string name.

RESULT STRUCTURES
  Produce result
        $result = {
            topics => [{
                topic      => 'name',
                partitions => [{
                    partition   => 0,
                    error_code  => 0,
                    base_offset => 42,
                }],
            }],
        };

  Fetch result
        $result = {
            topics => [{
                topic      => 'name',
                partitions => [{
                    partition      => 0,
                    error_code     => 0,
                    high_watermark => 100,
                    records => [{
                        offset    => 42,
                        timestamp => 1712345678000,
                        key       => 'key',      # or undef
                        value     => 'value',     # or undef
                        headers   => { h => 'v' },  # if present
                    }],
                }],
            }],
        };

  Metadata result
        $result = {
            controller_id => 0,
            brokers => [{ node_id => 0, host => '10.0.0.1', port => 9092 }],
            topics  => [{
                name       => 'topic',
                error_code => 0,
                partitions => [{
                    partition  => 0,
                    leader     => 0,
                    error_code => 0,
                }],
            }],
        };

ERROR HANDLING
    Errors are delivered through two channels: