kafka

Kafka 🔗

Kafka is the root namespace for every exported function in this module. All cluster constructors and security helpers hang off *Kafka so the generated Dagger SDK surfaces them under dag.Kafka().<Func>(...).

apacheCluster() 🔗

ApacheCluster spins up a KRaft Kafka cluster of the requested size with dedicated controller and broker containers, using the apache/kafka JVM image.

Identical in topology, caching, and security semantics to ApacheNativeCluster — only the image differs. The JVM image runs the same Scala wrapper but on HotSpot, so it does not share apache/kafka-native’s AOT-compiled getpwuid substitution (Pwd.getpwuid from SystemPropertiesSupport.userHomeValue) that has been observed to segfault during broker startup — see Dagger Cloud trace 377f2e176c4f0e9844cb7f958c1e911b. Prefer this constructor whenever startup robustness matters more than cold-start latency.

Return Type

Cluster !

Arguments

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaServerSecurity) *dagger.KafkaCluster  {
	return dag.
			Kafka().
			Apachecluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
}content_copy Copied!

@function
def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaServerSecurity) -> dagger.KafkaCluster:
	return (
		dag.kafka()
		.apachecluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
	)content_copy Copied!

@func()
example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaServerSecurity): KafkaCluster {
	return dag
		.kafka()
		.apacheCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
}content_copy Copied!

apacheNativeCluster() 🔗

ApacheNativeCluster spins up a KRaft Kafka cluster of the requested size with dedicated controller and broker containers, using the apache/kafka-native GraalVM-compiled image.

Topology: a single controller forms a one-node KRaft quorum; one or more brokers connect to it and discover each other over the engine’s session-wide DNS — no broker-to-broker WithServiceBinding needed.

Multi-controller (controllers > 1) is rejected for now: a true HA quorum needs every controller to know every other controller at static config time, which Dagger’s WithServiceBinding model can’t express without an unresolvable cycle. TLS / mTLS and multi-controller both land in a follow-up.

Session-cached so that repeated chained method calls on the returned cluster (Client.Produce → Consume → ListTopics) all observe the SAME underlying broker services. The internal CA + per-node leaves are minted with fresh random material that we can’t make content-addressable, so a `(with a brand-new CA the previous invocation’s franz-go client doesn’t trust) every time the test calls another method on the chain.

The GraalVM-compiled image has been observed to flake during the broker setup step under load — see Dagger Cloud trace 377f2e176c4f0e9844cb7f958c1e911b. If you need the JVM image instead, use ApacheCluster().

Return Type

Cluster !

Arguments

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaServerSecurity) *dagger.KafkaCluster  {
	return dag.
			Kafka().
			Apachenativecluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
}content_copy Copied!

@function
def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaServerSecurity) -> dagger.KafkaCluster:
	return (
		dag.kafka()
		.apachenativecluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
	)content_copy Copied!

@func()
example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaServerSecurity): KafkaCluster {
	return dag
		.kafka()
		.apacheNativeCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
}content_copy Copied!

apicurioSchemaRegistry() 🔗

ApicurioSchemaRegistry spins up an Apicurio Registry service (apicurio/apicurio-registry-kafkasql) alongside the given Kafka cluster. Apicurio stores its data in a Kafka topic of its own and exposes a Confluent-Schema-Registry-compatible REST API under /apis/ccompat/v7, so the same *SchemaRegistryClient that drives ConfluentSchemaRegistry works against it unchanged — the CSR-compat prefix is folded into BasePath.

Apicurio is a more permissively licensed alternative to cp-schema-registry with a broader native artifact-type catalogue (Avro, JSON Schema, Protobuf, OpenAPI, AsyncAPI, GraphQL, WSDL, XSD); over the CSR-compat surface only the AVRO / JSON / PROTOBUF subset is reachable.

security must match the backing cluster’s mode: a PLAINTEXT profile keeps the REST endpoint on HTTP and the kafkasql connection unencrypted; a TLS / mTLS profile terminates HTTPS on the REST endpoint and secures the kafkasql connection against a matching-mode cluster.

Session-cached for the same reason ConfluentSchemaRegistry is — a `for every chained client call.

Return Type

SchemaRegistry !

Arguments

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(cluster *dagger.KafkaCluster, registry string, tag string, security *dagger.KafkaSchemaRegistrySecurity) *dagger.KafkaSchemaRegistry  {
	return dag.
			Kafka().
			Apicurioschemaregistry(cluster, registry, tag, security)
}content_copy Copied!

@function
def example(cluster: dagger.KafkaCluster, registry: str, tag: str, security: dagger.KafkaSchemaRegistrySecurity) -> dagger.KafkaSchemaRegistry:
	return (
		dag.kafka()
		.apicurioschemaregistry(cluster, registry, tag, security)
	)content_copy Copied!

@func()
example(cluster: KafkaCluster, registry: string, tag: string, security: KafkaSchemaRegistrySecurity): KafkaSchemaRegistry {
	return dag
		.kafka()
		.apicurioSchemaRegistry(cluster, registry, tag, security)
}content_copy Copied!

client() 🔗

Client constructs a franz-go-backed Kafka client that targets the given bootstrap servers. No I/O happens at construction time.

Return Type

Client !

Arguments

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(bootstrapServers []string, security *dagger.KafkaClientSecurity) *dagger.KafkaClient  {
	return dag.
			Kafka().
			Client(bootstrapServers, security)
}content_copy Copied!

@function
def example(bootstrapservers: List[str], security: dagger.KafkaClientSecurity) -> dagger.KafkaClient:
	return (
		dag.kafka()
		.client(bootstrapservers, security)
	)content_copy Copied!

@func()
example(bootstrapServers: string[], security: KafkaClientSecurity): KafkaClient {
	return dag
		.kafka()
		.client(bootstrapServers, security)
}content_copy Copied!

confluentCluster() 🔗

ConfluentCluster spins up a KRaft Kafka cluster of the requested size using the confluentinc/cp-kafka image — the Confluent Platform distribution. Confluent Platform 8.x bundles Apache Kafka 4.x (CP 8.2.0 ships Kafka 4.2.0), and cp-kafka speaks the same Scala-wrapper KAFKA_* env-var contract that ApacheCluster does, so the returned *Cluster and ServerSecurity API are identical to the Apache constructors — callers swap distros by changing the constructor name alone.

The constructor silently disables Confluent’s phone-home telemetry (KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE=false) on every broker so the cluster behaves the same way the Apache variants do at startup.

Return Type

Cluster !

Arguments

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaServerSecurity) *dagger.KafkaCluster  {
	return dag.
			Kafka().
			Confluentcluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
}content_copy Copied!

@function
def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaServerSecurity) -> dagger.KafkaCluster:
	return (
		dag.kafka()
		.confluentcluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
	)content_copy Copied!

@func()
example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaServerSecurity): KafkaCluster {
	return dag
		.kafka()
		.confluentCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
}content_copy Copied!

confluentSchemaRegistry() 🔗

ConfluentSchemaRegistry spins up a Confluent Schema Registry service (confluentinc/cp-schema-registry) alongside the given Kafka cluster. The registry talks the Kafka wire protocol to the cluster’s brokers for its _schemas topic and exposes its own REST API on top, so it composes on any *Cluster regardless of distro — cp-schema-registry simply pairs most naturally with a cp-kafka ConfluentCluster.

Only PLAINTEXT clusters are supported in this story: the constructor rejects TLS / mTLS clusters and points callers at the TLS follow-up.

Session-cached for the same reason the cluster constructors are — a `for every chained client call.

Return Type

SchemaRegistry !

Arguments

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(cluster *dagger.KafkaCluster, registry string, tag string, security *dagger.KafkaSchemaRegistrySecurity) *dagger.KafkaSchemaRegistry  {
	return dag.
			Kafka().
			Confluentschemaregistry(cluster, registry, tag, security)
}content_copy Copied!

@function
def example(cluster: dagger.KafkaCluster, registry: str, tag: str, security: dagger.KafkaSchemaRegistrySecurity) -> dagger.KafkaSchemaRegistry:
	return (
		dag.kafka()
		.confluentschemaregistry(cluster, registry, tag, security)
	)content_copy Copied!

@func()
example(cluster: KafkaCluster, registry: string, tag: string, security: KafkaSchemaRegistrySecurity): KafkaSchemaRegistry {
	return dag
		.kafka()
		.confluentSchemaRegistry(cluster, registry, tag, security)
}content_copy Copied!

karapaceSchemaRegistry() 🔗

KarapaceSchemaRegistry spins up a Karapace service (ghcr.io/aiven-open/karapace) alongside the given Kafka cluster. Karapace is Aiven’s drop-in Python reimplementation of the Confluent Schema Registry: it talks the Kafka wire protocol to the cluster’s brokers for its _schemas topic and serves a Confluent-Schema-Registry-compatible REST API at the root, so the same *SchemaRegistryClient that drives ConfluentSchemaRegistry works against it unchanged (BasePath stays empty).

Unlike the other registry constructors, registry defaults to ghcr.io: Karapace publishes to GitHub Container Registry rather than Docker Hub, which also keeps CI clear of Docker Hub rate limits and Confluent’s image licensing.

security must match the backing cluster’s mode. Karapace consumes PEM (not PKCS#12) for its own listener and aiokafka storage; the module extracts PEM from the supplied CA internally, so callers pass the same PKCS#12 profile shape as the other registries.

Session-cached for the same reason ConfluentSchemaRegistry is — a `for every chained client call.

