EV-Kafka
view release on metacpan or search on metacpan
lib/EV/Kafka.pm view on Meta::CPAN
batch_size => delete $opts{batch_size} // 16384,
partitioner => delete $opts{partitioner},
compression => delete $opts{compression}, # 'lz4', 'gzip', or undef
idempotent => delete $opts{idempotent} // 0,
transactional_id => delete $opts{transactional_id}, # enables transactions
fetch_max_wait_ms => delete $opts{fetch_max_wait_ms} // 500,
fetch_max_bytes => delete $opts{fetch_max_bytes} // 1048576,
fetch_min_bytes => delete $opts{fetch_min_bytes} // 1,
metadata_refresh => delete $opts{metadata_refresh} // 300,
flush_timeout => delete $opts{flush_timeout} // 30,
};
Carp::croak("EV::Kafka: unknown option(s): " . join(', ', sort keys %opts))
if %opts;
# Internal state
$cfg->{closed} = 0; # set by close()/DESTROY; gates everything
$cfg->{conns} = {}; # node_id => EV::Kafka::Conn
$cfg->{meta} = undef; # latest metadata response
$cfg->{leaders} = {}; # "topic:partition" => node_id
$cfg->{broker_map}= {}; # node_id => {host, port}
$cfg->{connected} = 0;
$cfg->{meta_pending} = 0;
$cfg->{pending_ops} = []; # ops waiting for metadata
# Producer state
$cfg->{batches} = {}; # "topic:partition" => [{rec, cb}]
$cfg->{next_sequence} = {}; # "topic:partition" => next sequence number
$cfg->{producer_id} = -1;
$cfg->{producer_epoch} = -1;
$cfg->{rr_counter} = 0;
# InitProducerId state machine: idle | init | ready | failed:$msg.
# Single-flight: 'init' queues completion callbacks in _pid_cbs.
$cfg->{_pid} = 'idle';
$cfg->{_pid_cbs} = [];
$cfg->{_txn_gen} = 0; # bumped on every txn state transition
$cfg->{_conn_state} = {}; # "$conn" => {auto_reconnect, was_ready}
# Consumer state
$cfg->{assignments} = []; # [{topic, partition, offset}]
$cfg->{fetch_active} = 0;
$cfg->{group} = undef;
my $self = bless { cfg => $cfg, loop => $loop }, "${class}::Client";
# Warn on credentials over plaintext.
if ($cfg->{sasl} && !$cfg->{tls}) {
my $mech = $cfg->{sasl}{mechanism} // '';
if ($mech eq 'PLAIN' || $mech =~ /^SCRAM-/) {
warn "EV::Kafka: SASL $mech configured without TLS — "
. "credentials will be sent over plaintext\n";
}
}
return $self;
}
package EV::Kafka::Client;
use EV;
use Carp 'croak';
use Scalar::Util 'weaken';
sub _any_conn {
my ($self) = @_;
my $cfg = $self->{cfg};
return undef if $cfg->{closed};
my $conn = $cfg->{bootstrap_conn};
for my $c (values %{$cfg->{conns}}) {
if ($c->connected) { $conn = $c; last }
}
return ($conn && $conn->connected) ? $conn : undef;
}
sub _get_or_create_conn {
my ($self, $node_id) = @_;
my $cfg = $self->{cfg};
return undef if $cfg->{closed};
return $cfg->{conns}{$node_id} if $cfg->{conns}{$node_id};
my $info = $cfg->{broker_map}{$node_id};
return undef unless $info;
my $conn = EV::Kafka::Conn::_new('EV::Kafka::Conn', $self->{loop});
$self->_configure_conn($conn);
$conn->auto_reconnect(1, 1000);
$cfg->{_conn_state}{"$conn"}{auto_reconnect} = 1;
$cfg->{conns}{$node_id} = $conn;
weaken(my $weak = $self);
$conn->on_connect(sub {
my $s = $weak or return;
my $cfg = $s->{cfg};
return if $cfg->{closed};
$cfg->{_conn_state}{"$conn"}{was_ready} = 1;
$s->_drain_pending_for($node_id);
});
$conn->connect($info->{host}, $info->{port}, 10.0);
return $conn;
}
sub _configure_conn {
my ($self, $conn) = @_;
my $cfg = $self->{cfg};
$conn->client_id($cfg->{client_id});
if ($cfg->{tls}) {
# Omit an undef CA: it would arrive in XS as "" and break
# SSL_CTX_load_verify_locations (NULL = system trust store).
my @tls_args = (1);
if (defined $cfg->{tls_ca_file}) {
push @tls_args, $cfg->{tls_ca_file}, $cfg->{tls_skip_verify};
} elsif ($cfg->{tls_skip_verify}) {
push @tls_args, '', 1;
}
$conn->tls(@tls_args);
}
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}
};
}
for my $t (@{$meta->{topics} // []}) {
next if $t->{error_code};
# Evict existing leaders first: a topic that shrank must not keep
# routing to its old partition count. (Names can't contain ':'.)
my $prefix = "$t->{name}:";
delete $cfg->{leaders}{$_}
for grep { index($_, $prefix) == 0 } keys %{$cfg->{leaders}};
for my $p (@{$t->{partitions} // []}) {
$cfg->{leaders}{"$t->{name}:$p->{partition}"} = $p->{leader};
}
}
}
sub _refresh_metadata {
my ($self) = @_;
my $cfg = $self->{cfg};
return if $cfg->{closed} || $cfg->{meta_pending};
$cfg->{meta_pending} = 1;
my $conn = $self->_any_conn;
unless ($conn) { $cfg->{meta_pending} = 0; return }
weaken(my $weak = $self);
$conn->metadata(undef, sub {
my ($meta, $err) = @_;
my $s = $weak or return;
my $cfg = $s->{cfg};
$cfg->{meta_pending} = 0;
return if $cfg->{closed};
if ($err) {
$cfg->{on_error}->("metadata: $err") if $cfg->{on_error};
return;
}
$cfg->{meta} = $meta;
_merge_metadata($cfg, $meta);
# assign bootstrap_conn a node_id if possible
if ($cfg->{bootstrap_conn} && $meta->{brokers} && @{$meta->{brokers}}) {
my $binfo = $meta->{brokers}[0];
$cfg->{conns}{$binfo->{node_id}} //= $cfg->{bootstrap_conn};
}
# Fire the one-shot connect callbacks; delete the queue first so
# close() can't fail a callback that is already firing.
my $fire = sub {
my @cbs = @{delete $cfg->{_connect_cbs} // []};
$_->($meta) for @cbs;
$cfg->{on_connect}->() if $cfg->{on_connect};
$cfg->{on_connect} = undef;
};
if (($cfg->{idempotent} || $cfg->{transactional_id}) && $cfg->{_pid} eq 'idle') {
$s->_init_idempotent(sub {
my $s2 = $weak or return;
return if $s2->{cfg}{closed};
if ($cfg->{_pid} =~ /^failed:(.*)/) {
# idempotence was requested but could not be set up:
# the connect callbacks get the error, not metadata.
my @cbs = @{delete $cfg->{_connect_cbs} // []};
$_->(undef, $1) for @cbs;
return;
}
$s2->_drain_all_pending;
$fire->();
});
} else {
$s->_drain_all_pending;
$fire->();
}
# Arm periodic metadata refresh (after first successful fetch).
$s->_arm_metadata_timer;
});
}
sub _arm_metadata_timer {
my ($self) = @_;
my $cfg = $self->{cfg};
return if $cfg->{_meta_timer};
my $interval = $cfg->{metadata_refresh} || 0;
return if $interval <= 0;
weaken(my $weak = $self);
$cfg->{_meta_timer} = EV::timer $interval, $interval, sub {
return unless $weak;
$weak->_refresh_metadata unless $weak->{cfg}{meta_pending};
};
}
sub _disarm_metadata_timer {
my ($self) = @_;
undef $self->{cfg}{_meta_timer};
}
sub _refresh_metadata_for_topic {
my ($self, $topic) = @_;
my $cfg = $self->{cfg};
return if $cfg->{closed} || $cfg->{meta_pending};
# Bound retries so a permanently-unavailable topic can't accumulate
# produce ops in pending_ops indefinitely.
my $tries = ++$cfg->{_topic_meta_tries}{$topic};
if ($tries > 10) {
delete $cfg->{_topic_meta_tries}{$topic};
my $msg = "metadata: topic '$topic' unavailable after $tries tries";
# Drop this topic's pending ops and report: to the op's callback
# if it has one, else on_error — never both (default on_error dies).
my (@keep, $reported);
for my $op (@{$cfg->{pending_ops}}) {
if ($op->{topic} && $op->{topic} eq $topic) {
if ($op->{cb}) { $op->{cb}->(undef, $msg); $reported = 1 }
} else {
push @keep, $op;
}
}
$cfg->{pending_ops} = \@keep;
$cfg->{on_error}->($msg) if !$reported && $cfg->{on_error};
return;
}
$cfg->{meta_pending} = 1;
my $conn = $self->_any_conn;
unless ($conn) { $cfg->{meta_pending} = 0; return }
weaken(my $weak = $self);
$conn->metadata([$topic], sub {
my ($meta, $err) = @_;
my $s = $weak or return;
my $cfg = $s->{cfg};
$cfg->{meta_pending} = 0;
return if $cfg->{closed};
if ($err) {
$cfg->{on_error}->("metadata: $err") if $cfg->{on_error};
return;
}
_merge_metadata($cfg, $meta);
# Fold the per-topic response into $cfg->{meta} so _num_partitions
# (which reads from {meta}{topics}) sees auto-created topics.
if ($cfg->{meta} && ref $cfg->{meta}{topics} eq 'ARRAY') {
my %by_name = map { $_->{name} => $_ } @{$cfg->{meta}{topics}};
for my $t (@{$meta->{topics} // []}) {
$by_name{$t->{name}} = $t;
}
$cfg->{meta}{topics} = [values %by_name];
} else {
$cfg->{meta} = $meta;
}
# if topic still has error, retry after delay
my $topic_ok = 0;
for my $t (@{$meta->{topics} // []}) {
if ($t->{name} eq $topic && !$t->{error_code} && @{$t->{partitions} // []}) {
$topic_ok = 1;
last;
}
}
if ($topic_ok) {
delete $cfg->{_topic_meta_tries}{$topic};
$s->_drain_all_pending;
} else {
# retry after short delay (topic being created); bounded above.
my $t; $t = EV::timer 0.5, 0, sub {
undef $t;
$weak->_refresh_metadata_for_topic($topic) if $weak;
};
}
});
}
# InitProducerId, single-flight. 'init' queues completion callbacks in
# _pid_cbs; failed:$msg is sticky (produce fails, begin_transaction dies).
sub _init_idempotent {
my ($self, $cb) = @_;
my $cfg = $self->{cfg};
return if $cfg->{closed};
if ($cfg->{_pid} eq 'init') {
push @{$cfg->{_pid_cbs}}, $cb if $cb;
return;
}
if ($cfg->{_pid} eq 'ready' || $cfg->{_pid} =~ /^failed:/) {
$cb->() if $cb;
return;
}
$cfg->{_pid} = 'init';
$cfg->{_pid_cbs} = [];
push @{$cfg->{_pid_cbs}}, $cb if $cb;
$self->_pid_attempt(1);
}
sub _pid_attempt {
my ($self, $attempt) = @_;
my $cfg = $self->{cfg};
return if $cfg->{closed};
weaken(my $weak = $self);
my $retry = sub {
my ($err) = @_;
my $s = $weak or return;
if ($attempt < 3) {
my $t; $t = EV::timer 0.5, 0, sub {
undef $t;
$weak->_pid_attempt($attempt + 1) if $weak;
};
} else {
$s->_pid_finish($err);
}
};
my $do_init = sub {
my ($conn) = @_;
return if $cfg->{closed};
$conn->init_producer_id($cfg->{transactional_id}, 30000, sub {
my ($res, $err) = @_;
return if $cfg->{closed};
if (!$err && $res && !$res->{error_code}) {
$cfg->{producer_id} = $res->{producer_id};
$cfg->{producer_epoch} = $res->{producer_epoch};
$weak->_pid_finish(undef) if $weak;
} else {
my $msg = $err || "InitProducerId error: " . ($res->{error_code} // '?');
$retry->($msg);
}
});
};
if ($cfg->{transactional_id}) {
# Transactional InitProducerId must go to the transaction
# coordinator; any other broker answers NOT_COORDINATOR.
my $conn = $self->_any_conn;
unless ($conn) { $retry->("no broker connection for FindCoordinator"); return }
$conn->find_coordinator($cfg->{transactional_id}, sub {
my ($res, $err) = @_;
my $s = $weak or return;
my $cfg = $s->{cfg};
return if $cfg->{closed};
if ($err || ($res && $res->{error_code})) {
my $msg = $err || "FindCoordinator(txn) error: $res->{error_code}";
$retry->($msg);
return;
}
$cfg->{broker_map}{$res->{node_id}} = {
host => $res->{host}, port => $res->{port}
};
my $txn_conn = $s->_get_or_create_conn($res->{node_id});
if ($txn_conn && $txn_conn->connected) {
$cfg->{_txn_coordinator} = $txn_conn;
$do_init->($txn_conn);
} else {
push @{$cfg->{pending_ops}}, {
node_id => $res->{node_id},
run => sub {
my $s2 = $weak or return;
my $tc = $cfg->{_txn_coordinator} =
$s2->_get_or_create_conn($res->{node_id});
if ($tc && $tc->connected) {
lib/EV/Kafka.pm view on Meta::CPAN
for my $op (@ops) {
if (defined $op->{node_id}) {
my $conn = $self->_get_or_create_conn($op->{node_id});
if ($conn && $conn->connected) {
$op->{run}->();
} else {
push @{$cfg->{pending_ops}}, $op;
}
} else {
$op->{run}->();
}
}
}
sub _get_leader {
my ($self, $topic, $partition) = @_;
return $self->{cfg}{leaders}{"$topic:$partition"};
}
sub _num_partitions {
my ($self, $topic) = @_;
my $meta = $self->{cfg}{meta} or return 0;
for my $t (@{$meta->{topics} // []}) {
return scalar @{$t->{partitions}} if $t->{name} eq $topic;
}
return 0;
}
sub _select_partition {
my ($self, $topic, $key) = @_;
my $np = $self->_num_partitions($topic);
return 0 unless $np > 0;
my $cfg = $self->{cfg};
if ($cfg->{partitioner}) {
return $cfg->{partitioner}->($topic, $key, $np);
}
if (defined $key && length $key) {
return EV::Kafka::_murmur2($key) % $np;
}
return $cfg->{rr_counter}++ % $np;
}
# --- Producer ---
sub produce {
my ($self, $topic, $key, $value, @rest) = @_;
my $cb;
my %opts;
for my $a (@rest) {
if (ref $a eq 'CODE') { $cb = $a }
elsif (ref $a eq 'HASH') { %opts = %$a }
}
for my $k (keys %opts) {
croak "EV::Kafka: produce: unknown option '$k'"
unless $k eq 'partition' || $k eq 'headers';
}
my $cfg = $self->{cfg};
croak "EV::Kafka: client is closed" if $cfg->{closed};
weaken(my $weak = $self);
# ensure we have metadata
unless ($cfg->{meta}) {
push @{$cfg->{pending_ops}}, {
topic => $topic,
cb => $cb,
run => sub { $weak->produce($topic, $key, $value, @rest) if $weak },
};
$self->_refresh_metadata unless $cfg->{meta_pending};
return;
}
# Idempotent/transactional producers must wait for InitProducerId:
# a batch sent without producer_id/epoch silently voids exactly-once.
if (($cfg->{idempotent} || $cfg->{transactional_id})
&& $cfg->{_pid} ne 'ready') {
if ($cfg->{_pid} =~ /^failed:(.*)/) {
$cb->(undef, "idempotent producer unavailable: $1") if $cb;
return;
}
push @{$cfg->{pending_ops}}, {
topic => $topic,
cb => $cb,
run => sub { $weak->produce($topic, $key, $value, @rest) if $weak },
};
$self->_init_idempotent if $cfg->{_pid} eq 'idle';
return;
}
my $partition = exists $opts{partition}
? $opts{partition}
: $self->_select_partition($topic, $key);
my $leader_id = $self->_get_leader($topic, $partition);
unless (defined $leader_id) {
# topic/partition unknown — request metadata to trigger auto-creation.
# Tag with topic+cb so the retry-bound cleanup can report the error
# instead of leaking the op in pending_ops forever.
push @{$cfg->{pending_ops}}, {
topic => $topic,
cb => $cb,
run => sub { $weak->produce($topic, $key, $value, @rest) if $weak },
};
$self->_refresh_metadata_for_topic($topic) unless $cfg->{meta_pending};
return;
}
my $conn = $self->_get_or_create_conn($leader_id);
unless ($conn && $conn->connected) {
# Tag with topic+cb too, so the retry-bound cleanup can report it.
push @{$cfg->{pending_ops}}, {
node_id => $leader_id,
topic => $topic,
cb => $cb,
run => sub { $weak->produce($topic, $key, $value, @rest) if $weak },
};
return;
}
# Accumulate into batch
my $bkey = "$topic:$partition";
my $rec = { key => $key, value => $value };
$rec->{headers} = $opts{headers} if $opts{headers};
push @{$cfg->{batches}{$bkey} //= []}, { rec => $rec, cb => $cb };
# Check batch size threshold
my $batch = $cfg->{batches}{$bkey};
my $batch_bytes = 0;
for my $b (@$batch) {
$batch_bytes += length($b->{rec}{value} // '') + length($b->{rec}{key} // '') + 20;
}
if ($batch_bytes >= $cfg->{batch_size}) {
$self->_flush_batch($topic, $partition, $conn);
} elsif (!$cfg->{_linger_active}) {
# start linger timer
$cfg->{_linger_active} = 1;
weaken(my $weak = $self);
$cfg->{_linger_timer} = EV::timer $cfg->{linger_ms} / 1000.0, 0, sub {
$cfg->{_linger_active} = 0;
$weak->_flush_all_batches if $weak;
};
}
}
sub _flush_batch {
my ($self, $topic, $partition, $conn) = @_;
my $cfg = $self->{cfg};
my $bkey = "$topic:$partition";
my $idempotent = defined $cfg->{producer_id} && $cfg->{producer_id} >= 0;
# Idempotent: one in-flight batch per partition — racing batches can
# alias sequence numbers (OutOfOrderSequenceNumber, non-retriable).
return if $idempotent && $cfg->{_inflight}{$bkey};
my $batch = delete $cfg->{batches}{$bkey};
return unless $batch && @$batch;
my @records = map { $_->{rec} } @$batch;
my @cbs = map { $_->{cb} } @$batch;
my %popts = (acks => $cfg->{acks});
$popts{compression} = $cfg->{compression} if $cfg->{compression};
$popts{transactional_id} = $cfg->{transactional_id} if $cfg->{_txn_active};
my $saved_seq;
if ($idempotent) {
$popts{producer_id} = $cfg->{producer_id};
$popts{producer_epoch} = $cfg->{producer_epoch};
$saved_seq = $cfg->{next_sequence}{$bkey} // 0;
$popts{base_sequence} = $saved_seq;
$cfg->{next_sequence}{$bkey} = $saved_seq + scalar @records;
$cfg->{_inflight}{$bkey} = 1;
}
$self->_add_txn_partition($topic, $partition) if $cfg->{_txn_active};
# retry count persists on the batch across re-queues
$cfg->{_batch_retries}{$bkey} //= 3;
# Transaction generation at send time: a late response must never
# re-queue a batch from an aborted/committed transaction.
my $txn_gen = $cfg->{_txn_gen};
weaken(my $weak_self = $self);
$conn->produce_batch($topic, $partition, \@records, \%popts, sub {
my ($result, $err) = @_;
delete $cfg->{_inflight}{$bkey} if $idempotent;
my $retriable = 0;
my $fatal_seq = 0;
my $dup_seq = 0;
if (!$err && $result && ref $result->{topics} eq 'ARRAY') {
for my $t (@{$result->{topics}}) {
for my $p (@{$t->{partitions} // []}) {
my $ec = $p->{error_code} // 0;
$retriable = $ec if $ec == 6 || $ec == 15 || $ec == 16;
# 45/47: broker fenced this producer — re-init and retry.
# 46 = DUPLICATE_SEQUENCE: broker already has this batch;
# treat as an ack (re-sending would create the duplicates).
$fatal_seq = $ec if $ec == 45 || $ec == 47;
$dup_seq = 1 if $ec == 46;
}
}
}
# 46 is an ack: clear it in the delivered result.
if ($dup_seq && !$fatal_seq && !$retriable && $result) {
for my $t (@{$result->{topics}}) {
for my $p (@{$t->{partitions} // []}) {
$p->{error_code} = 0 if ($p->{error_code} // 0) == 46;
}
}
}
# Fatal sequence error: re-init producer id/epoch and retry the
# batch (guarded against a stale txn generation by the $txn_gen check).
if ($idempotent && $fatal_seq && !$cfg->{closed}
&& $txn_gen == $cfg->{_txn_gen}) {
if ($cfg->{_txn_active}) {
# fenced mid-transaction: the txn is dead; only abort ->
# begin can continue. Surface it, don't silently re-init.
my $msg = "producer fenced by broker (code $fatal_seq) during transaction";
$cfg->{on_error}->($msg) if $cfg->{on_error};
delete $cfg->{_batch_retries}{$bkey};
for my $cb (@cbs) {
next unless $cb;
eval { $cb->(undef, $msg) };
warn "EV::Kafka: produce callback error: $@" if $@;
}
return;
}
if (($cfg->{_epoch_retries}{$bkey} //= 1) > 0) {
$cfg->{_epoch_retries}{$bkey}--;
# rewind sequence and re-queue the batch
$cfg->{next_sequence} = {};
$cfg->{_acked_sequence} = {};
if (exists $cfg->{batches}{$bkey}) {
unshift @{$cfg->{batches}{$bkey}}, @$batch;
} else {
$cfg->{batches}{$bkey} = $batch;
}
# force re-init then re-flush; _pid single-flight makes N
# fenced partitions share ONE InitProducerId round trip.
$cfg->{producer_id} = -1;
lib/EV/Kafka.pm view on Meta::CPAN
if ($retriable && !$cfg->{closed} && $txn_gen == $cfg->{_txn_gen}
&& ($cfg->{_batch_retries}{$bkey} // 0) > 0) {
$cfg->{_batch_retries}{$bkey}--;
$cfg->{next_sequence}{$bkey} = $saved_seq if defined $saved_seq;
if (exists $cfg->{batches}{$bkey}) {
unshift @{$cfg->{batches}{$bkey}}, @$batch;
} else {
$cfg->{batches}{$bkey} = $batch;
}
$weak_self->_refresh_metadata if $weak_self && !$cfg->{meta_pending};
my $rt; $rt = EV::timer 0.5, 0, sub {
undef $rt;
$weak_self->_flush_all_batches if $weak_self;
};
return;
}
delete $cfg->{_batch_retries}{$bkey};
# Track the broker-acked sequence high-water mark so
# abort_transaction can roll back next_sequence without gaps, and
# restore the epoch-retry budget. 45/47 don't advance the mark.
if ($idempotent && !$err && !$fatal_seq && defined $saved_seq) {
my $next = $saved_seq + scalar @records;
my $cur = $cfg->{_acked_sequence}{$bkey} // 0;
$cfg->{_acked_sequence}{$bkey} = $next if $next > $cur;
delete $cfg->{_epoch_retries}{$bkey};
}
# Each record callback fires even if a sibling's callback dies.
for my $cb (@cbs) {
next unless $cb;
eval { $cb->($result, $err) };
warn "EV::Kafka: produce callback error: $@" if $@;
}
# Idempotent: kick any batch that accumulated while we were in-flight.
if ($idempotent && !$cfg->{closed} && $weak_self
&& $cfg->{batches}{$bkey} && @{$cfg->{batches}{$bkey}}) {
$weak_self->_flush_all_batches;
}
});
}
sub _flush_all_batches {
my ($self) = @_;
my $cfg = $self->{cfg};
return if $cfg->{closed};
my $skipped = 0;
for my $bkey (keys %{$cfg->{batches}}) {
my ($topic, $partition) = split /:/, $bkey, 2;
my $leader_id = $self->_get_leader($topic, $partition);
unless (defined $leader_id) { $skipped++; next }
my $conn = $self->_get_or_create_conn($leader_id);
unless ($conn && $conn->connected) { $skipped++; next }
$self->_flush_batch($topic, $partition, $conn);
}
# re-arm timer if batches were skipped (connection not yet ready)
if ($skipped && keys %{$cfg->{batches}}) {
$cfg->{_linger_active} = 1;
weaken(my $weak = $self);
$cfg->{_linger_timer} = EV::timer 0.1, 0, sub {
$cfg->{_linger_active} = 0;
$weak->_flush_all_batches if $weak;
};
}
}
sub produce_many {
my ($self, $messages, $cb) = @_;
croak "EV::Kafka: client is closed" if $self->{cfg}{closed};
my $remaining = scalar @$messages;
return $cb->() if $cb && !$remaining;
my @errors;
my $acks0 = ($self->{cfg}{acks} == 0);
for my $msg (@$messages) {
my ($topic, $key, $value, @rest);
if (ref $msg eq 'ARRAY') {
($topic, $key, $value, @rest) = @$msg;
} else {
($topic, $key, $value) = @{$msg}{qw(topic key value)};
my %opts = %$msg;
delete @opts{qw(topic key value)};
push @rest, \%opts if %opts;
}
if ($acks0) {
$self->produce($topic, $key, $value, @rest);
} else {
$self->produce($topic, $key, $value, @rest, sub {
my ($result, $err) = @_;
push @errors, $err if $err;
if (--$remaining <= 0 && $cb) {
$cb->(@errors ? \@errors : ());
}
});
}
}
$cb->(@errors ? \@errors : ()) if $cb && $acks0;
}
sub flush {
my ($self, $cb) = @_;
my $cfg = $self->{cfg};
croak "EV::Kafka: client is closed" if $cfg->{closed};
# flush any accumulated linger batches first
$self->_flush_all_batches;
undef $cfg->{_linger_timer};
$cfg->{_linger_active} = 0;
# Drained when all of these reach zero: in-flight requests on every
# connection, pre-metadata pending_ops, gated/queued idempotent batches.
my $outstanding = sub {
my $p = 0;
my %seen;
for my $c (values %{$cfg->{conns} // {}}) {
next unless $c && $c->connected;
$p += $c->pending;
$seen{$$c} = 1;
}
$p += $cfg->{bootstrap_conn}->pending
lib/EV/Kafka.pm view on Meta::CPAN
};
# Stored so close() can fail the callback instead of stranding it.
$cfg->{_flush_timer} = $check;
$cfg->{_flush_cb} = $cb;
}
# --- Consumer ---
sub assign {
my ($self, $partitions) = @_;
my $cfg = $self->{cfg};
croak "EV::Kafka: client is closed" if $cfg->{closed};
$cfg->{assignments} = $partitions;
$cfg->{_fetch_err_seen} = {}; # fresh assignment: fresh error budget
}
sub seek {
my ($self, $topic, $partition, $offset_or_ts, $cb) = @_;
croak "EV::Kafka: client is closed" if $self->{cfg}{closed};
# offset_or_ts: integer offset, or -1 (latest), -2 (earliest)
for my $a (@{$self->{cfg}{assignments}}) {
if ($a->{topic} eq $topic && $a->{partition} == $partition) {
if ($offset_or_ts >= 0) {
$a->{offset} = $offset_or_ts;
$cb->() if $cb;
} else {
# resolve via list_offsets
my $leader_id = $self->_get_leader($topic, $partition);
my $conn = defined($leader_id) ? $self->_get_or_create_conn($leader_id) : undef;
if ($conn && $conn->connected) {
$conn->list_offsets($topic, $partition, $offset_or_ts, sub {
my ($res, $err) = @_;
if (!$err && $res) {
my $off = $res->{topics}[0]{partitions}[0]{offset};
$a->{offset} = $off if defined $off;
}
$cb->($res, $err) if $cb;
});
} else {
$cb->(undef, "seek: no connection to leader for $topic:$partition")
if $cb;
}
}
return;
}
}
$cb->(undef, "seek: $topic:$partition is not assigned") if $cb;
}
sub poll {
my ($self, $cb) = @_;
my $cfg = $self->{cfg};
croak "EV::Kafka: client is closed" if $cfg->{closed};
# Empty assignment is a successful empty poll, not a dropped callback.
if (!@{$cfg->{assignments}}) {
$cb->() if $cb;
return;
}
unless ($cfg->{meta}) {
weaken(my $weak = $self);
push @{$cfg->{pending_ops}}, {
cb => $cb,
run => sub { $weak->poll($cb) if $weak },
};
$self->_refresh_metadata unless $cfg->{meta_pending};
return;
}
# Group assignments by leader for multi-partition fetch
my %by_leader; # leader_id => { topic => [{partition, offset, assign_ref}] }
for my $a (@{$cfg->{assignments}}) {
my $leader_id = $self->_get_leader($a->{topic}, $a->{partition});
next unless defined $leader_id;
push @{$by_leader{$leader_id}{$a->{topic}}}, {
partition => $a->{partition},
offset => $a->{offset},
_assign => $a,
};
}
my $dispatched = 0;
my $first_err;
weaken(my $weak = $self);
for my $leader_id (keys %by_leader) {
my $conn = $self->_get_or_create_conn($leader_id);
next unless $conn && $conn->connected;
$dispatched++;
# build fetch_multi argument: {topic => [{partition, offset}]}
my %fetch_arg;
my %assign_map; # "topic:partition" => [assignment ref, requested offset]
for my $topic (keys %{$by_leader{$leader_id}}) {
for my $p (@{$by_leader{$leader_id}{$topic}}) {
push @{$fetch_arg{$topic}}, {
partition => $p->{partition},
offset => $p->{offset},
};
$assign_map{"$topic:$p->{partition}"} = [$p->{_assign}, $p->{offset}];
}
}
$conn->fetch_multi(\%fetch_arg, {
max_bytes => $cfg->{fetch_max_bytes},
max_wait_ms => $cfg->{fetch_max_wait_ms},
min_bytes => $cfg->{fetch_min_bytes},
}, sub {
my ($result, $err) = @_;
$dispatched--;
$first_err //= $err if $err;
# close() silences messages but never drops the poll callback.
if (!$cfg->{closed} && !$err && $result && ref $result->{topics} eq 'ARRAY') {
for my $t (@{$result->{topics}}) {
for my $p (@{$t->{partitions} // []}) {
my $entry = $assign_map{"$t->{topic}:$p->{partition}"};
my $ec = $p->{error_code} // 0;
if ($ec) {
# Per-partition error: recover or report.
if ($weak) {
my $emsg = $weak->_fetch_partition_error(
$t->{topic}, $p->{partition}, $ec,
$entry ? $entry->[0] : undef);
$first_err //= $emsg if $emsg;
}
next;
}
my $records = $p->{records} // [];
for my $r (@$records) {
next unless $cfg->{on_message};
# a die must not skip the remaining records
eval {
$cfg->{on_message}->(
$t->{topic}, $p->{partition},
$r->{offset}, $r->{key}, $r->{value},
$r->{headers}
);
};
warn "EV::Kafka: on_message error: $@" if $@;
}
# Seek-clobber guard: a seek() that landed while
# this fetch was in flight wins; don't overwrite.
if (@$records && $entry
&& $entry->[0]{offset} == $entry->[1]) {
lib/EV/Kafka.pm view on Meta::CPAN
} else {
$result->{$key} = { current => $a->{offset}, latest => 0, lag => 0 };
$cb->($result, $first_err) if $cb && --$remaining <= 0;
}
}
}
sub error_name {
shift if ref $_[0]; # allow $kafka->error_name or EV::Kafka::Client::error_name
return EV::Kafka::_error_name($_[0]);
}
# --- Consumer Group ---
sub subscribe {
my ($self, @args) = @_;
my @topics;
my %opts;
# subscribe('topic1', 'topic2', group_id => 'g', ...)
while (@args) {
if ($args[0] =~ /^(group_id|group_instance_id|on_assign|on_revoke|session_timeout|rebalance_timeout|heartbeat_interval|auto_commit|auto_offset_reset)$/) {
my $k = shift @args;
croak "EV::Kafka: subscribe: option '$k' requires a value"
unless @args;
$opts{$k} = shift @args;
} else {
push @topics, shift @args;
}
}
for my $t (@topics) {
croak "EV::Kafka: subscribe: invalid topic name '$t'"
unless $t =~ /^[A-Za-z0-9._-]+$/;
}
my $cfg = $self->{cfg};
croak "EV::Kafka: client is closed" if $cfg->{closed};
my $group_id = $opts{group_id};
die "group_id required" unless defined $group_id;
$cfg->{group} = {
group_id => $group_id,
member_id => '',
generation => -1,
topics => \@topics,
on_assign => $opts{on_assign},
on_revoke => $opts{on_revoke},
session_timeout => $opts{session_timeout} // 30000,
rebalance_timeout => $opts{rebalance_timeout} // 60000,
heartbeat_interval => $opts{heartbeat_interval} // 3,
auto_commit => $opts{auto_commit} // 1,
auto_offset_reset => $opts{auto_offset_reset} // 'earliest',
group_instance_id => $opts{group_instance_id},
coordinator => undef,
heartbeat_timer => undef,
state => 'init',
};
# Step 1: ensure we have metadata
unless ($cfg->{meta}) {
weaken(my $weak = $self);
push @{$cfg->{pending_ops}}, {
run => sub { $weak->_group_start if $weak },
};
$self->_refresh_metadata unless $cfg->{meta_pending};
return;
}
$self->_group_start;
}
sub _group_start {
my ($self) = @_;
my $cfg = $self->{cfg};
return if $cfg->{closed};
my $g = $cfg->{group} or return;
my $conn = $self->_any_conn;
return $self->_group_retry('no broker connection for group start')
unless $conn;
$g->{state} = 'finding';
weaken(my $weak = $self);
$conn->find_coordinator($g->{group_id}, sub {
my ($res, $err) = @_;
my $s = $weak or return;
my $cfg = $s->{cfg};
return if $cfg->{closed};
my $g = $cfg->{group} or return;
if ($err || $res->{error_code}) {
my $msg = $err || "FindCoordinator error: $res->{error_code}";
$cfg->{on_error}->($msg) if $cfg->{on_error};
# retry after delay
my $t; $t = EV::timer 1, 0, sub { undef $t; $weak->_group_start if $weak };
return;
}
# Store coordinator info and connect
$cfg->{broker_map}{$res->{node_id}} = {
host => $res->{host}, port => $res->{port}
};
my $coord = $s->_get_or_create_conn($res->{node_id});
$g->{coordinator} = $coord;
$g->{coordinator_id} = $res->{node_id};
if ($coord && $coord->connected) {
$s->_group_join;
} else {
push @{$cfg->{pending_ops}}, {
node_id => $res->{node_id},
run => sub { $weak->_group_join if $weak },
};
}
});
}
# Bounded retry for coordinator request failures (usually a dead
# coordinator conn): re-run discovery via _group_start. After 5 attempts
# stop the group and surface the error. Budget resets at 'stable'.
sub _group_retry {
my ($self, $msg) = @_;
my $cfg = $self->{cfg};
return if $cfg->{closed};
my $g = $cfg->{group} or return;
if (++$g->{_start_retries} > 5) {
$g->{_start_retries} = 0;
$g->{state} = 'stopped';
$cfg->{on_error}->($msg) if $cfg->{on_error};
return;
}
weaken(my $weak = $self);
my $t; $t = EV::timer 1, 0, sub { undef $t; $weak->_group_start if $weak };
}
sub _group_join {
my ($self) = @_;
my $cfg = $self->{cfg};
return if $cfg->{closed};
my $g = $cfg->{group} or return;
my $coord = $g->{coordinator} or return;
$g->{state} = 'joining';
weaken(my $weak = $self);
$coord->join_group(
$g->{group_id}, $g->{member_id},
$g->{topics}, sub {
my ($res, $err) = @_;
my $s = $weak or return;
my $cfg = $s->{cfg};
return if $cfg->{closed};
my $g = $cfg->{group} or return;
if ($err) {
$s->_group_retry("JoinGroup: $err");
return;
}
if ($res->{error_code} == 15 || $res->{error_code} == 16) {
# COORDINATOR_NOT_AVAILABLE / NOT_COORDINATOR — re-discover
my $t; $t = EV::timer 1, 0, sub { undef $t; $weak->_group_start if $weak };
return;
}
if ($res->{error_code} == 27) {
# REBALANCE_IN_PROGRESS — retry
my $t; $t = EV::timer 1, 0, sub { undef $t; $weak->_group_join if $weak };
return;
}
if ($res->{error_code} == 79) {
# MEMBER_ID_REQUIRED — retry with assigned member_id
$g->{member_id} = $res->{member_id} if $res->{member_id};
$s->_group_join;
return;
}
if ($res->{error_code} == 22 || $res->{error_code} == 25) {
# ILLEGAL_GENERATION / UNKNOWN_MEMBER_ID — broker-side
# session expired or generation rolled. Reset state and
# rejoin from scratch.
$g->{member_id} = '';
$g->{generation} = -1;
my $t; $t = EV::timer 1, 0, sub { undef $t; $weak->_group_start if $weak };
return;
}
if ($res->{error_code}) {
$cfg->{on_error}->(sprintf "JoinGroup: %s (code %d)",
$s->error_name($res->{error_code}),
$res->{error_code}) if $cfg->{on_error};
return;
}
$g->{member_id} = $res->{member_id};
$g->{generation} = $res->{generation_id};
my $is_leader = defined($res->{leader}) && defined($res->{member_id})
&& $res->{leader} eq $res->{member_id};
# Build assignments (if leader)
my $assignments = [];
if ($is_leader && $res->{members} && @{$res->{members}}) {
$assignments = $s->_assign_partitions($res->{members}, $g->{topics});
}
$s->_group_sync($assignments);
},
$g->{session_timeout}, $g->{rebalance_timeout},
$g->{group_instance_id}
lib/EV/Kafka.pm view on Meta::CPAN
$assigned{$key} = 1;
}
}
}
}
# Step 2: distribute unassigned partitions to least-loaded members
my @unassigned = grep { !$assigned{"$_->{topic}:$_->{partition}"} } @all_parts;
for my $p (@unassigned) {
my $min_mid = $member_ids[0];
my $min_count = scalar @{$member_parts{$min_mid}};
for my $mid (@member_ids) {
if (scalar @{$member_parts{$mid}} < $min_count) {
$min_count = scalar @{$member_parts{$mid}};
$min_mid = $mid;
}
}
push @{$member_parts{$min_mid}}, $p;
}
# Save for next rebalance
$cfg->{_prev_assignments} = { %member_parts };
# Encode assignments
my @assignments;
for my $mid (@member_ids) {
my %by_topic;
for my $p (@{$member_parts{$mid}}) {
push @{$by_topic{$p->{topic}}}, $p->{partition};
}
my $buf = '';
$buf .= pack('n', 0); # version
$buf .= pack('N', scalar keys %by_topic);
for my $t (sort keys %by_topic) {
$buf .= pack('n', length($t)) . $t;
$buf .= pack('N', scalar @{$by_topic{$t}});
for my $pid (@{$by_topic{$t}}) {
$buf .= pack('N', $pid);
}
}
$buf .= pack('N', -1); # user_data = null
push @assignments, {
member_id => $mid,
assignment => $buf,
};
}
return \@assignments;
}
sub _group_sync {
my ($self, $assignments) = @_;
my $cfg = $self->{cfg};
return if $cfg->{closed};
my $g = $cfg->{group} or return;
my $coord = $g->{coordinator} or return;
$g->{state} = 'syncing';
weaken(my $weak = $self);
my $sync_cb = sub {
my ($res, $err) = @_;
my $s = $weak or return;
my $cfg = $s->{cfg};
return if $cfg->{closed};
my $g = $cfg->{group} or return;
if ($err) {
$s->_group_retry("SyncGroup: $err");
return;
}
if ($res->{error_code} == 27) {
# REBALANCE_IN_PROGRESS — rejoin
my $t; $t = EV::timer 1, 0, sub { undef $t; $weak->_group_join if $weak };
return;
}
if ($res->{error_code} == 22 || $res->{error_code} == 25) {
# ILLEGAL_GENERATION / UNKNOWN_MEMBER_ID — start over.
$g->{member_id} = '';
$g->{generation} = -1;
my $t; $t = EV::timer 1, 0, sub { undef $t; $weak->_group_start if $weak };
return;
}
if ($res->{error_code}) {
$cfg->{on_error}->(sprintf "SyncGroup: %s (code %d)",
$s->error_name($res->{error_code}),
$res->{error_code}) if $cfg->{on_error};
return;
}
# Decode assignment
my $data = $res->{assignment} // '';
my $dlen = length $data;
my @my_assignments;
if ($dlen >= 6) {
my $off = 2; # skip version
my $tc = unpack('N', substr($data, $off, 4)); $off += 4;
for my $i (0..$tc-1) {
last unless $off + 2 <= $dlen;
my $tlen = unpack('n', substr($data, $off, 2)); $off += 2;
last unless $off + $tlen <= $dlen;
my $tname = substr($data, $off, $tlen); $off += $tlen;
last unless $off + 4 <= $dlen;
my $pc = unpack('N', substr($data, $off, 4)); $off += 4;
for my $j (0..$pc-1) {
last unless $off + 4 <= $dlen;
my $pid = unpack('N', substr($data, $off, 4)); $off += 4;
my $reset = ($g->{auto_offset_reset} // 'earliest') eq 'latest' ? -1 : -2;
push @my_assignments, {
topic => $tname, partition => $pid, offset => $reset
};
}
}
}
$g->{state} = 'stable';
$g->{_start_retries} = 0; # join/sync retry budget resets
# Fetch committed offsets, then start consuming
$s->_fetch_committed_offsets(\@my_assignments, sub {
my $s2 = $weak or return;
my $cfg = $s2->{cfg};
return if $cfg->{closed};
my $g = $cfg->{group} or return;
$cfg->{assignments} = \@my_assignments;
$cfg->{_fetch_err_seen} = {}; # rebalance: fresh error budget
# Fire on_assign
$g->{on_assign}->(\@my_assignments) if $g->{on_assign};
# Start heartbeat
$s2->_start_heartbeat;
# Start fetch loop
$s2->_start_fetch_loop;
});
};
$coord->sync_group(
$g->{group_id}, $g->{generation}, $g->{member_id},
$assignments, $sync_cb, $g->{group_instance_id}
);
}
sub _fetch_committed_offsets {
my ($self, $assignments, $cb) = @_;
my $cfg = $self->{cfg};
return if $cfg->{closed};
my $g = $cfg->{group} or return $cb->();
my $coord = $g->{coordinator};
return $cb->() unless $coord && $coord->connected && @$assignments;
# Build topics array for offset_fetch
my %by_topic;
for my $a (@$assignments) {
push @{$by_topic{$a->{topic}}}, $a->{partition};
}
my @topics;
for my $t (sort keys %by_topic) {
push @topics, { topic => $t, partitions => $by_topic{$t} };
}
weaken(my $weak = $self);
$coord->offset_fetch($g->{group_id}, \@topics, sub {
my ($res, $err) = @_;
my $s = $weak or return;
my $cfg = $s->{cfg};
return if $cfg->{closed};
if (!$err && $res && ref $res->{topics} eq 'ARRAY') {
for my $t (@{$res->{topics}}) {
for my $p (@{$t->{partitions} // []}) {
next if $p->{error_code};
next if $p->{offset} < 0; # no committed offset
for my $a (@$assignments) {
if ($a->{topic} eq $t->{topic} && $a->{partition} == $p->{partition}) {
$a->{offset} = $p->{offset};
}
}
}
}
}
# For partitions with unresolved offset (-2=earliest, -1=latest), resolve via ListOffsets
my @need_offsets = grep { $_->{offset} < 0 } @$assignments;
if (@need_offsets) {
my $remaining = scalar @need_offsets;
for my $a (@need_offsets) {
my $leader_id = $s->_get_leader($a->{topic}, $a->{partition});
my $lconn = defined($leader_id) ? $s->_get_or_create_conn($leader_id) : undef;
if ($lconn && $lconn->connected) {
$lconn->list_offsets($a->{topic}, $a->{partition}, $a->{offset}, sub {
my ($lres, $lerr) = @_;
if (!$lerr && $lres && ref $lres->{topics} eq 'ARRAY') {
for my $lt (@{$lres->{topics}}) {
for my $lp (@{$lt->{partitions} // []}) {
$a->{offset} = $lp->{offset} if !$lp->{error_code};
}
}
}
$remaining--;
$cb->() if $remaining <= 0;
});
} else {
$a->{offset} = 0;
$remaining--;
$cb->() if $remaining <= 0;
}
}
} else {
$cb->();
}
});
}
sub _start_heartbeat {
my ($self) = @_;
my $cfg = $self->{cfg};
return if $cfg->{closed};
my $g = $cfg->{group} or return;
# Weak $self only; group state is re-derived per tick, so a stale
# group after re-subscribe is never acted on.
weaken(my $weak = $self);
$g->{heartbeat_timer} = EV::timer $g->{heartbeat_interval}, $g->{heartbeat_interval}, sub {
my $s = $weak or return;
my $cfg = $s->{cfg};
return if $cfg->{closed};
my $g = $cfg->{group} or return;
return unless $g->{state} eq 'stable';
my $coord = $g->{coordinator};
return unless $coord && $coord->connected;
$coord->heartbeat($g->{group_id}, $g->{generation}, $g->{member_id}, sub {
my ($res, $err) = @_;
my $s = $weak or return;
my $cfg = $s->{cfg};
return if $cfg->{closed};
my $g = $cfg->{group} or return;
if ($err) { return }
return unless $res;
my $ec = $res->{error_code} // 0;
if ($ec == 27) {
# REBALANCE_IN_PROGRESS
$g->{state} = 'rebalancing';
$g->{on_revoke}->($cfg->{assignments}) if $g->{on_revoke};
$s->_stop_heartbeat;
$s->_stop_fetch_loop;
$s->_group_join;
} elsif ($ec == 22 || $ec == 25) {
# ILLEGAL_GENERATION / UNKNOWN_MEMBER_ID — start over.
$g->{state} = 'rebalancing';
$g->{member_id} = '';
$g->{generation} = -1;
$s->_stop_heartbeat;
$s->_stop_fetch_loop;
$s->_group_start;
} elsif ($ec == 15 || $ec == 16) {
# Coordinator moved — re-discover. Stop the fetch loop
# too: assignments are stale until the new coordinator
# confirms generation.
$g->{state} = 'rebalancing';
$s->_stop_heartbeat;
$s->_stop_fetch_loop;
$s->_group_start;
}
}, $g->{group_instance_id});
};
}
sub _stop_heartbeat {
my ($self) = @_;
my $g = $self->{cfg}{group} or return;
undef $g->{heartbeat_timer};
}
sub _start_fetch_loop {
my ($self) = @_;
my $cfg = $self->{cfg};
return if $cfg->{closed} || $cfg->{fetch_active};
$cfg->{fetch_active} = 1;
# Generation-tagged in-flight flag: a completion from a previous
# loop instance must not clear THIS loop's flag.
my $gen = ++$cfg->{_fetch_gen};
$cfg->{_fetch_in_flight} = 0;
weaken(my $weak = $self);
$cfg->{fetch_timer} = EV::timer 0, 0.1, sub {
return unless $weak && $cfg->{fetch_active};
# skip ticks while a prior poll round is still in flight
return if $cfg->{_fetch_in_flight};
$cfg->{_fetch_in_flight} = $gen;
$weak->poll(sub {
$cfg->{_fetch_in_flight} = 0
if ($cfg->{_fetch_in_flight} // 0) == $gen;
});
};
}
sub _stop_fetch_loop {
my ($self) = @_;
my $cfg = $self->{cfg};
$cfg->{fetch_active} = 0;
$cfg->{_fetch_gen}++; # invalidate completions from the stopped loop
$cfg->{_fetch_in_flight} = 0;
undef $cfg->{fetch_timer};
}
sub commit {
my ($self, $cb) = @_;
my $cfg = $self->{cfg};
croak "EV::Kafka: client is closed" if $cfg->{closed};
my $g = $cfg->{group};
unless ($g) { $cb->() if $cb; return }
my $coord = $g->{coordinator};
unless ($coord && $coord->connected) { $cb->() if $cb; return }
# Build offset commit data from current assignments
my %by_topic;
for my $a (@{$cfg->{assignments} // []}) {
push @{$by_topic{$a->{topic}}}, {
partition => $a->{partition},
offset => $a->{offset},
};
}
my @topics;
for my $t (sort keys %by_topic) {
push @topics, { topic => $t, partitions => $by_topic{$t} };
}
if (!@topics) { $cb->() if $cb; return }
$coord->offset_commit($g->{group_id}, $g->{generation}, $g->{member_id}, \@topics, sub {
my ($res, $err) = @_;
$cb->($err) if $cb;
});
}
sub unsubscribe {
my ($self, $cb) = @_;
my $cfg = $self->{cfg};
croak "EV::Kafka: client is closed" if $cfg->{closed};
my $g = $cfg->{group};
$self->_stop_heartbeat;
$self->_stop_fetch_loop;
( run in 0.901 second using v1.01-cache-2.11-cpan-b16cb0d3907 )