view release on metacpan or  search on metacpan

eg/tls_sasl.pl  view on Meta::CPAN

#!/usr/bin/env perl
# TLS + SASL/SCRAM-SHA-256 example.
#
# Set KAFKA_BROKER to a TLS listener (e.g. host:9093). Provide the CA cert
# in KAFKA_TLS_CA, and credentials via KAFKA_USER and KAFKA_PASS. The
# client warns at construction if SASL is configured without TLS.
#
# Quick local test with Redpanda:
#   docker run -e RP_BOOTSTRAP_USER=admin:secret123 \
#     redpandadata/redpanda:latest start --kafka-addr=...

use strict;
use warnings;
use EV;
use EV::Kafka;

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

    $cfg->{producer_epoch} = -1;
    $cfg->{rr_counter} = 0;

    # 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 Scalar::Util 'weaken';