Return Type

SchemaRegistry !

Arguments

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(cluster *dagger.KafkaCluster, registry string, tag string, security *dagger.KafkaSchemaRegistrySecurity) *dagger.KafkaSchemaRegistry  {
	return dag.
			Kafka().
			Karapaceschemaregistry(cluster, registry, tag, security)
}content_copy Copied!

@function
def example(cluster: dagger.KafkaCluster, registry: str, tag: str, security: dagger.KafkaSchemaRegistrySecurity) -> dagger.KafkaSchemaRegistry:
	return (
		dag.kafka()
		.karapaceschemaregistry(cluster, registry, tag, security)
	)content_copy Copied!

@func()
example(cluster: KafkaCluster, registry: string, tag: string, security: KafkaSchemaRegistrySecurity): KafkaSchemaRegistry {
	return dag
		.kafka()
		.karapaceSchemaRegistry(cluster, registry, tag, security)
}content_copy Copied!

mtlsClientSecurity() 🔗

MtlsClientSecurity returns a ClientSecurity profile that opens an mTLS connection: the broker presents its server cert (verified against trustStore) and the client presents its own leaf cert from keyStore (signed by a CA the broker trusts via its clientTrustStore).

Return Type

ClientSecurity !

Arguments

dagger -m github.com/z5labs/devex/daggerverse/kafka@fcc487e54d35bc7a879dd2f943855bf506d38965 call \
 mtls-client-security --key-store file:path --key-store-password env:MYSECRET --trust-store file:path --trust-store-password env:MYSECRETcontent_copy Copied! 

func (m *MyModule) Example(keyStore *dagger.File, keyStorePassword *dagger.Secret, trustStore *dagger.File, trustStorePassword *dagger.Secret) *dagger.KafkaClientSecurity  {
	return dag.
			Kafka().
			Mtlsclientsecurity(keyStore, keyStorePassword, trustStore, trustStorePassword)
}content_copy Copied!

@function
def example(keystore: dagger.File, keystorepassword: dagger.Secret, truststore: dagger.File, truststorepassword: dagger.Secret) -> dagger.KafkaClientSecurity:
	return (
		dag.kafka()
		.mtlsclientsecurity(keystore, keystorepassword, truststore, truststorepassword)
	)content_copy Copied!

@func()
example(keyStore: File, keyStorePassword: Secret, trustStore: File, trustStorePassword: Secret): KafkaClientSecurity {
	return dag
		.kafka()
		.mtlsClientSecurity(keyStore, keyStorePassword, trustStore, trustStorePassword)
}content_copy Copied!

mtlsSchemaRegistryClientSecurity() 🔗

MtlsSchemaRegistryClientSecurity returns a SchemaRegistryClientSecurity profile that opens an mTLS HTTPS connection: the registry presents its REST server cert (verified against trustStore) and the client presents its own leaf cert from keyStore (signed by a CA the registry trusts via its clientTrustStore).

Return Type

SchemaRegistryClientSecurity !

Arguments

dagger -m github.com/z5labs/devex/daggerverse/kafka@fcc487e54d35bc7a879dd2f943855bf506d38965 call \
 mtls-schema-registry-client-security --key-store file:path --key-store-password env:MYSECRET --trust-store file:path --trust-store-password env:MYSECRETcontent_copy Copied! 

func (m *MyModule) Example(keyStore *dagger.File, keyStorePassword *dagger.Secret, trustStore *dagger.File, trustStorePassword *dagger.Secret) *dagger.KafkaSchemaRegistryClientSecurity  {
	return dag.
			Kafka().
			Mtlsschemaregistryclientsecurity(keyStore, keyStorePassword, trustStore, trustStorePassword)
}content_copy Copied!

@function
def example(keystore: dagger.File, keystorepassword: dagger.Secret, truststore: dagger.File, truststorepassword: dagger.Secret) -> dagger.KafkaSchemaRegistryClientSecurity:
	return (
		dag.kafka()
		.mtlsschemaregistryclientsecurity(keystore, keystorepassword, truststore, truststorepassword)
	)content_copy Copied!

@func()
example(keyStore: File, keyStorePassword: Secret, trustStore: File, trustStorePassword: Secret): KafkaSchemaRegistryClientSecurity {
	return dag
		.kafka()
		.mtlsSchemaRegistryClientSecurity(keyStore, keyStorePassword, trustStore, trustStorePassword)
}content_copy Copied!

mtlsSchemaRegistrySecurity() 🔗

MtlsSchemaRegistrySecurity returns a SchemaRegistrySecurity profile that terminates mTLS on the registry REST endpoint. caKeyStore signs the registry’s server leaf (and the registry’s own client leaf presented to the broker over mTLS); clientTrustStore holds the CA(s) the registry accepts incoming REST client certs from. Pair with an MTLS cluster minted from the same CA.

Return Type

SchemaRegistrySecurity !

Arguments

dagger -m github.com/z5labs/devex/daggerverse/kafka@fcc487e54d35bc7a879dd2f943855bf506d38965 call \
 mtls-schema-registry-security --ca-key-store file:path --ca-key-store-password env:MYSECRET --client-trust-store file:path --client-trust-store-password env:MYSECRETcontent_copy Copied! 

func (m *MyModule) Example(caKeyStore *dagger.File, caKeyStorePassword *dagger.Secret, clientTrustStore *dagger.File, clientTrustStorePassword *dagger.Secret) *dagger.KafkaSchemaRegistrySecurity  {
	return dag.
			Kafka().
			Mtlsschemaregistrysecurity(caKeyStore, caKeyStorePassword, clientTrustStore, clientTrustStorePassword)
}content_copy Copied!

@function
def example(cakeystore: dagger.File, cakeystorepassword: dagger.Secret, clienttruststore: dagger.File, clienttruststorepassword: dagger.Secret) -> dagger.KafkaSchemaRegistrySecurity:
	return (
		dag.kafka()
		.mtlsschemaregistrysecurity(cakeystore, cakeystorepassword, clienttruststore, clienttruststorepassword)
	)content_copy Copied!

@func()
example(caKeyStore: File, caKeyStorePassword: Secret, clientTrustStore: File, clientTrustStorePassword: Secret): KafkaSchemaRegistrySecurity {
	return dag
		.kafka()
		.mtlsSchemaRegistrySecurity(caKeyStore, caKeyStorePassword, clientTrustStore, clientTrustStorePassword)
}content_copy Copied!

mtlsServerSecurity() 🔗

MtlsServerSecurity returns a ServerSecurity profile that terminates mTLS on the external listener. caKeyStore signs per-broker server leaves; clientTrustStore holds the CA(s) the broker will accept incoming client certs from (this can be the same CA as caKeyStore or an independent one for asymmetric trust).

Return Type

ServerSecurity !

Arguments

dagger -m github.com/z5labs/devex/daggerverse/kafka@fcc487e54d35bc7a879dd2f943855bf506d38965 call \
 mtls-server-security --ca-key-store file:path --ca-key-store-password env:MYSECRET --client-trust-store file:path --client-trust-store-password env:MYSECRETcontent_copy Copied! 

func (m *MyModule) Example(caKeyStore *dagger.File, caKeyStorePassword *dagger.Secret, clientTrustStore *dagger.File, clientTrustStorePassword *dagger.Secret) *dagger.KafkaServerSecurity  {
	return dag.
			Kafka().
			Mtlsserversecurity(caKeyStore, caKeyStorePassword, clientTrustStore, clientTrustStorePassword)
}content_copy Copied!

@function
def example(cakeystore: dagger.File, cakeystorepassword: dagger.Secret, clienttruststore: dagger.File, clienttruststorepassword: dagger.Secret) -> dagger.KafkaServerSecurity:
	return (
		dag.kafka()
		.mtlsserversecurity(cakeystore, cakeystorepassword, clienttruststore, clienttruststorepassword)
	)content_copy Copied!

@func()
example(caKeyStore: File, caKeyStorePassword: Secret, clientTrustStore: File, clientTrustStorePassword: Secret): KafkaServerSecurity {
	return dag
		.kafka()
		.mtlsServerSecurity(caKeyStore, caKeyStorePassword, clientTrustStore, clientTrustStorePassword)
}content_copy Copied!

plaintextClientSecurity() 🔗

PlaintextClientSecurity returns a ClientSecurity profile configured for unencrypted, unauthenticated traffic.

Return Type

ClientSecurity !

dagger -m github.com/z5labs/devex/daggerverse/kafka@fcc487e54d35bc7a879dd2f943855bf506d38965 call \
 plaintext-client-securitycontent_copy Copied! 

func (m *MyModule) Example() *dagger.KafkaClientSecurity  {
	return dag.
			Kafka().
			Plaintextclientsecurity()
}content_copy Copied!

@function
def example() -> dagger.KafkaClientSecurity:
	return (
		dag.kafka()
		.plaintextclientsecurity()
	)content_copy Copied!

@func()
example(): KafkaClientSecurity {
	return dag
		.kafka()
		.plaintextClientSecurity()
}content_copy Copied!

