Authentication

PLAIN, SCRAM-SHA-256, SCRAM-SHA-512 and OAUTHBEARER, same fields on KafkaConfig and KafkaConsumerConfig. A wrong password or rejected token raises KafkaConnectionError with the broker’s SASL error at construction.

With several bootstrap_servers, construction tries each in turn and uses the first that accepts, each getting an equal share of the bootstrap timeout (request_timeout_ms for a producer, session_timeout_ms for a consumer). A broker that rejects the credentials ends the attempt at once (SASL authentication failed). If every address fails otherwise, the error names each one.

A listener that closes the connection during the handshake usually speaks another protocol than the client, such as an SSL client on a PLAINTEXT port. The error then says to check security_protocol (security.protocol in the confluent shim). A PLAINTEXT client on a SASL_PLAINTEXT listener gets no such hint.

Username and password

from pyrula.kafka import KafkaConfig

cfg = KafkaConfig(
    bootstrap_servers="broker:9093",
    security_protocol="SASL_SSL",
    sasl_mechanism="SCRAM-SHA-512",   # or "PLAIN", "SCRAM-SHA-256"
    sasl_username="svc-orders",
    sasl_password="...",
)

SASL_PLAINTEXT sends credentials in the clear; use SASL_SSL off a trusted network.

OAUTHBEARER

Supply a callback that returns a token and its expiry. The runtime caches the token and calls back only near expiry, however many connections it opens.

cfg = KafkaConfig(
    bootstrap_servers="broker:9092",
    security_protocol="SASL_SSL",
    sasl_mechanism="OAUTHBEARER",
    oauth_cb=my_token_source,   # () -> (token: str, expiry_epoch_secs: float)
)

A provider that needs SASL extensions (RFC 7628), such as Confluent Cloud’s logicalCluster and identityPoolId, returns confluent-kafka’s (token, expiry_epoch_secs, principal, extensions) with extensions a dict[str, str]; the three-tuple is accepted too. The principal is not sent, as in librdkafka. Keys must be ASCII letters (not auth) and values visible ASCII, space, tab, CR or LF, or empty; anything else fails authentication with the reason.

Tokens refresh within 30 seconds (or 20 percent of remaining lifetime) of expiry. There is no mid-connection re-authentication: the broker closes a connection whose token expired and the reconnect picks up a fresh one.

OIDC client credentials

For Confluent Cloud, Azure Event Hubs or Keycloak, make_oidc_cb does the client-credentials grant against your token endpoint (stdlib only, 10 s HTTP timeout):

from pyrula.kafka.oauth import make_oidc_cb

cfg = KafkaConfig(
    bootstrap_servers="pkc-xxxxx.us-east-1.aws.confluent.cloud:9092",
    security_protocol="SASL_SSL",
    sasl_mechanism="OAUTHBEARER",
    oauth_cb=make_oidc_cb(
        token_url="https://login.example.com/oauth2/token",
        client_id="my-client",
        client_secret="...",
        scope="kafka",          # optional
    ),
)

AWS MSK IAM

make_msk_iam_cb wraps the AWS SigV4 signer. Install "pyrula[kafka,msk]"; credentials resolve through the boto3 chain.

from pyrula.kafka.oauth import make_msk_iam_cb

cfg = KafkaConfig(
    bootstrap_servers="b-1.cluster.xxxxx.kafka.us-east-1.amazonaws.com:9098",
    security_protocol="SASL_SSL",
    sasl_mechanism="OAUTHBEARER",
    oauth_cb=make_msk_iam_cb(region="us-east-1"),
)

TLS

ssl_ca_location overrides the system CA bundle. mTLS uses ssl_certificate_location and ssl_key_location, set together.

cfg = KafkaConfig(
    bootstrap_servers="broker:9094",
    security_protocol="SSL",
    ssl_ca_location="/etc/pki/kafka-ca.pem",
    ssl_certificate_location="/etc/pki/client.pem",
    ssl_key_location="/etc/pki/client.key",
)

Migrating from confluent-kafka

pyrula.kafka.compat.confluent translates sasl.* keys, including the oauth_cb(config_str) convention and its four-tuple answer. librdkafka’s built-in token retriever (sasl.oauthbearer.config alone) has no equivalent and is rejected with a pointer to the helpers above.