view release on metacpan or  search on metacpan

lib/EV/Kafka.pm  view on Meta::CPAN

    if ($cfg->{sasl}) {
        $conn->sasl($cfg->{sasl}{mechanism}, $cfg->{sasl}{username}, $cfg->{sasl}{password});
    }
    $self->_wire_conn_handlers($conn);
}

# Error/disconnect reporting for client-managed conns. An
# auto_reconnect conn reports its loss ONCE via on_disconnect; reconnect
# attempts stay silent so on_error (default: die) doesn't storm.
sub _wire_conn_handlers {
    my ($self, $conn) = @_;
    my $cfg = $self->{cfg};
    $cfg->{_conn_state}{"$conn"} //= { auto_reconnect => 0, was_ready => 0 };
    weaken(my $weak_cfg = $cfg);
    $conn->on_error(sub {
        my $cfg = $weak_cfg or return;
        return if $cfg->{closed};
        my $st = $cfg->{_conn_state}{"$conn"} // {};
        return if $st->{auto_reconnect} && !$conn->connected;
        $cfg->{on_error}->($_[0]) if $cfg->{on_error};
    });
    $conn->on_disconnect(sub {
        my $cfg = $weak_cfg or return;
        return if $cfg->{closed};
        my $st = $cfg->{_conn_state}{"$conn"} // {};
        return unless $st->{was_ready};   # never came up: nothing lost
        $st->{was_ready} = 0;             # report once per loss
        $cfg->{on_error}->("connection to broker lost; reconnecting")
            if $cfg->{on_error};
    });
}

sub _bootstrap_connect {
    my ($self) = @_;
    $self->_bootstrap_try(0);
}

sub _bootstrap_try {
    my ($self, $idx) = @_;
    my $cfg = $self->{cfg};
    return if $cfg->{closed};

    my @bs = @{$cfg->{bootstrap}};
    if ($idx >= @bs) {
        $cfg->{_bootstrap_attempt} = undef;
        # Connect callbacks are queued (not captured) so close() can fail
        # them and nothing retains them past the first fire.
        my @cbs = @{delete $cfg->{_connect_cbs} // []};
        if (@cbs) {
            $_->(undef, "all bootstrap brokers unreachable") for @cbs;
        } elsif ($cfg->{on_error}) {
            $cfg->{on_error}->("all bootstrap brokers unreachable");
        }
        return;
    }
    my ($host, $port) = @{$bs[$idx]};

    my $conn = EV::Kafka::Conn::_new('EV::Kafka::Conn', $self->{loop});
    $self->_configure_conn($conn);

    # Keeps the attempt conn alive until on_connect/on_error fires,
    # without a closure cycle.
    $cfg->{_bootstrap_attempt} = $conn;

    weaken(my $weak = $self);
    # Replace the forwarding on_error from _configure_conn with the
    # try-next-broker handler for the duration of the bootstrap.
    $conn->on_error(sub {
        my $s = $weak or return;
        my $cfg = $s->{cfg};
        return if $cfg->{closed};
        $cfg->{_bootstrap_attempt} = undef;
        $s->_bootstrap_try($idx + 1);
    });

    # State-driven: this handler outlives the bootstrap (XS refires
    # on_connect after auto-reconnect), so it must not refire the cb.
    $conn->on_connect(sub {
        my $s = $weak or return;
        my $cfg = $s->{cfg};
        return if $cfg->{closed};
        if (!$cfg->{connected}) {
            # first successful bootstrap: promote attempt to bootstrap_conn
            my $conn = delete $cfg->{_bootstrap_attempt} or return;
            $cfg->{bootstrap_conn} = $conn;
            $cfg->{connected} = 1;
            # Now a pooled conn: auto-reconnect on loss, and the standard
            # suppressing handlers replace the try-next-broker one.
            $conn->auto_reconnect(1, 1000);
            $cfg->{_conn_state}{"$conn"} = { auto_reconnect => 1, was_ready => 1 };
            $s->_wire_conn_handlers($conn);
            $s->_refresh_metadata;
        } else {
            # auto-reconnect: refresh metadata, do NOT refire the callback
            $s->_refresh_metadata unless $cfg->{meta_pending};
        }
    });

    $conn->connect($host, $port, 10.0);
}

sub connect {
    my ($self, $cb) = @_;
    my $cfg = $self->{cfg};
    croak "EV::Kafka: client is closed" if $cfg->{closed};
    # Already up: fire immediately. Otherwise queue; every connect cb
    # fires exactly once — ($meta) on success, (undef, $err) on failure.
    if ($cfg->{connected} && $cfg->{meta}) {
        $cb->($cfg->{meta}) if $cb;
        return;
    }
    push @{$cfg->{_connect_cbs}}, $cb if $cb;
    $self->_bootstrap_connect
        unless $cfg->{_bootstrap_attempt} || $cfg->{meta_pending};
}

sub _merge_metadata {
    my ($cfg, $meta) = @_;
    for my $b (@{$meta->{brokers} // []}) {
        $cfg->{broker_map}{$b->{node_id}} = {
            host => $b->{host}, port => $b->{port}

lib/EV/Kafka.pm  view on Meta::CPAN

Enable automatic reconnection. C<$delay_ms> (default 1000) is the base
delay: the first retry happens after exactly C<$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 C<connect()> call.

=head2 leave_group($group_id, $member_id, $cb)

Send LeaveGroup to coordinator for fast partition rebalance.

=head2 create_topics(\@topics, $timeout_ms, $cb)

Create topics. Each element: C<{name, num_partitions, replication_factor}>.

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

=head2 delete_topics(\@topic_names, $timeout_ms, $cb)

Delete topics by name.

=head2 init_producer_id($transactional_id, $txn_timeout_ms, $cb)

Initialize a producer ID for idempotent/transactional produce.
Pass C<undef> for non-transactional idempotent producer.

=head2 add_partitions_to_txn($txn_id, $producer_id, $epoch, \@topics, $cb)

Register partitions with the transaction coordinator.

=head2 end_txn($txn_id, $producer_id, $epoch, $committed, $cb)

Commit (C<$committed=1>) or abort (C<$committed=0>) a transaction.

=head2 txn_offset_commit($txn_id, $group_id, $producer_id, $epoch, $generation, $member_id, \@offsets, $cb)

Commit consumer offsets within a transaction (API 28).

=head2 pending

Number of requests awaiting broker response.

=head2 state

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

=head2 Object lifetime

The connection is torn down when its last reference drops, or when
C<DESTROY> is called explicitly; explicit destruction is idempotent.
Pending request callbacks are invoked with a C<'destroyed'> error during
teardown. Once destroyed, the object is inert: every subsequent method
call on it (from any copy of the reference) croaks with
C<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
C<warn()>). A conn passed a custom C<EV::Loop> holds a reference on it,
so the loop cannot be destroyed while the conn is alive.

=head1 UTILITY FUNCTIONS

=head2 EV::Kafka::_murmur2($key)

Kafka-compatible murmur2 hash. Returns a non-negative 31-bit integer.

=head2 EV::Kafka::_crc32c($data)

CRC32C checksum (Castagnoli). Used internally for RecordBatch integrity.

=head2 EV::Kafka::_error_name($code)

Convert Kafka error code to string name.

=head1 RESULT STRUCTURES

=head2 Produce result

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

=head2 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
                }],
            }],
        }],
    };

=head2 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,