plaintextSchemaRegistryClientSecurity() 🔗

PlaintextSchemaRegistryClientSecurity returns a SchemaRegistryClientSecurity profile configured for unencrypted HTTP traffic.

Return Type

SchemaRegistryClientSecurity !

dagger -m github.com/z5labs/devex/daggerverse/kafka@fcc487e54d35bc7a879dd2f943855bf506d38965 call \
 plaintext-schema-registry-client-securitycontent_copy Copied! 

func (m *MyModule) Example() *dagger.KafkaSchemaRegistryClientSecurity  {
	return dag.
			Kafka().
			Plaintextschemaregistryclientsecurity()
}content_copy Copied!

@function
def example() -> dagger.KafkaSchemaRegistryClientSecurity:
	return (
		dag.kafka()
		.plaintextschemaregistryclientsecurity()
	)content_copy Copied!

@func()
example(): KafkaSchemaRegistryClientSecurity {
	return dag
		.kafka()
		.plaintextSchemaRegistryClientSecurity()
}content_copy Copied!

plaintextSchemaRegistrySecurity() 🔗

PlaintextSchemaRegistrySecurity returns a SchemaRegistrySecurity profile configured for unencrypted, unauthenticated traffic on the registry REST endpoint. It pairs with a PLAINTEXT cluster.

Return Type

SchemaRegistrySecurity !

dagger -m github.com/z5labs/devex/daggerverse/kafka@fcc487e54d35bc7a879dd2f943855bf506d38965 call \
 plaintext-schema-registry-securitycontent_copy Copied! 

func (m *MyModule) Example() *dagger.KafkaSchemaRegistrySecurity  {
	return dag.
			Kafka().
			Plaintextschemaregistrysecurity()
}content_copy Copied!

@function
def example() -> dagger.KafkaSchemaRegistrySecurity:
	return (
		dag.kafka()
		.plaintextschemaregistrysecurity()
	)content_copy Copied!

@func()
example(): KafkaSchemaRegistrySecurity {
	return dag
		.kafka()
		.plaintextSchemaRegistrySecurity()
}content_copy Copied!

plaintextServerSecurity() 🔗

PlaintextServerSecurity returns a ServerSecurity profile configured for unencrypted, unauthenticated traffic on the external listener. Internal listeners (inter-broker + controller-quorum) still use mTLS.

Return Type

ServerSecurity !

dagger -m github.com/z5labs/devex/daggerverse/kafka@fcc487e54d35bc7a879dd2f943855bf506d38965 call \
 plaintext-server-securitycontent_copy Copied! 

func (m *MyModule) Example() *dagger.KafkaServerSecurity  {
	return dag.
			Kafka().
			Plaintextserversecurity()
}content_copy Copied!

@function
def example() -> dagger.KafkaServerSecurity:
	return (
		dag.kafka()
		.plaintextserversecurity()
	)content_copy Copied!

@func()
example(): KafkaServerSecurity {
	return dag
		.kafka()
		.plaintextServerSecurity()
}content_copy Copied!

redpandaCluster() 🔗

RedpandaCluster spins up a single-node Redpanda cluster using the redpandadata/redpanda image. Redpanda runs broker and Raft duties in the same process, so there is no separate controller container.

Multi-node (controllers != 1 or brokers != 1) is rejected — multi-broker Redpanda needs --seeds plumbing + per-node rpc_server advertising that doesn’t fit single-story scope. The wire protocol matches Kafka, so RedpandaCluster.Client() returns the same *Client type the Apache constructors return.

Return Type

RedpandaCluster !

Arguments

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaRedpandaServerSecurity) *dagger.KafkaRedpandaCluster  {
	return dag.
			Kafka().
			Redpandacluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
}content_copy Copied!

@function
def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaRedpandaServerSecurity) -> dagger.KafkaRedpandaCluster:
	return (
		dag.kafka()
		.redpandacluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
	)content_copy Copied!

@func()
example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaRedpandaServerSecurity): KafkaRedpandaCluster {
	return dag
		.kafka()
		.redpandaCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
}content_copy Copied!

redpandaPlaintextServerSecurity() 🔗

RedpandaPlaintextServerSecurity returns a RedpandaServerSecurity profile configured for unencrypted, unauthenticated traffic on the external Kafka listener.

Return Type

RedpandaServerSecurity !

dagger -m github.com/z5labs/devex/daggerverse/kafka@fcc487e54d35bc7a879dd2f943855bf506d38965 call \
 redpanda-plaintext-server-securitycontent_copy Copied! 

func (m *MyModule) Example() *dagger.KafkaRedpandaServerSecurity  {
	return dag.
			Kafka().
			Redpandaplaintextserversecurity()
}content_copy Copied!

@function
def example() -> dagger.KafkaRedpandaServerSecurity:
	return (
		dag.kafka()
		.redpandaplaintextserversecurity()
	)content_copy Copied!

@func()
example(): KafkaRedpandaServerSecurity {
	return dag
		.kafka()
		.redpandaPlaintextServerSecurity()
}content_copy Copied!

redpandaTlsServerSecurity() 🔗

RedpandaTlsServerSecurity returns a RedpandaServerSecurity profile that terminates TLS on the external Kafka listener. caKeyStore is a PKCS#12 archive of the CA cert + private key used to mint the per-cluster server leaf — same shape as Kafka.TlsServerSecurity, so callers don’t have to convert between formats even though Redpanda itself reads PEM internally. The leaf carries the broker’s stable hostname as a DNS SAN so franz-go clients dialing the bootstrap address can verify the cert against the matching truststore.

Return Type

RedpandaServerSecurity !

Arguments

dagger -m github.com/z5labs/devex/daggerverse/kafka@fcc487e54d35bc7a879dd2f943855bf506d38965 call \
 redpanda-tls-server-security --ca-key-store file:path --ca-key-store-password env:MYSECRETcontent_copy Copied! 

func (m *MyModule) Example(caKeyStore *dagger.File, caKeyStorePassword *dagger.Secret) *dagger.KafkaRedpandaServerSecurity  {
	return dag.
			Kafka().
			Redpandatlsserversecurity(caKeyStore, caKeyStorePassword)
}content_copy Copied!

@function
def example(cakeystore: dagger.File, cakeystorepassword: dagger.Secret) -> dagger.KafkaRedpandaServerSecurity:
	return (
		dag.kafka()
		.redpandatlsserversecurity(cakeystore, cakeystorepassword)
	)content_copy Copied!

@func()
example(caKeyStore: File, caKeyStorePassword: Secret): KafkaRedpandaServerSecurity {
	return dag
		.kafka()
		.redpandaTlsServerSecurity(caKeyStore, caKeyStorePassword)
}content_copy Copied!

tlsClientSecurity() 🔗

TlsClientSecurity returns a ClientSecurity profile that opens a TLS connection to the broker. trustStore is a PKCS#12 archive of the CA(s) the client uses to verify the broker’s leaf certificate (typically the truststore that pairs with the CA passed to TlsServerSecurity on the server side).

Return Type

ClientSecurity !

Arguments

dagger -m github.com/z5labs/devex/daggerverse/kafka@fcc487e54d35bc7a879dd2f943855bf506d38965 call \
 tls-client-security --trust-store file:path --trust-store-password env:MYSECRETcontent_copy Copied! 

func (m *MyModule) Example(trustStore *dagger.File, trustStorePassword *dagger.Secret) *dagger.KafkaClientSecurity  {
	return dag.
			Kafka().
			Tlsclientsecurity(trustStore, trustStorePassword)
}content_copy Copied!

@function
def example(truststore: dagger.File, truststorepassword: dagger.Secret) -> dagger.KafkaClientSecurity:
	return (
		dag.kafka()
		.tlsclientsecurity(truststore, truststorepassword)
	)content_copy Copied!

@func()
example(trustStore: File, trustStorePassword: Secret): KafkaClientSecurity {
	return dag
		.kafka()
		.tlsClientSecurity(trustStore, trustStorePassword)
}content_copy Copied!

tlsSchemaRegistryClientSecurity() 🔗

TlsSchemaRegistryClientSecurity returns a SchemaRegistryClientSecurity profile that opens an HTTPS connection to the registry. trustStore is a PKCS#12 archive of the CA(s) the client uses to verify the registry’s REST leaf certificate (typically the truststore that pairs with the CA passed to TlsSchemaRegistrySecurity on the server side).

Return Type

SchemaRegistryClientSecurity !

Arguments

dagger -m github.com/z5labs/devex/daggerverse/kafka@fcc487e54d35bc7a879dd2f943855bf506d38965 call \
 tls-schema-registry-client-security --trust-store file:path --trust-store-password env:MYSECRETcontent_copy Copied! 

func (m *MyModule) Example(trustStore *dagger.File, trustStorePassword *dagger.Secret) *dagger.KafkaSchemaRegistryClientSecurity  {
	return dag.
			Kafka().
			Tlsschemaregistryclientsecurity(trustStore, trustStorePassword)
}content_copy Copied!

@function
def example(truststore: dagger.File, truststorepassword: dagger.Secret) -> dagger.KafkaSchemaRegistryClientSecurity:
	return (
		dag.kafka()
		.tlsschemaregistryclientsecurity(truststore, truststorepassword)
	)content_copy Copied!

