KafkaStreamSrcConn

class

class KafkaStreamSrcConn(
    servers: KafkaServersSpec,
    credentials: KafkaCredentialsSpec | None = None,
    group_id: KafkaGroupSpec,
    schema_registry: SchemaRegistrySpec | None = None,
    conn_cfg: CfgSpec | None = None,
)

Bases: Conn

Categories: source

Kafka connection used by KafkaStreamSrc.

Registered on a collection; the runtime resolves it and hands it to the source at run time. servers and group_id accept a plain string or a Secret.

Examples

A connection with a Confluent schema registry for SR-aware serdes:

KafkaStreamSrcConn(
    servers="broker:9092",
    group_id="ingest",
    schema_registry=ConfluentSchemaRegistry(
        url="http://registry:8081",
    ),
)

Parameters

parameter

serversKafkaServersSpec (str | Secret)

Bootstrap-server list as a single comma-separated string of host[:port] items (any non-empty value), or a Secret that resolves to one.

parameter

credentialsKafkaCredentialsSpec | None (UserPassword | None)

Optional UserPassword for SASL/SCRAM auth; omit for an unauthenticated cluster.

parameter

group_idKafkaGroupSpec (str | Secret)

Consumer-group id used by the polling consumer (any non-empty value). Required.

parameter

schema_registrySchemaRegistrySpec | None (ConfluentSchemaRegistry | AwsGlueSchemaRegistry | None)

Optional ConfluentSchemaRegistry or AwsGlueSchemaRegistry; omit for schema-less ingest.

parameter

conn_cfgCfgSpec | None (dict | None)

Optional connector config forwarded as consumer configs to confluent_kafka.Consumer (see CfgSpec). The connector sets auto.offset.reset=earliest and enable.auto.commit=false itself, and defaults partition.assignment.strategy=cooperative-sticky, topic.metadata.refresh.interval.ms=30000, fetch.queue.backoff.ms=0 and socket.keepalive.enable=true; a value given here wins over each default. Settings worth knowing: max.poll.interval.ms (raise it when a stage takes longer than its 5 minute default; the stream warns past half of it and restarts itself if evicted), queued.max.messages.kbytes (memory the fetch queue may hold), fetch.min.bytes / fetch.wait.max.ms (latency against throughput) and fetch.message.max.bytes.