@func()
example(trustStore: File, trustStorePassword: Secret): KafkaSchemaRegistryClientSecurity {
	return dag
		.kafka()
		.tlsSchemaRegistryClientSecurity(trustStore, trustStorePassword)
}content_copy Copied!

tlsSchemaRegistrySecurity() 🔗

TlsSchemaRegistrySecurity returns a SchemaRegistrySecurity profile that terminates TLS on the registry REST endpoint. caKeyStore is a PKCS#12 archive containing the CA cert + private key the registry uses to mint its per-registry server leaf (bound to the registry’s service hostname as a DNS SAN) and to derive the truststore its kafka-storage connection uses to verify the backing broker. Pair with a TLS cluster minted from the same CA.

Return Type

SchemaRegistrySecurity !

Arguments

dagger -m github.com/z5labs/devex/daggerverse/kafka@fcc487e54d35bc7a879dd2f943855bf506d38965 call \
 tls-schema-registry-security --ca-key-store file:path --ca-key-store-password env:MYSECRETcontent_copy Copied! 

func (m *MyModule) Example(caKeyStore *dagger.File, caKeyStorePassword *dagger.Secret) *dagger.KafkaSchemaRegistrySecurity  {
	return dag.
			Kafka().
			Tlsschemaregistrysecurity(caKeyStore, caKeyStorePassword)
}content_copy Copied!

@function
def example(cakeystore: dagger.File, cakeystorepassword: dagger.Secret) -> dagger.KafkaSchemaRegistrySecurity:
	return (
		dag.kafka()
		.tlsschemaregistrysecurity(cakeystore, cakeystorepassword)
	)content_copy Copied!

@func()
example(caKeyStore: File, caKeyStorePassword: Secret): KafkaSchemaRegistrySecurity {
	return dag
		.kafka()
		.tlsSchemaRegistrySecurity(caKeyStore, caKeyStorePassword)
}content_copy Copied!

tlsServerSecurity() 🔗

TlsServerSecurity returns a ServerSecurity profile that terminates TLS on the external listener. caKeyStore is a PKCS#12 archive containing the CA cert + private key the cluster uses to mint per-broker leaf certs; each broker leaf carries its stable hostname (e.g. “broker-100”) as a DNS SAN so franz-go clients dialing the bootstrap address can verify the broker against the same CA’s truststore.

Return Type

ServerSecurity !

Arguments

dagger -m github.com/z5labs/devex/daggerverse/kafka@fcc487e54d35bc7a879dd2f943855bf506d38965 call \
 tls-server-security --ca-key-store file:path --ca-key-store-password env:MYSECRETcontent_copy Copied! 

func (m *MyModule) Example(caKeyStore *dagger.File, caKeyStorePassword *dagger.Secret) *dagger.KafkaServerSecurity  {
	return dag.
			Kafka().
			Tlsserversecurity(caKeyStore, caKeyStorePassword)
}content_copy Copied!

@function
def example(cakeystore: dagger.File, cakeystorepassword: dagger.Secret) -> dagger.KafkaServerSecurity:
	return (
		dag.kafka()
		.tlsserversecurity(cakeystore, cakeystorepassword)
	)content_copy Copied!

@func()
example(caKeyStore: File, caKeyStorePassword: Secret): KafkaServerSecurity {
	return dag
		.kafka()
		.tlsServerSecurity(caKeyStore, caKeyStorePassword)
}content_copy Copied!

Cluster 🔗

Cluster represents a running KRaft Kafka cluster, holding references to every broker service so callers can bind them into their own containers or open a franz-go Client against them.

bindBrokers() 🔗

BindBrokers attaches every broker service to the given container under the same hostname BootstrapServers reports, so the container can dial brokers using the same address strings as a franz-go Client returned from Cluster.Client.

Return Type

Container !

Arguments

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaServerSecurity, ctr *dagger.Container) *dagger.Container  {
	return dag.
			Kafka().
			Confluentcluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity).
			Bindbrokers(ctr)
}content_copy Copied!

@function
def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaServerSecurity, ctr: dagger.Container) -> dagger.Container:
	return (
		dag.kafka()
		.confluentcluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
		.bindbrokers(ctr)
	)content_copy Copied!

@func()
example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaServerSecurity, ctr: Container): Container {
	return dag
		.kafka()
		.confluentCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
		.bindBrokers(ctr)
}content_copy Copied!

bootstrapServers() 🔗

BootstrapServers returns the host:port pairs each broker advertises on its client-facing listener.

Return Type

[String ! ] !

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(ctx context.Context, clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaServerSecurity) []string  {
	return dag.
			Kafka().
			Confluentcluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity).
			Bootstrapservers(ctx)
}content_copy Copied!

@function
async def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaServerSecurity) -> List[str]:
	return await (
		dag.kafka()
		.confluentcluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
		.bootstrapservers()
	)content_copy Copied!

@func()
async example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaServerSecurity): Promise<string[]> {
	return dag
		.kafka()
		.confluentCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
		.bootstrapServers()
}content_copy Copied!

client() 🔗

Client starts every broker service in the cluster and returns a franz-go Client wired with their bootstrap addresses.

Return Type

Client !

Arguments

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaServerSecurity, security *dagger.KafkaClientSecurity) *dagger.KafkaClient  {
	return dag.
			Kafka().
			Confluentcluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity).
			Client(security)
}content_copy Copied!

@function
def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaServerSecurity, security: dagger.KafkaClientSecurity) -> dagger.KafkaClient:
	return (
		dag.kafka()
		.confluentcluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
		.client(security)
	)content_copy Copied!

@func()
example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaServerSecurity, security: KafkaClientSecurity): KafkaClient {
	return dag
		.kafka()
		.confluentCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
		.client(security)
}content_copy Copied!

stop() 🔗

Stop tears down every service container backing this cluster (the controller plus every broker). Tests should call this in a defer so each broker Container.asService span closes when the test work is done, rather than running out to the parent parallel group’s lifetime.

Kill is set so Service.Stop skips graceful shutdown — Kafka’s broker shutdown path waits on replica-drain timeouts that on a torn-down test cluster just run out the clock (~5 min observed in Dagger trace 972bc311bf374f817b7c88481229a10c). SIGKILL returns immediately, which is all a test needs.

Return Type

Void !

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(ctx context.Context, clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaServerSecurity)   {
	return dag.
			Kafka().
			Confluentcluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity).
			Stop(ctx)
}content_copy Copied!

@function
async def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaServerSecurity) -> None:
	return await (
		dag.kafka()
		.confluentcluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
		.stop()
	)content_copy Copied!

@func()
async example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaServerSecurity): Promise<void> {
	return dag
		.kafka()
		.confluentCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
		.stop()
}content_copy Copied!

ServerSecurity 🔗

ServerSecurity describes how a Kafka cluster’s external listener authenticates and encrypts traffic from clients. Internal listeners (inter-broker + controller-quorum) are always mTLS, regardless of mode.

SchemaRegistry 🔗

SchemaRegistry is the module’s shared Schema Registry abstraction, bound to a Kafka cluster’s brokers. It stores schemas in the cluster’s _schemas topic and exposes a REST API for registering and looking up Avro / JSON Schema / Protobuf schemas by subject.

The same type is returned both by Kafka.ConfluentSchemaRegistry — a separate cp-schema-registry container — and by RedpandaCluster.SchemaRegistry, which surfaces the Schema Registry bundled inside the Redpanda broker process. Callers treat the two uniformly; the Bundled field records which kind this is so Stop behaves correctly.

The constructor is session-cached so chained calls (Client().RegisterSchema(…) → LookupSchemaByID(…)) all observe the same underlying service.

bindTo() 🔗

BindTo attaches the Schema Registry service to the given container under the same hostname Endpoint reports, so the container resolves the registry at that address.

Return Type

Container !

Arguments

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaRedpandaServerSecurity, security *dagger.KafkaSchemaRegistrySecurity, ctr *dagger.Container) *dagger.Container  {
	return dag.
			Kafka().
			Redpandacluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity).
			Schemaregistry(security).
			Bindto(ctr)
}content_copy Copied!

@function
def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaRedpandaServerSecurity, security: dagger.KafkaSchemaRegistrySecurity, ctr: dagger.Container) -> dagger.Container:
	return (
		dag.kafka()
		.redpandacluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
		.schemaregistry(security)
		.bindto(ctr)
	)content_copy Copied!

@func()
example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaRedpandaServerSecurity, security: KafkaSchemaRegistrySecurity, ctr: Container): Container {
	return dag
		.kafka()
		.redpandaCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
		.schemaRegistry(security)
		.bindTo(ctr)
}content_copy Copied!

client() 🔗

Client returns a typed HTTP client targeting this registry’s REST API. The URL scheme (http vs https) follows the registry’s own security mode; the supplied security profile must match it (a TLS/mTLS registry needs a TLS/mTLS client profile, verified when the first request runs). No I/O happens at construction time.

Return Type

SchemaRegistryClient !

Arguments

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaRedpandaServerSecurity, security *dagger.KafkaSchemaRegistrySecurity, security1 *dagger.KafkaSchemaRegistryClientSecurity) *dagger.KafkaSchemaRegistryClient  {
	return dag.
			Kafka().
			Redpandacluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity).
			Schemaregistry(security).
			Client(security1)
}content_copy Copied!

@function
def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaRedpandaServerSecurity, security: dagger.KafkaSchemaRegistrySecurity, security1: dagger.KafkaSchemaRegistryClientSecurity) -> dagger.KafkaSchemaRegistryClient:
	return (
		dag.kafka()
		.redpandacluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
		.schemaregistry(security)
		.client(security1)
	)content_copy Copied!

@func()
example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaRedpandaServerSecurity, security: KafkaSchemaRegistrySecurity, security1: KafkaSchemaRegistryClientSecurity): KafkaSchemaRegistryClient {
	return dag
		.kafka()
		.redpandaCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
		.schemaRegistry(security)
		.client(security1)
}content_copy Copied!

endpoint() 🔗

Endpoint returns the host:port other containers (and the module runtime) can reach the Schema Registry REST API on.

Return Type

String !

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(ctx context.Context, clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaRedpandaServerSecurity, security *dagger.KafkaSchemaRegistrySecurity) string  {
	return dag.
			Kafka().
			Redpandacluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity).
			Schemaregistry(security).
			Endpoint(ctx)
}content_copy Copied!

@function
async def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaRedpandaServerSecurity, security: dagger.KafkaSchemaRegistrySecurity) -> str:
	return await (
		dag.kafka()
		.redpandacluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
		.schemaregistry(security)
		.endpoint()
	)content_copy Copied!

@func()
async example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaRedpandaServerSecurity, security: KafkaSchemaRegistrySecurity): Promise<string> {
	return dag
		.kafka()
		.redpandaCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
		.schemaRegistry(security)
		.endpoint()
}content_copy Copied!

stop() 🔗

Stop tears down the Schema Registry service. Kill is set so the stop returns immediately rather than waiting on graceful shutdown, mirroring Cluster.Stop — tests should call this in a defer.

For a bundled registry (Bundled == true) the service is shared with the owning cluster, so Stop is a no-op: the cluster owns that lifecycle and stopping it here would tear the whole cluster down. Callers that uniformly defer sr.Stop(ctx) stay safe regardless of which registry they hold.

Return Type

Void !

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(ctx context.Context, clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaRedpandaServerSecurity, security *dagger.KafkaSchemaRegistrySecurity)   {
	return dag.
			Kafka().
			Redpandacluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity).
			Schemaregistry(security).
			Stop(ctx)
}content_copy Copied!

@function
async def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaRedpandaServerSecurity, security: dagger.KafkaSchemaRegistrySecurity) -> None:
	return await (
		dag.kafka()
		.redpandacluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
		.schemaregistry(security)
		.stop()
	)content_copy Copied!

@func()
async example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaRedpandaServerSecurity, security: KafkaSchemaRegistrySecurity): Promise<void> {
	return dag
		.kafka()
		.redpandaCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
		.schemaRegistry(security)
		.stop()
}content_copy Copied!

SchemaRegistrySecurity 🔗

SchemaRegistrySecurity describes how a Schema Registry’s REST endpoint authenticates and encrypts traffic from clients, mirroring *ServerSecurity on the broker side. The same profile also parameterises the registry’s kafka-storage connection to the backing cluster: the CA it carries doubles as the trust anchor the registry uses to dial the cluster’s TLS/mTLS client listener (the single-CA convention — pass the same CA to the cluster and the registry).

Client 🔗

Client is a franz-go-backed Kafka client. Each method opens a fresh connection so the function call is stateless from Dagger’s perspective.

consume() 🔗

Consume reads up to maxMessages records from the topic, starting at the earliest offset, returning when either maxMessages have been gathered or the parsed timeout elapses. Each record’s key and value are encoded into the requested string forms before being returned. The records are returned as a JSON array string (a []ConsumedRecord marshaled with encoding/json); callers unmarshal it into their own struct — see ConsumedRecord for why.

When group is non-empty, the consume runs as a member of that consumer group: the broker assigns partitions and the join itself writes group metadata to __consumer_offsets (offsets are not committed — the function stays idempotent underdefault), partitions are consumed directly with no group state.

When schemaRegistryAware is true, each record’s key and value are inspected for the Confluent Schema Registry wire-format header (0x00 || uint32be(schemaID) || payload). When present, the 5-byte header is stripped before encoding the payload and the extracted schema ID is surfaced on ConsumedRecord.KeySchemaID / ConsumedRecord.ValueSchemaID. Unframed fields pass through with a zero schema ID. When false (the default), bytes are returned verbatim and the schema ID fields are always zero.

keyDeserializeAs / valueDeserializeAs, when set to “JSON”, validate each consumed record’s post-frame-strip payload bytes via encoding/json’s json.Valid. The pipeline order is unframe → deserialize → encode, so SchemaRegistryAware and a JSON deserializer compose: framed bytes are stripped first and validation runs on the payload alone. Records whose payloads fail to parse cause Consume to error out and abandon the remaining poll. Default “” is pass-through; “JSON” validates.

keyDeserializeAs / valueDeserializeAs set to “AVRO” decode each record’s post-frame-strip Avro binary payload (via github.com/z5labs/avro-go/generic) against the schema identified by the wire id and re-serialise it to JSON. This mode requires schemaRegistryAware=true (so the wire id is available) and a registry (so the schema text can be resolved by id); an unframed record errors out pointing at the missing wire header. Schema resolution is cached per id for the duration of the call. The JSON shape follows the Avro spec’s JSON encoding; logical types, decimal, and fixed are not yet supported.

Return Type

String !

Arguments

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(ctx context.Context, clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaRedpandaServerSecurity, security *dagger.KafkaClientSecurity, topic string, maxMessages int, timeout string, keyEncoding string, valueEncoding string, group string, schemaRegistryAware bool, keyDeserializeAs string, valueDeserializeAs string) string  {
	return dag.
			Kafka().
			Redpandacluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity).
			Client(security).
			Consume(ctx, topic, maxMessages, timeout, keyEncoding, valueEncoding, group, schemaRegistryAware, keyDeserializeAs, valueDeserializeAs)
}content_copy Copied!

@function
async def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaRedpandaServerSecurity, security: dagger.KafkaClientSecurity, topic: str, maxmessages: int, timeout: str, keyencoding: str, valueencoding: str, group: str, schemaregistryaware: bool, keydeserializeas: str, valuedeserializeas: str) -> str:
	return await (
		dag.kafka()
		.redpandacluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
		.client(security)
		.consume(topic, maxmessages, timeout, keyencoding, valueencoding, group, schemaregistryaware, keydeserializeas, valuedeserializeas)
	)content_copy Copied!

@func()
async example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaRedpandaServerSecurity, security: KafkaClientSecurity, topic: string, maxMessages: number, timeout: string, keyEncoding: string, valueEncoding: string, group: string, schemaRegistryAware: boolean, keyDeserializeAs: string, valueDeserializeAs: string): Promise<string> {
	return dag
		.kafka()
		.redpandaCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
		.client(security)
		.consume(topic, maxMessages, timeout, keyEncoding, valueEncoding, group, schemaRegistryAware, keyDeserializeAs, valueDeserializeAs)
}content_copy Copied!

createTopic() 🔗

CreateTopic creates a new topic with the given partition count and replication factor. Errors out if the topic already exists.

Return Type

Void !

Arguments

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(ctx context.Context, clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaRedpandaServerSecurity, security *dagger.KafkaClientSecurity, name string, partitions int, replicationFactor int)   {
	return dag.
			Kafka().
			Redpandacluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity).
			Client(security).
			Createtopic(ctx, name, partitions, replicationFactor)
}content_copy Copied!

@function
async def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaRedpandaServerSecurity, security: dagger.KafkaClientSecurity, name: str, partitions: int, replicationfactor: int) -> None:
	return await (
		dag.kafka()
		.redpandacluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
		.client(security)
		.createtopic(name, partitions, replicationfactor)
	)content_copy Copied!

@func()
async example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaRedpandaServerSecurity, security: KafkaClientSecurity, name: string, partitions: number, replicationFactor: number): Promise<void> {
	return dag
		.kafka()
		.redpandaCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
		.client(security)
		.createTopic(name, partitions, replicationFactor)
}content_copy Copied!

deleteTopic() 🔗

DeleteTopic deletes the named topic.

Return Type

Void !

Arguments

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(ctx context.Context, clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaRedpandaServerSecurity, security *dagger.KafkaClientSecurity, name string)   {
	return dag.
			Kafka().
			Redpandacluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity).
			Client(security).
			Deletetopic(ctx, name)
}content_copy Copied!

@function
async def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaRedpandaServerSecurity, security: dagger.KafkaClientSecurity, name: str) -> None:
	return await (
		dag.kafka()
		.redpandacluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
		.client(security)
		.deletetopic(name)
	)content_copy Copied!

@func()
async example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaRedpandaServerSecurity, security: KafkaClientSecurity, name: string): Promise<void> {
	return dag
		.kafka()
		.redpandaCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
		.client(security)
		.deleteTopic(name)
}content_copy Copied!

listTopics() 🔗

ListTopics returns the names of every topic the broker reports.

Return Type

[String ! ] !

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(ctx context.Context, clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaRedpandaServerSecurity, security *dagger.KafkaClientSecurity) []string  {
	return dag.
			Kafka().
			Redpandacluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity).
			Client(security).
			Listtopics(ctx)
}content_copy Copied!

@function
async def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaRedpandaServerSecurity, security: dagger.KafkaClientSecurity) -> List[str]:
	return await (
		dag.kafka()
		.redpandacluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
		.client(security)
		.listtopics()
	)content_copy Copied!

@func()
async example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaRedpandaServerSecurity, security: KafkaClientSecurity): Promise<string[]> {
	return dag
		.kafka()
		.redpandaCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
		.client(security)
		.listTopics()
}content_copy Copied!

produce() 🔗

Produce synchronously writes one record to the topic. Key and value are decoded from their named encodings into raw bytes before being sent.

keySchemaID / valueSchemaID, when positive, prepend the Confluent Schema Registry wire-format header to the corresponding field: 0x00 || uint32be(schemaID) || payload. The header is laid down via franz-go’s sr.ConfluentHeader so the byte layout matches what Schema-Registry-aware consumers expect. Default 0 means no framing (the field is sent verbatim). Negative IDs are rejected.

keySerializeAs / valueSerializeAs, when set to “JSON”, parse the corresponding decoded bytes with encoding/json and re-marshal to the canonical form before any framing is applied. The pipeline order is decode → serialize → frame, so a single call can both canonicalise a JSON payload and prepend the Confluent header. Invalid JSON is rejected before any broker I/O. Default “” is pass-through; “JSON” canonicalises.

keySerializeAs / valueSerializeAs set to “AVRO” interpret the decoded bytes as a JSON document and Avro-binary-encode it (via github.com/z5labs/avro-go/generic) against the schema identified by keySchemaID / valueSchemaID before framing. The id is required in this mode — a zero / negative id errors out before any broker or registry I/O — and registry must be supplied so the schema text can be resolved by id. The JSON shape follows the Avro spec’s JSON encoding; logical types, decimal, and fixed are not yet supported.

Return Type

Void !

Arguments

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(ctx context.Context, clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaRedpandaServerSecurity, security *dagger.KafkaClientSecurity, topic string, key string, value string, keyEncoding string, valueEncoding string, keySchemaId int, valueSchemaId int, keySerializeAs string, valueSerializeAs string)   {
	return dag.
			Kafka().
			Redpandacluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity).
			Client(security).
			Produce(ctx, topic, key, value, keyEncoding, valueEncoding, keySchemaId, valueSchemaId, keySerializeAs, valueSerializeAs)
}content_copy Copied!

@function
async def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaRedpandaServerSecurity, security: dagger.KafkaClientSecurity, topic: str, key: str, value: str, keyencoding: str, valueencoding: str, keyschemaid: int, valueschemaid: int, keyserializeas: str, valueserializeas: str) -> None:
	return await (
		dag.kafka()
		.redpandacluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
		.client(security)
		.produce(topic, key, value, keyencoding, valueencoding, keyschemaid, valueschemaid, keyserializeas, valueserializeas)
	)content_copy Copied!

@func()
async example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaRedpandaServerSecurity, security: KafkaClientSecurity, topic: string, key: string, value: string, keyEncoding: string, valueEncoding: string, keySchemaId: number, valueSchemaId: number, keySerializeAs: string, valueSerializeAs: string): Promise<void> {
	return dag
		.kafka()
		.redpandaCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
		.client(security)
		.produce(topic, key, value, keyEncoding, valueEncoding, keySchemaId, valueSchemaId, keySerializeAs, valueSerializeAs)
}content_copy Copied!

propertiesFile() 🔗

PropertiesFile renders this client’s connection settings as a Java client.properties file so callers can hand it to the Apache Kafka command-line tools or to other JVM-based consumers.

For TLS / mTLS modes the properties reference PKCS#12 truststore (and keystore for mTLS) by basename — the matching p12 files are written alongside client.properties in the same directory. Callers should export the parent directory (props.Directory()) so the relative references resolve. Passwords appear plaintext, which is a Kafka CLI constraint.

Return Type

File !

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaRedpandaServerSecurity, security *dagger.KafkaClientSecurity) *dagger.File  {
	return dag.
			Kafka().
			Redpandacluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity).
			Client(security).
			Propertiesfile()
}content_copy Copied!

@function
def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaRedpandaServerSecurity, security: dagger.KafkaClientSecurity) -> dagger.File:
	return (
		dag.kafka()
		.redpandacluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
		.client(security)
		.propertiesfile()
	)content_copy Copied!

@func()
example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaRedpandaServerSecurity, security: KafkaClientSecurity): File {
	return dag
		.kafka()
		.redpandaCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
		.client(security)
		.propertiesFile()
}content_copy Copied!

ClientSecurity 🔗

ClientSecurity describes how a franz-go client authenticates to a Kafka broker.

SchemaRegistryClientSecurity 🔗

SchemaRegistryClientSecurity describes how a SchemaRegistryClient’s HTTP client authenticates to a Schema Registry’s REST endpoint, mirroring *ClientSecurity on the broker side.

RedpandaCluster 🔗

RedpandaCluster is the Redpanda counterpart to *Cluster. Redpanda speaks the Kafka wire protocol but is a from-scratch C++ implementation with a completely different configuration layer (rpk redpanda start, a YAML config file, PEM cert/key files instead of PKCS#12), so it gets its own return type to make the divergence visible at the API surface. Single node only in this story (controllers=1, brokers=1).

bindBrokers() 🔗

BindBrokers binds the single Redpanda broker service into the given container so the container can reach it by hostname.

Return Type

Container !

Arguments

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaRedpandaServerSecurity, ctr *dagger.Container) *dagger.Container  {
	return dag.
			Kafka().
			Redpandacluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity).
			Bindbrokers(ctr)
}content_copy Copied!

@function
def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaRedpandaServerSecurity, ctr: dagger.Container) -> dagger.Container:
	return (
		dag.kafka()
		.redpandacluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
		.bindbrokers(ctr)
	)content_copy Copied!

@func()
example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaRedpandaServerSecurity, ctr: Container): Container {
	return dag
		.kafka()
		.redpandaCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
		.bindBrokers(ctr)
}content_copy Copied!

bootstrapServers() 🔗

BootstrapServers returns the bootstrap address (single broker:9092) for this Redpanda cluster.

Return Type

[String ! ] !

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(ctx context.Context, clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaRedpandaServerSecurity) []string  {
	return dag.
			Kafka().
			Redpandacluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity).
			Bootstrapservers(ctx)
}content_copy Copied!

@function
async def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaRedpandaServerSecurity) -> List[str]:
	return await (
		dag.kafka()
		.redpandacluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
		.bootstrapservers()
	)content_copy Copied!

@func()
async example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaRedpandaServerSecurity): Promise<string[]> {
	return dag
		.kafka()
		.redpandaCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
		.bootstrapServers()
}content_copy Copied!

client() 🔗

Client starts the Redpanda broker service and returns a franz-go-backed *Client targeting it. The Kafka wire protocol matches Apache Kafka, so the existing *Client + *ClientSecurity (PKCS#12) are reused unchanged.

Return Type

Client !

Arguments

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaRedpandaServerSecurity, security *dagger.KafkaClientSecurity) *dagger.KafkaClient  {
	return dag.
			Kafka().
			Redpandacluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity).
			Client(security)
}content_copy Copied!

@function
def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaRedpandaServerSecurity, security: dagger.KafkaClientSecurity) -> dagger.KafkaClient:
	return (
		dag.kafka()
		.redpandacluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
		.client(security)
	)content_copy Copied!

@func()
example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaRedpandaServerSecurity, security: KafkaClientSecurity): KafkaClient {
	return dag
		.kafka()
		.redpandaCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
		.client(security)
}content_copy Copied!

schemaRegistry() 🔗

SchemaRegistry exposes Redpanda’s bundled Schema Registry as the same *SchemaRegistry type Kafka.ConfluentSchemaRegistry returns, so callers can treat the bundled and separate-container registries uniformly.

rpk redpanda start runs a Schema Registry inside the broker process on :8081 — no extra container — so the returned *SchemaRegistry points at the broker service itself. Redpanda’s SR speaks the Confluent Schema Registry REST API, so the *SchemaRegistryClient from Client() works unchanged.

security must match the cluster’s mode (PLAINTEXT or TLS — Redpanda has no mTLS): on a TLS cluster the bundled SR REST endpoint terminates HTTPS reusing the broker’s server leaf (configured at cluster-build time in renderRedpandaYaml), so the caller must pass a TLS profile to get an HTTPS client. The profile’s CA keystore is unused here (the leaf is already minted); it is required only for API uniformity with the other registries.

The returned registry is Bundled: its service is the broker itself, so Stop is a no-op on it — call cluster.Stop to tear the registry down with the cluster. A caller that uniformly defer sr.Stop(ctx) over the shared *SchemaRegistry type therefore can’t accidentally kill the cluster.

Return Type

SchemaRegistry !

Arguments

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaRedpandaServerSecurity, security *dagger.KafkaSchemaRegistrySecurity) *dagger.KafkaSchemaRegistry  {
	return dag.
			Kafka().
			Redpandacluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity).
			Schemaregistry(security)
}content_copy Copied!

@function
def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaRedpandaServerSecurity, security: dagger.KafkaSchemaRegistrySecurity) -> dagger.KafkaSchemaRegistry:
	return (
		dag.kafka()
		.redpandacluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
		.schemaregistry(security)
	)content_copy Copied!

@func()
example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaRedpandaServerSecurity, security: KafkaSchemaRegistrySecurity): KafkaSchemaRegistry {
	return dag
		.kafka()
		.redpandaCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
		.schemaRegistry(security)
}content_copy Copied!

stop() 🔗

Stop tears down the broker container backing this Redpanda cluster. Tests should call this in a defer so the broker Container.asService span closes when the test work is done. Kill is set so Service.Stop skips graceful shutdown — see Cluster.Stop for the rationale.

Return Type

Void !

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(ctx context.Context, clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaRedpandaServerSecurity)   {
	return dag.
			Kafka().
			Redpandacluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity).
			Stop(ctx)
}content_copy Copied!

@function
async def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaRedpandaServerSecurity) -> None:
	return await (
		dag.kafka()
		.redpandacluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
		.stop()
	)content_copy Copied!

@func()
async example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaRedpandaServerSecurity): Promise<void> {
	return dag
		.kafka()
		.redpandaCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
		.stop()
}content_copy Copied!

RedpandaServerSecurity 🔗

RedpandaServerSecurity carries the external-listener security profile for a Redpanda cluster. Same shape as *ServerSecurity (PKCS#12 CAso callers don’t have to convert; the constructor extracts PEM from the issued leaf internally for redpanda.yaml. Separate type from *ServerSecurity so a caller can’t accidentally hand an Apache profile (e.g. MtlsServerSecurity, not supported here yet) to RedpandaCluster.

SchemaRegistryClient 🔗

SchemaRegistryClient is a pure-Go net/http client for a Schema Registry’s admin REST API. Each method opens a fresh request so the function call is stateless from Dagger’s perspective.

deleteSubject() 🔗

DeleteSubject deletes every version of the given subject and returns the version numbers that were deleted.

Return Type

[Integer ! ] !

Arguments

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(ctx context.Context, clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaRedpandaServerSecurity, security *dagger.KafkaSchemaRegistrySecurity, security1 *dagger.KafkaSchemaRegistryClientSecurity, subject string) []int  {
	return dag.
			Kafka().
			Redpandacluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity).
			Schemaregistry(security).
			Client(security1).
			Deletesubject(ctx, subject)
}content_copy Copied!

@function
async def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaRedpandaServerSecurity, security: dagger.KafkaSchemaRegistrySecurity, security1: dagger.KafkaSchemaRegistryClientSecurity, subject: str) -> List[int]:
	return await (
		dag.kafka()
		.redpandacluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
		.schemaregistry(security)
		.client(security1)
		.deletesubject(subject)
	)content_copy Copied!

@func()
async example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaRedpandaServerSecurity, security: KafkaSchemaRegistrySecurity, security1: KafkaSchemaRegistryClientSecurity, subject: string): Promise<number[]> {
	return dag
		.kafka()
		.redpandaCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
		.schemaRegistry(security)
		.client(security1)
		.deleteSubject(subject)
}content_copy Copied!

getCompatibility() 🔗

GetCompatibility returns the compatibility level configured for the given subject, falling back to the registry-wide default when the subject has no explicit configuration.

Return Type

String !

Arguments

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(ctx context.Context, clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaRedpandaServerSecurity, security *dagger.KafkaSchemaRegistrySecurity, security1 *dagger.KafkaSchemaRegistryClientSecurity, subject string) string  {
	return dag.
			Kafka().
			Redpandacluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity).
			Schemaregistry(security).
			Client(security1).
			Getcompatibility(ctx, subject)
}content_copy Copied!

@function
async def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaRedpandaServerSecurity, security: dagger.KafkaSchemaRegistrySecurity, security1: dagger.KafkaSchemaRegistryClientSecurity, subject: str) -> str:
	return await (
		dag.kafka()
		.redpandacluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
		.schemaregistry(security)
		.client(security1)
		.getcompatibility(subject)
	)content_copy Copied!

@func()
async example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaRedpandaServerSecurity, security: KafkaSchemaRegistrySecurity, security1: KafkaSchemaRegistryClientSecurity, subject: string): Promise<string> {
	return dag
		.kafka()
		.redpandaCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
		.schemaRegistry(security)
		.client(security1)
		.getCompatibility(subject)
}content_copy Copied!

listSubjects() 🔗

ListSubjects returns the names of every subject registered.

Return Type

[String ! ] !

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(ctx context.Context, clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaRedpandaServerSecurity, security *dagger.KafkaSchemaRegistrySecurity, security1 *dagger.KafkaSchemaRegistryClientSecurity) []string  {
	return dag.
			Kafka().
			Redpandacluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity).
			Schemaregistry(security).
			Client(security1).
			Listsubjects(ctx)
}content_copy Copied!

@function
async def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaRedpandaServerSecurity, security: dagger.KafkaSchemaRegistrySecurity, security1: dagger.KafkaSchemaRegistryClientSecurity) -> List[str]:
	return await (
		dag.kafka()
		.redpandacluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
		.schemaregistry(security)
		.client(security1)
		.listsubjects()
	)content_copy Copied!

@func()
async example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaRedpandaServerSecurity, security: KafkaSchemaRegistrySecurity, security1: KafkaSchemaRegistryClientSecurity): Promise<string[]> {
	return dag
		.kafka()
		.redpandaCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
		.schemaRegistry(security)
		.client(security1)
		.listSubjects()
}content_copy Copied!

lookupLatestBySubject() 🔗

LookupLatestBySubject returns the latest registered schema version for the given subject.

Return Type

RegisteredSchema !

Arguments

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaRedpandaServerSecurity, security *dagger.KafkaSchemaRegistrySecurity, security1 *dagger.KafkaSchemaRegistryClientSecurity, subject string) *dagger.KafkaRegisteredSchema  {
	return dag.
			Kafka().
			Redpandacluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity).
			Schemaregistry(security).
			Client(security1).
			Lookuplatestbysubject(subject)
}content_copy Copied!

@function
def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaRedpandaServerSecurity, security: dagger.KafkaSchemaRegistrySecurity, security1: dagger.KafkaSchemaRegistryClientSecurity, subject: str) -> dagger.KafkaRegisteredSchema:
	return (
		dag.kafka()
		.redpandacluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
		.schemaregistry(security)
		.client(security1)
		.lookuplatestbysubject(subject)
	)content_copy Copied!

@func()
example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaRedpandaServerSecurity, security: KafkaSchemaRegistrySecurity, security1: KafkaSchemaRegistryClientSecurity, subject: string): KafkaRegisteredSchema {
	return dag
		.kafka()
		.redpandaCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
		.schemaRegistry(security)
		.client(security1)
		.lookupLatestBySubject(subject)
}content_copy Copied!

lookupSchemaById() 🔗

LookupSchemaByID returns the schema registered under the given global id.

The registry’s GET /schemas/ids/{id} endpoint reports only the schema text and type, so a second call to GET /schemas/ids/{id}/versions resolves the subject and version. When an id maps to more than one subject/version pair, the first association is returned.

Return Type

RegisteredSchema !

Arguments

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaRedpandaServerSecurity, security *dagger.KafkaSchemaRegistrySecurity, security1 *dagger.KafkaSchemaRegistryClientSecurity, id int) *dagger.KafkaRegisteredSchema  {
	return dag.
			Kafka().
			Redpandacluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity).
			Schemaregistry(security).
			Client(security1).
			Lookupschemabyid(id)
}content_copy Copied!

@function
def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaRedpandaServerSecurity, security: dagger.KafkaSchemaRegistrySecurity, security1: dagger.KafkaSchemaRegistryClientSecurity, id: int) -> dagger.KafkaRegisteredSchema:
	return (
		dag.kafka()
		.redpandacluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
		.schemaregistry(security)
		.client(security1)
		.lookupschemabyid(id)
	)content_copy Copied!

@func()
example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaRedpandaServerSecurity, security: KafkaSchemaRegistrySecurity, security1: KafkaSchemaRegistryClientSecurity, id: number): KafkaRegisteredSchema {
	return dag
		.kafka()
		.redpandaCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
		.schemaRegistry(security)
		.client(security1)
		.lookupSchemaById(id)
}content_copy Copied!

registerSchema() 🔗

RegisterSchema registers schema under subject and returns the globally unique schema id the registry assigned. schemaType must be one of AVRO, JSON, or PROTOBUF.

Return Type

Integer !

Arguments

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(ctx context.Context, clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaRedpandaServerSecurity, security *dagger.KafkaSchemaRegistrySecurity, security1 *dagger.KafkaSchemaRegistryClientSecurity, subject string, schema string, schemaType string) int  {
	return dag.
			Kafka().
			Redpandacluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity).
			Schemaregistry(security).
			Client(security1).
			Registerschema(ctx, subject, schema, schemaType)
}content_copy Copied!

@function
async def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaRedpandaServerSecurity, security: dagger.KafkaSchemaRegistrySecurity, security1: dagger.KafkaSchemaRegistryClientSecurity, subject: str, schema: str, schematype: str) -> int:
	return await (
		dag.kafka()
		.redpandacluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
		.schemaregistry(security)
		.client(security1)
		.registerschema(subject, schema, schematype)
	)content_copy Copied!

@func()
async example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaRedpandaServerSecurity, security: KafkaSchemaRegistrySecurity, security1: KafkaSchemaRegistryClientSecurity, subject: string, schema: string, schemaType: string): Promise<number> {
	return dag
		.kafka()
		.redpandaCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
		.schemaRegistry(security)
		.client(security1)
		.registerSchema(subject, schema, schemaType)
}content_copy Copied!

setCompatibility() 🔗

SetCompatibility sets the compatibility level for the given subject. level must be one of NONE, BACKWARD, BACKWARD_TRANSITIVE, FORWARD, FORWARD_TRANSITIVE, FULL, or FULL_TRANSITIVE.

Return Type

Void !

Arguments

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(ctx context.Context, clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaRedpandaServerSecurity, security *dagger.KafkaSchemaRegistrySecurity, security1 *dagger.KafkaSchemaRegistryClientSecurity, subject string, level string)   {
	return dag.
			Kafka().
			Redpandacluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity).
			Schemaregistry(security).
			Client(security1).
			Setcompatibility(ctx, subject, level)
}content_copy Copied!

@function
async def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaRedpandaServerSecurity, security: dagger.KafkaSchemaRegistrySecurity, security1: dagger.KafkaSchemaRegistryClientSecurity, subject: str, level: str) -> None:
	return await (
		dag.kafka()
		.redpandacluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
		.schemaregistry(security)
		.client(security1)
		.setcompatibility(subject, level)
	)content_copy Copied!

@func()
async example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaRedpandaServerSecurity, security: KafkaSchemaRegistrySecurity, security1: KafkaSchemaRegistryClientSecurity, subject: string, level: string): Promise<void> {
	return dag
		.kafka()
		.redpandaCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
		.schemaRegistry(security)
		.client(security1)
		.setCompatibility(subject, level)
}content_copy Copied!

RegisteredSchema 🔗

RegisteredSchema is one schema version as the Confluent Schema Registry reports it.

The field names deliberately diverge from the REST API’s JSON keys (id, schema): an exported ID field collides with the synthetic Dagger object id, and Schema is a GraphQL keyword that breaks consumer-module codegen — see daggerverse/CLAUDE.md.

subject() 🔗

registry subject the schema is registered under

Return Type

String !

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(ctx context.Context, clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaRedpandaServerSecurity, security *dagger.KafkaSchemaRegistrySecurity, security1 *dagger.KafkaSchemaRegistryClientSecurity, id int) string  {
	return dag.
			Kafka().
			Redpandacluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity).
			Schemaregistry(security).
			Client(security1).
			Lookupschemabyid(id).
			Subject(ctx)
}content_copy Copied!

@function
async def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaRedpandaServerSecurity, security: dagger.KafkaSchemaRegistrySecurity, security1: dagger.KafkaSchemaRegistryClientSecurity, id: int) -> str:
	return await (
		dag.kafka()
		.redpandacluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
		.schemaregistry(security)
		.client(security1)
		.lookupschemabyid(id)
		.subject()
	)content_copy Copied!

@func()
async example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaRedpandaServerSecurity, security: KafkaSchemaRegistrySecurity, security1: KafkaSchemaRegistryClientSecurity, id: number): Promise<string> {
	return dag
		.kafka()
		.redpandaCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
		.schemaRegistry(security)
		.client(security1)
		.lookupSchemaById(id)
		.subject()
}content_copy Copied!

version() 🔗

monotonic version within the subject

Return Type

Integer !

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(ctx context.Context, clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaRedpandaServerSecurity, security *dagger.KafkaSchemaRegistrySecurity, security1 *dagger.KafkaSchemaRegistryClientSecurity, id int) int  {
	return dag.
			Kafka().
			Redpandacluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity).
			Schemaregistry(security).
			Client(security1).
			Lookupschemabyid(id).
			Version(ctx)
}content_copy Copied!

@function
async def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaRedpandaServerSecurity, security: dagger.KafkaSchemaRegistrySecurity, security1: dagger.KafkaSchemaRegistryClientSecurity, id: int) -> int:
	return await (
		dag.kafka()
		.redpandacluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
		.schemaregistry(security)
		.client(security1)
		.lookupschemabyid(id)
		.version()
	)content_copy Copied!

@func()
async example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaRedpandaServerSecurity, security: KafkaSchemaRegistrySecurity, security1: KafkaSchemaRegistryClientSecurity, id: number): Promise<number> {
	return dag
		.kafka()
		.redpandaCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
		.schemaRegistry(security)
		.client(security1)
		.lookupSchemaById(id)
		.version()
}content_copy Copied!

schemaId() 🔗

globally-unique registry schema id

Return Type

Integer !

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(ctx context.Context, clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaRedpandaServerSecurity, security *dagger.KafkaSchemaRegistrySecurity, security1 *dagger.KafkaSchemaRegistryClientSecurity, id int) int  {
	return dag.
			Kafka().
			Redpandacluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity).
			Schemaregistry(security).
			Client(security1).
			Lookupschemabyid(id).
			Schemaid(ctx)
}content_copy Copied!

@function
async def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaRedpandaServerSecurity, security: dagger.KafkaSchemaRegistrySecurity, security1: dagger.KafkaSchemaRegistryClientSecurity, id: int) -> int:
	return await (
		dag.kafka()
		.redpandacluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
		.schemaregistry(security)
		.client(security1)
		.lookupschemabyid(id)
		.schemaid()
	)content_copy Copied!

@func()
async example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaRedpandaServerSecurity, security: KafkaSchemaRegistrySecurity, security1: KafkaSchemaRegistryClientSecurity, id: number): Promise<number> {
	return dag
		.kafka()
		.redpandaCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
		.schemaRegistry(security)
		.client(security1)
		.lookupSchemaById(id)
		.schemaId()
}content_copy Copied!

definition() 🔗

the schema text itself (Avro / JSON Schema / Protobuf)

Return Type

String !

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(ctx context.Context, clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaRedpandaServerSecurity, security *dagger.KafkaSchemaRegistrySecurity, security1 *dagger.KafkaSchemaRegistryClientSecurity, id int) string  {
	return dag.
			Kafka().
			Redpandacluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity).
			Schemaregistry(security).
			Client(security1).
			Lookupschemabyid(id).
			Definition(ctx)
}content_copy Copied!

@function
async def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaRedpandaServerSecurity, security: dagger.KafkaSchemaRegistrySecurity, security1: dagger.KafkaSchemaRegistryClientSecurity, id: int) -> str:
	return await (
		dag.kafka()
		.redpandacluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
		.schemaregistry(security)
		.client(security1)
		.lookupschemabyid(id)
		.definition()
	)content_copy Copied!

@func()
async example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaRedpandaServerSecurity, security: KafkaSchemaRegistrySecurity, security1: KafkaSchemaRegistryClientSecurity, id: number): Promise<string> {
	return dag
		.kafka()
		.redpandaCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
		.schemaRegistry(security)
		.client(security1)
		.lookupSchemaById(id)
		.definition()
}content_copy Copied!

schemaType() 🔗

AVRO | JSON | PROTOBUF

Return Type

String !

echo 'Custom types are not supported in shell examples'content_copy Copied! 

func (m *MyModule) Example(ctx context.Context, clusterId string, controllers int, brokers int, registry string, tag string, clientListenerSecurity *dagger.KafkaRedpandaServerSecurity, security *dagger.KafkaSchemaRegistrySecurity, security1 *dagger.KafkaSchemaRegistryClientSecurity, id int) string  {
	return dag.
			Kafka().
			Redpandacluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity).
			Schemaregistry(security).
			Client(security1).
			Lookupschemabyid(id).
			Schematype(ctx)
}content_copy Copied!

@function
async def example(clusterid: str, controllers: int, brokers: int, registry: str, tag: str, clientlistenersecurity: dagger.KafkaRedpandaServerSecurity, security: dagger.KafkaSchemaRegistrySecurity, security1: dagger.KafkaSchemaRegistryClientSecurity, id: int) -> str:
	return await (
		dag.kafka()
		.redpandacluster(clusterid, controllers, brokers, registry, tag, clientlistenersecurity)
		.schemaregistry(security)
		.client(security1)
		.lookupschemabyid(id)
		.schematype()
	)content_copy Copied!

@func()
async example(clusterId: string, controllers: number, brokers: number, registry: string, tag: string, clientListenerSecurity: KafkaRedpandaServerSecurity, security: KafkaSchemaRegistrySecurity, security1: KafkaSchemaRegistryClientSecurity, id: number): Promise<string> {
	return dag
		.kafka()
		.redpandaCluster(clusterId, controllers, brokers, registry, tag, clientListenerSecurity)
		.schemaRegistry(security)
		.client(security1)
		.lookupSchemaById(id)
		.schemaType()
}content_copy Copied!