tests
Installation
Entrypoint
Types
Tests

tests

dagger function so it can be invoked individually during TDD; All wires them
up for grouped, parallel execution under `dagger call all`.

File map (all `package main`, surfaced as one Dagger module):

- main.go — Tests struct, the All() orchestrator, and the four
per-distro group functions (nativeTests,
apacheJVMTests, confluentTests, redpandaTests).
- helpers.go — cross-cutting scaffolding shared across distros:
newClusterId, freshCa, randHex, randomTopicName,
contains.
- tests_native.go — ApacheNativeCluster (apache/kafka-native) cluster
helpers (freshCluster / freshTlsCluster /
freshMtlsCluster) + every test that drives the
GraalVM image (the bulk of the suite, including
shared roundTripBinaryOn).
- tests_apache.go — ApacheCluster (apache/kafka JVM) cluster helpers
+ the three Apache-JVM round-trip tests.
- tests_confluent.go — ConfluentCluster (confluentinc/cp-kafka) cluster
helpers + the three cp-kafka round-trip tests.
- tests_redpanda.go — RedpandaCluster (redpandadata/redpanda) cluster
helpers + the two Redpanda Kafka-wire round-trip
tests, the PLAINTEXT + TLS bundled-Schema-
Registry round-trips, and the bundled-registry
Stop-is-a-no-op lifecycle test.
- tests_schema_registry.go — ConfluentSchemaRegistry tests: the
register/lookup/delete round-trip and the
non-PLAINTEXT-cluster rejection.

Installation

dagger install github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453

Entrypoint

Return Type

Tests

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \

func (m *MyModule) Example() *dagger.Tests  {
	return dag.
			Tests()
}

@function
def example() -> dagger.Tests:
	return (
		dag.tests()
	)

@func()
example(): Tests {
	return dag
		.tests()
}

Types

Tests 🔗

all() 🔗

All runs every kafka round-trip test as a convenience for local dagger call all invocations that want the entire suite in one shot. CI does NOT call All: each per-distro group below carries its own +check directive, so GH Actions schedules each onto its own runner in parallel — running All on top would double-bill the same work.

kafkaImageTag picks the tag every spawned Apache cluster runs against — applied to both the apache/kafka-native image (ApacheNativeCluster) and the apache/kafka JVM image (ApacheCluster). confluentImageTag is the independent knob for the cp-kafka tests (Confluent Platform versioning is not aligned with Apache’s release numbering). redpandaImageTag is the independent knob for the redpandadata/redpanda tests.

parallel is the concurrency cap applied at both levels — how many groups run at once and how many tests run at once within each group. Defaults to 0 (unbounded). Pass any positive integer for a specific cap.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided
confluentImageTag	String !	"8.2.0"	No description provided
redpandaImageTag	String !	"v26.1.7"	No description provided
parallel	Integer !	0	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 all --kafka-image-tag string --confluent-image-tag string --redpanda-image-tag string --parallel integer

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string, confluentImageTag string, redpandaImageTag string, parallel int)   {
	return dag.
			Tests().
			All(ctx, kafkaImageTag, confluentImageTag, redpandaImageTag, parallel)
}

@function
async def example(kafkaimagetag: str, confluentimagetag: str, redpandaimagetag: str, parallel: int) -> None:
	return await (
		dag.tests()
		.all(kafkaimagetag, confluentimagetag, redpandaimagetag, parallel)
	)

@func()
async example(kafkaImageTag: string, confluentImageTag: string, redpandaImageTag: string, parallel: number): Promise<void> {
	return dag
		.tests()
		.all(kafkaImageTag, confluentImageTag, redpandaImageTag, parallel)
}

apacheClusterMtlsRoundTrip() 🔗

ApacheClusterMtlsRoundTrip is the MTLS happy-path round-trip for Kafka.ApacheCluster. Mirrors MtlsRoundTrip but on the JVM image to rule out image-specific differences in how client-cert challenge is handled.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 apache-cluster-mtls-round-trip --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Apacheclustermtlsroundtrip(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.apacheclustermtlsroundtrip(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.apacheClusterMtlsRoundTrip(kafkaImageTag)
}

apacheClusterProduceListTopicsRoundTrip() 🔗

ApacheClusterProduceListTopicsRoundTrip is the PLAINTEXT happy-path smoke test for Kafka.ApacheCluster (the JVM image variant): produce a single raw record, then call ListTopics and assert the freshly-created topic shows up. Together these prove the JVM image’s data plane and control plane both work; the env-var contract matches ApacheNativeCluster so this single test pins down “JVM image actually serves traffic”.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 apache-cluster-produce-list-topics-round-trip --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Apacheclusterproducelisttopicsroundtrip(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.apacheclusterproducelisttopicsroundtrip(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.apacheClusterProduceListTopicsRoundTrip(kafkaImageTag)
}

apacheClusterTlsRoundTrip() 🔗

ApacheClusterTlsRoundTrip is the TLS happy-path round-trip for Kafka.ApacheCluster. Mirrors TlsRoundTrip but on the JVM image to rule out image-specific differences in keystore mounts, hostname verification, and SSL listener bring-up.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 apache-cluster-tls-round-trip --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Apacheclustertlsroundtrip(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.apacheclustertlsroundtrip(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.apacheClusterTlsRoundTrip(kafkaImageTag)
}

apacheJvm() 🔗

ApacheJVM runs the three apache/kafka JVM-image round-trip tests. Each test owns a fresh ApacheCluster, so the group holds no shared clusters of its own.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided
parallel	Integer !	0	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 apache-jvm --kafka-image-tag string --parallel integer

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string, parallel int)   {
	return dag.
			Tests().
			Apachejvm(ctx, kafkaImageTag, parallel)
}

@function
async def example(kafkaimagetag: str, parallel: int) -> None:
	return await (
		dag.tests()
		.apachejvm(kafkaimagetag, parallel)
	)

@func()
async example(kafkaImageTag: string, parallel: number): Promise<void> {
	return dag
		.tests()
		.apacheJvm(kafkaImageTag, parallel)
}

apicurioSchemaRegistryRegisterLookupRoundTrip() 🔗

ApicurioSchemaRegistryRegisterLookupRoundTrip is the PLAINTEXT happy-path test for Kafka.ApicurioSchemaRegistry: stand an apicurio-registry-kafkasql up next to a fresh cluster, then exercise register → lookup-by-id → lookup-latest-by-subject → list-subjects → set/get-compatibility → delete against its Confluent-compatible REST surface — mirroring SchemaRegistryRegisterLookupRoundTrip to prove the shared *SchemaRegistryClient drives Apicurio unchanged.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 apicurio-schema-registry-register-lookup-round-trip --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Apicurioschemaregistryregisterlookuproundtrip(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.apicurioschemaregistryregisterlookuproundtrip(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.apicurioSchemaRegistryRegisterLookupRoundTrip(kafkaImageTag)
}

apicurioSchemaRegistryTlsRegisterLookupRoundTrip() 🔗

ApicurioSchemaRegistryTlsRegisterLookupRoundTrip drives the Apicurio TLS path (Quarkus HTTPS REST + kafkasql SSL storage) end-to-end.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 apicurio-schema-registry-tls-register-lookup-round-trip --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Apicurioschemaregistrytlsregisterlookuproundtrip(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.apicurioschemaregistrytlsregisterlookuproundtrip(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.apicurioSchemaRegistryTlsRegisterLookupRoundTrip(kafkaImageTag)
}

autoCreateTopicsDisabled() 🔗

AutoCreateTopicsDisabled produces to a topic that was never created and asserts the call errors out. With KAFKA_AUTO_CREATE_TOPICS_ENABLE=false on the broker, the produce path must surface a topic-not-found error rather than silently auto-creating, so producer typos can’t pass tests.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 auto-create-topics-disabled --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Autocreatetopicsdisabled(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.autocreatetopicsdisabled(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.autoCreateTopicsDisabled(kafkaImageTag)
}

avroBytesFieldRoundTrip() 🔗

AvroBytesFieldRoundTrip pins the Avro-spec JSON encoding of a bytes field. Per the Avro specification, the JSON encoding of bytes (and fixed) is a string whose characters are the byte values as Unicode code points 0-255 — one character per byte, NOT base64. The input value here contains byte 0xFF (code point U+00FF), which is outside the base64 alphabet, so a round-trip that preserves it byte-for-byte proves the one-char-per-byte mapping and would be impossible under a base64 interpretation.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 avro-bytes-field-round-trip --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Avrobytesfieldroundtrip(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.avrobytesfieldroundtrip(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.avroBytesFieldRoundTrip(kafkaImageTag)
}

avroConsumeUnframedErrors() 🔗

AvroConsumeUnframedErrors pins the negative consume path: a record produced without a Confluent wire header, consumed with valueDeserializeAs=“AVRO” + schemaRegistryAware=true, must error out pointing at the missing header. The header check fires before any schema lookup, so the registry service never has to start.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 avro-consume-unframed-errors --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Avroconsumeunframederrors(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.avroconsumeunframederrors(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.avroConsumeUnframedErrors(kafkaImageTag)
}

avroFramedProduceConsumeRoundTrip() 🔗

AvroFramedProduceConsumeRoundTrip is the happy-path data round-trip for AVRO serde: register an Avro record schema to get an id, Produce a JSON document with valueSerializeAs=“AVRO”then framed), and Consume it back with valueDeserializeAs=“AVRO” + schemaRegistryAware=true. The asserted invariant is byte-equality of the consumed value to the canonical JSON form of the original input, proving the JSON->Avro-binary->JSON pipeline preserves the datum.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 avro-framed-produce-consume-round-trip --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Avroframedproduceconsumeroundtrip(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.avroframedproduceconsumeroundtrip(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.avroFramedProduceConsumeRoundTrip(kafkaImageTag)
}

avroMtlsFramedProduceConsumeRoundTrip() 🔗

AvroMtlsFramedProduceConsumeRoundTrip is the mTLS counterpart: the same round-trip where the kafka-wire client and the REST schema-resolution client each present their own leaf (both signed by the single CA). It proves the threaded RegistrySecurity profile also carries a client key store through to the avro path, not just a trust store.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 avro-mtls-framed-produce-consume-round-trip --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Avromtlsframedproduceconsumeroundtrip(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.avromtlsframedproduceconsumeroundtrip(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.avroMtlsFramedProduceConsumeRoundTrip(kafkaImageTag)
}

avroSerializeRequiresSchemaId() 🔗

AvroSerializeRequiresSchemaID pins the up-front validation contract of valueSerializeAs=“AVRO”: Produce must reject a zero schema id before any broker or registry I/O. dag.Kafka().Client(…) builds without I/O, so no cluster boots — the failure is purely the missing-id guard on the AVRO serializer.

Return Type

Void !

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 avro-serialize-requires-schema-id

func (m *MyModule) Example(ctx context.Context)   {
	return dag.
			Tests().
			Avroserializerequiresschemaid(ctx)
}

@function
async def example() -> None:
	return await (
		dag.tests()
		.avroserializerequiresschemaid()
	)

@func()
async example(): Promise<void> {
	return dag
		.tests()
		.avroSerializeRequiresSchemaId()
}

avroTlsFramedProduceConsumeRoundTrip() 🔗

AvroTlsFramedProduceConsumeRoundTrip is the TLS counterpart of AvroFramedProduceConsumeRoundTrip: it stands up a TLS cluster and a TLS cp-schema-registry rooted at one CA and drives the same JSON->Avro-binary-> JSON round-trip, but every hop is encrypted — the kafka-wire client speaks TLS to the brokers and the Produce/Consume avro path resolves the schema over HTTPS via the threaded RegistrySecurity profile (#141). Without that profile the resolution would fail with “serves HTTPS but client is PLAINTEXT”, so a green round-trip proves the profile reaches Client(…).

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 avro-tls-framed-produce-consume-round-trip --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Avrotlsframedproduceconsumeroundtrip(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.avrotlsframedproduceconsumeroundtrip(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.avroTlsFramedProduceConsumeRoundTrip(kafkaImageTag)
}

bindBrokersExposesBothListeners() 🔗

BindBrokersExposesBothListeners binds the cluster’s brokers into a vanilla alpine container and asserts that both the host-facing client port (9092) and the inter-broker port (19092) are reachable from inside that container — together they cover the dual-listener contract (PLAINTEXT_HOST:9092 for clients, PLAINTEXT:19092 for inter-broker).

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 bind-brokers-exposes-both-listeners --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Bindbrokersexposesbothlisteners(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.bindbrokersexposesbothlisteners(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.bindBrokersExposesBothListeners(kafkaImageTag)
}

clusterClientCanListTopicsOnFreshCluster() 🔗

ClusterClientCanListTopicsOnFreshCluster opens a franz-go-backed Client against a fresh cluster and asserts that ListTopics returns without error. A fresh KRaft cluster has no user topics, so the result may be empty — but the call itself must succeed, which proves module-runtime networking can reach the started broker service.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 cluster-client-can-list-topics-on-fresh-cluster --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Clusterclientcanlisttopicsonfreshcluster(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.clusterclientcanlisttopicsonfreshcluster(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.clusterClientCanListTopicsOnFreshCluster(kafkaImageTag)
}

confluent() 🔗

Confluent runs the three confluentinc/cp-kafka round-trip tests. Each test owns a fresh ConfluentCluster.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
confluentImageTag	String !	"8.2.0"	No description provided
parallel	Integer !	0	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 confluent --confluent-image-tag string --parallel integer

func (m *MyModule) Example(ctx context.Context, confluentImageTag string, parallel int)   {
	return dag.
			Tests().
			Confluent(ctx, confluentImageTag, parallel)
}

@function
async def example(confluentimagetag: str, parallel: int) -> None:
	return await (
		dag.tests()
		.confluent(confluentimagetag, parallel)
	)

@func()
async example(confluentImageTag: string, parallel: number): Promise<void> {
	return dag
		.tests()
		.confluent(confluentImageTag, parallel)
}

confluentClusterMtlsRoundTrip() 🔗

ConfluentClusterMtlsRoundTrip is the MTLS happy-path round-trip for Kafka.ConfluentCluster. Mirrors MtlsRoundTrip but on cp-kafka to rule out distro-specific differences in how client-cert challenge is handled.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
confluentImageTag	String !	"8.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 confluent-cluster-mtls-round-trip --confluent-image-tag string

func (m *MyModule) Example(ctx context.Context, confluentImageTag string)   {
	return dag.
			Tests().
			Confluentclustermtlsroundtrip(ctx, confluentImageTag)
}

@function
async def example(confluentimagetag: str) -> None:
	return await (
		dag.tests()
		.confluentclustermtlsroundtrip(confluentimagetag)
	)

@func()
async example(confluentImageTag: string): Promise<void> {
	return dag
		.tests()
		.confluentClusterMtlsRoundTrip(confluentImageTag)
}

confluentClusterProduceListTopicsRoundTrip() 🔗

ConfluentClusterProduceListTopicsRoundTrip is the PLAINTEXT happy-path smoke test for Kafka.ConfluentCluster (the cp-kafka image variant): produce a single raw record, then call ListTopics and assert the freshly-created topic shows up. Confluent Platform’s cp-kafka image uses the same KAFKA_* Scala-wrapper contract as Apache, so this single test pins down “cp-kafka actually serves traffic”.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
confluentImageTag	String !	"8.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 confluent-cluster-produce-list-topics-round-trip --confluent-image-tag string

func (m *MyModule) Example(ctx context.Context, confluentImageTag string)   {
	return dag.
			Tests().
			Confluentclusterproducelisttopicsroundtrip(ctx, confluentImageTag)
}

@function
async def example(confluentimagetag: str) -> None:
	return await (
		dag.tests()
		.confluentclusterproducelisttopicsroundtrip(confluentimagetag)
	)

@func()
async example(confluentImageTag: string): Promise<void> {
	return dag
		.tests()
		.confluentClusterProduceListTopicsRoundTrip(confluentImageTag)
}

confluentClusterTlsRoundTrip() 🔗

ConfluentClusterTlsRoundTrip is the TLS happy-path round-trip for Kafka.ConfluentCluster. Mirrors TlsRoundTrip but on cp-kafka to rule out distro-specific differences in keystore mounts, hostname verification, and SSL listener bring-up.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
confluentImageTag	String !	"8.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 confluent-cluster-tls-round-trip --confluent-image-tag string

func (m *MyModule) Example(ctx context.Context, confluentImageTag string)   {
	return dag.
			Tests().
			Confluentclustertlsroundtrip(ctx, confluentImageTag)
}

@function
async def example(confluentimagetag: str) -> None:
	return await (
		dag.tests()
		.confluentclustertlsroundtrip(confluentimagetag)
	)

@func()
async example(confluentImageTag: string): Promise<void> {
	return dag
		.tests()
		.confluentClusterTlsRoundTrip(confluentImageTag)
}

confluentSchemaRegistryMtlsRegisterLookupRoundTrip() 🔗

ConfluentSchemaRegistryMtlsRegisterLookupRoundTrip is the mTLS counterpart: the REST endpoint requires a client certificate and the registry presents its own leaf to the mTLS broker for the kafkastore connection, all rooted at one CA.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 confluent-schema-registry-mtls-register-lookup-round-trip --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Confluentschemaregistrymtlsregisterlookuproundtrip(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.confluentschemaregistrymtlsregisterlookuproundtrip(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.confluentSchemaRegistryMtlsRegisterLookupRoundTrip(kafkaImageTag)
}

confluentSchemaRegistryTlsRegisterLookupRoundTrip() 🔗

ConfluentSchemaRegistryTlsRegisterLookupRoundTrip is the canonical TLS test: a TLS cp-schema-registry terminates HTTPS on its REST endpoint and talks SSL to a TLS cluster’s brokers for the _schemas topic, and the round-trip succeeds over an HTTPS client verifying the registry cert against the shared CA.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 confluent-schema-registry-tls-register-lookup-round-trip --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Confluentschemaregistrytlsregisterlookuproundtrip(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.confluentschemaregistrytlsregisterlookuproundtrip(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.confluentSchemaRegistryTlsRegisterLookupRoundTrip(kafkaImageTag)
}

consumerGroupOnSingleBrokerWorks() 🔗

ConsumerGroupOnSingleBrokerWorks produces one record then consumes it back through a consumer group on a 1-broker cluster. A successful round-trip proves __consumer_offsets was created at the broker’s configured replication factor (1, after the system-topic env vars take effect). Without KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 the broker would refuse to create __consumer_offsets at the upstream default RF=3 and the group join would hang or error.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 consumer-group-on-single-broker-works --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Consumergrouponsinglebrokerworks(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.consumergrouponsinglebrokerworks(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.consumerGroupOnSingleBrokerWorks(kafkaImageTag)
}

createAndDeleteTopicRoundTrip() 🔗

CreateAndDeleteTopicRoundTrip exercises the create/list/delete cycle to confirm kadm wiring. The topic name is randomized so the test is repeatable against the same cluster and never collides with leftovers.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 create-and-delete-topic-round-trip --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Createanddeletetopicroundtrip(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.createanddeletetopicroundtrip(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.createAndDeleteTopicRoundTrip(kafkaImageTag)
}

dedicatedControllerAndBrokerProduceConsume() 🔗

DedicatedControllerAndBrokerProduceConsume verifies that the split controller+broker topology (introduced this increment) still supports a full produce/consume round-trip — i.e. the broker correctly joined the controller quorum over its WithServiceBinding alias.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 dedicated-controller-and-broker-produce-consume --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Dedicatedcontrollerandbrokerproduceconsume(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.dedicatedcontrollerandbrokerproduceconsume(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.dedicatedControllerAndBrokerProduceConsume(kafkaImageTag)
}

describeConsumerGroupReportsLag() 🔗

DescribeConsumerGroupReportsLag produces five records, consumes three of them through a committing consumer group, and asserts DescribeConsumerGroup reports the group in the Empty state with committed-offset lag of 2 (end offset 5 minus committed offset 3) on the single partition.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 describe-consumer-group-reports-lag --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Describeconsumergroupreportslag(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.describeconsumergroupreportslag(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.describeConsumerGroupReportsLag(kafkaImageTag)
}

describeTopicReportsPartitionsAndConfigs() 🔗

DescribeTopicReportsPartitionsAndConfigs creates a 3-partition RF=1 topic and asserts DescribeTopic reports the derived partition count / replication factor, one partition entry per partition, and a non-empty topic-level config set (proving the configs path is wired).

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 describe-topic-reports-partitions-and-configs --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Describetopicreportspartitionsandconfigs(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.describetopicreportspartitionsandconfigs(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.describeTopicReportsPartitionsAndConfigs(kafkaImageTag)
}

fiveControllerQuorumAccepted() 🔗

FiveControllerQuorumAccepted proves the constructor accepts a five-voter quorum (odd, > 3) and returns a *Cluster: resolving BootstrapServers forces the server-side constructor — validation, internal-CA minting of a leaf per controller, and the full container graph — to run without booting the containers, so the accept path is exercised cheaply. The 3-controller end-to-end quorum is covered by ThreeControllerQuorumProduceConsume.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 five-controller-quorum-accepted --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Fivecontrollerquorumaccepted(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.fivecontrollerquorumaccepted(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.fiveControllerQuorumAccepted(kafkaImageTag)
}

internalListenersAreEncrypted() 🔗

InternalListenersAreEncrypted spins up a 1+2 cluster with TLS on the external listener and creates an RF=2 topic. A successful produce → consume round-trip proves replication traffic flowed over the (always mTLS) INTERNAL inter-broker listener: without working internal mTLS, the second broker would never become an in-sync replica and the produce (with default acks=all-isr) would stall.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 internal-listeners-are-encrypted --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Internallistenersareencrypted(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.internallistenersareencrypted(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.internalListenersAreEncrypted(kafkaImageTag)
}

invalidControllerCountIsRejected() 🔗

InvalidControllerCountIsRejected pins the voter-count policy: a KRaft quorum wants an odd voter count for a clean majority, so even controller counts are rejected, and a sub-1 count is nonsensical. Both must fail at construction time with a clear error rather than spinning up a broken topology. (Controllers=0 can’t be exercised from the Go SDK — Dagger drops the zero value and applies the

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 invalid-controller-count-is-rejected --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Invalidcontrollercountisrejected(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.invalidcontrollercountisrejected(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.invalidControllerCountIsRejected(kafkaImageTag)
}

karapaceSchemaRegistryRegisterLookupRoundTrip() 🔗

KarapaceSchemaRegistryRegisterLookupRoundTrip is the PLAINTEXT happy-path test for Kafka.KarapaceSchemaRegistry: stand a Karapace service up next to a fresh cluster, then exercise register → lookup-by-id → lookup-latest-by-subject → list-subjects → set/get-compatibility → delete against its Confluent-compatible REST surface — mirroring SchemaRegistryRegisterLookupRoundTrip to prove the shared *SchemaRegistryClient drives Karapace unchanged.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 karapace-schema-registry-register-lookup-round-trip --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Karapaceschemaregistryregisterlookuproundtrip(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.karapaceschemaregistryregisterlookuproundtrip(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.karapaceSchemaRegistryRegisterLookupRoundTrip(kafkaImageTag)
}

karapaceSchemaRegistryTlsRegisterLookupRoundTrip() 🔗

KarapaceSchemaRegistryTlsRegisterLookupRoundTrip drives the Karapace TLS path (PEM REST listener + aiokafka SSL storage) end-to-end.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 karapace-schema-registry-tls-register-lookup-round-trip --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Karapaceschemaregistrytlsregisterlookuproundtrip(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.karapaceschemaregistrytlsregisterlookuproundtrip(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.karapaceSchemaRegistryTlsRegisterLookupRoundTrip(kafkaImageTag)
}

listConsumerGroupsReportsCommittedGroup() 🔗

ListConsumerGroupsReportsCommittedGroup produces a record, consumes it back through a committing consumer group, and asserts the group then appears in ListConsumerGroups. A fresh cluster reports no groups, so the group’s presence proves the join + commit reached __consumer_offsets.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 list-consumer-groups-reports-committed-group --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Listconsumergroupsreportscommittedgroup(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.listconsumergroupsreportscommittedgroup(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.listConsumerGroupsReportsCommittedGroup(kafkaImageTag)
}

mtlsRequiresClientCert() 🔗

MtlsRequiresClientCert points a TLS-only client (no keystore) at an MTLS broker and asserts the handshake fails. Confirms the broker’s client.auth=required setting is actually being honoured.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 mtls-requires-client-cert --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Mtlsrequiresclientcert(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.mtlsrequiresclientcert(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.mtlsRequiresClientCert(kafkaImageTag)
}

mtlsRoundTrip() 🔗

MtlsRoundTrip produces and consumes a single record over a mutual-TLS external listener. The broker presents its cert (signed by the server CA) and demands a client cert in return; the test client presents one signed by an independent client CA the broker is configured to trust.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 mtls-round-trip --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Mtlsroundtrip(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.mtlsroundtrip(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.mtlsRoundTrip(kafkaImageTag)
}

native() 🔗

Native runs every apache/kafka-native test as one group. It boots the three shared ApacheNativeClusters up front, fans the shared-cluster and fresh-cluster native tests across a par pool capped at parallel, and tears the shared clusters down on return.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided
parallel	Integer !	0	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 native --kafka-image-tag string --parallel integer

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string, parallel int)   {
	return dag.
			Tests().
			Native(ctx, kafkaImageTag, parallel)
}

@function
async def example(kafkaimagetag: str, parallel: int) -> None:
	return await (
		dag.tests()
		.native(kafkaimagetag, parallel)
	)

@func()
async example(kafkaImageTag: string, parallel: number): Promise<void> {
	return dag
		.tests()
		.native(kafkaImageTag, parallel)
}

oneControllerTwoBrokersReplicationFactorTwo() 🔗

OneControllerTwoBrokersReplicationFactorTwo spins up a 1+2 cluster and creates a replication-factor-2 topic so the produce path forces inter- broker replication. A successful round-trip proves brokers can reach each other over the engine network without explicit peer bindings.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 one-controller-two-brokers-replication-factor-two --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Onecontrollertwobrokersreplicationfactortwo(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.onecontrollertwobrokersreplicationfactortwo(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.oneControllerTwoBrokersReplicationFactorTwo(kafkaImageTag)
}

plaintextSecurityProfilesAreNonNil() 🔗

Return Type

Void !

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 plaintext-security-profiles-are-non-nil

func (m *MyModule) Example(ctx context.Context)   {
	return dag.
			Tests().
			Plaintextsecurityprofilesarenonnil(ctx)
}

@function
async def example() -> None:
	return await (
		dag.tests()
		.plaintextsecurityprofilesarenonnil()
	)

@func()
async example(): Promise<void> {
	return dag
		.tests()
		.plaintextSecurityProfilesAreNonNil()
}

produceConsumeRoundTripBase64() 🔗

ProduceConsumeRoundTripBase64 round-trips the same kind of binary payload through standard base64 (with padding).

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 produce-consume-round-trip-base-6-4 --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Produceconsumeroundtripbase64(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.produceconsumeroundtripbase64(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.produceConsumeRoundTripBase64(kafkaImageTag)
}

produceConsumeRoundTripHex() 🔗

ProduceConsumeRoundTripHex round-trips a binary payload through hex encoding. The non-UTF-8 bytes (including 0x00) verify that hex transports arbitrary binary safely.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 produce-consume-round-trip-hex --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Produceconsumeroundtriphex(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.produceconsumeroundtriphex(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.produceConsumeRoundTripHex(kafkaImageTag)
}

produceConsumeRoundTripRaw() 🔗

ProduceConsumeRoundTripRaw produces a single record with raw-encoded key and value, then consumes it back and asserts byte equality. The raw encoding round-trips Go strings verbatim, so the assertion is direct string equality.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 produce-consume-round-trip-raw --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Produceconsumeroundtripraw(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.produceconsumeroundtripraw(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.produceConsumeRoundTripRaw(kafkaImageTag)
}

produceRejectsUnknownEncoding() 🔗

ProduceRejectsUnknownEncoding verifies that a Produce call with a bogus encoding name fails fast rather than silently misbehaving.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 produce-rejects-unknown-encoding --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Producerejectsunknownencoding(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.producerejectsunknownencoding(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.produceRejectsUnknownEncoding(kafkaImageTag)
}

propertiesFileContainsBootstrapAndSecurityProtocol() 🔗

PropertiesFileContainsBootstrapAndSecurityProtocol verifies that the rendered Java client.properties file carries the bootstrap.servers list and a plaintext security.protocol entry — enough for the Apache Kafka CLI tools to pick up the connection settings.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 properties-file-contains-bootstrap-and-security-protocol --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Propertiesfilecontainsbootstrapandsecurityprotocol(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.propertiesfilecontainsbootstrapandsecurityprotocol(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.propertiesFileContainsBootstrapAndSecurityProtocol(kafkaImageTag)
}

propertiesFileContainsMtlsSettings() 🔗

PropertiesFileContainsMtlsSettings verifies that mTLS mode also renders the ssl.keystore.* triple referencing a keystore.p12 sidecar.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 properties-file-contains-mtls-settings --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Propertiesfilecontainsmtlssettings(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.propertiesfilecontainsmtlssettings(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.propertiesFileContainsMtlsSettings(kafkaImageTag)
}

propertiesFileContainsTlsSettings() 🔗

PropertiesFileContainsTlsSettings verifies the rendered Java client.properties carries security.protocol=SSL plus an ssl.truststore.* triple referencing a sidecar PKCS#12 file by basename.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 properties-file-contains-tls-settings --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Propertiesfilecontainstlssettings(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.propertiesfilecontainstlssettings(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.propertiesFileContainsTlsSettings(kafkaImageTag)
}

redpanda() 🔗

Redpanda runs the redpandadata/redpanda round-trip tests — the two Kafka-wire round-trips, the PLAINTEXT and TLS bundled-Schema-Registry round-trips, and the bundled-registry Stop-is-a-no-op lifecycle test.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
redpandaImageTag	String !	"v26.1.7"	No description provided
parallel	Integer !	0	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 redpanda --redpanda-image-tag string --parallel integer

func (m *MyModule) Example(ctx context.Context, redpandaImageTag string, parallel int)   {
	return dag.
			Tests().
			Redpanda(ctx, redpandaImageTag, parallel)
}

@function
async def example(redpandaimagetag: str, parallel: int) -> None:
	return await (
		dag.tests()
		.redpanda(redpandaimagetag, parallel)
	)

@func()
async example(redpandaImageTag: string, parallel: number): Promise<void> {
	return dag
		.tests()
		.redpanda(redpandaImageTag, parallel)
}

redpandaClusterProduceListTopicsRoundTrip() 🔗

RedpandaClusterProduceListTopicsRoundTrip is the PLAINTEXT happy-path round-trip for Kafka.RedpandaCluster: spin up a single-node Redpanda, create a topic, produce one record, then assert the freshly-created topic shows up in ListTopics. Pins down “redpanda actually serves Kafka-wire traffic on the external listener”.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
redpandaImageTag	String !	"v26.1.7"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 redpanda-cluster-produce-list-topics-round-trip --redpanda-image-tag string

func (m *MyModule) Example(ctx context.Context, redpandaImageTag string)   {
	return dag.
			Tests().
			Redpandaclusterproducelisttopicsroundtrip(ctx, redpandaImageTag)
}

@function
async def example(redpandaimagetag: str) -> None:
	return await (
		dag.tests()
		.redpandaclusterproducelisttopicsroundtrip(redpandaimagetag)
	)

@func()
async example(redpandaImageTag: string): Promise<void> {
	return dag
		.tests()
		.redpandaClusterProduceListTopicsRoundTrip(redpandaImageTag)
}

redpandaClusterTlsRoundTrip() 🔗

RedpandaClusterTlsRoundTrip is the TLS happy-path round-trip for Kafka.RedpandaCluster: spin up Redpanda with kafka_api_tls.enabled=true using PEM cert/key/CA mounted into /etc/redpanda/certs, then produce and consume one record over the TLS listener with the franz-go client verifying the broker leaf against the matching truststore.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
redpandaImageTag	String !	"v26.1.7"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 redpanda-cluster-tls-round-trip --redpanda-image-tag string

func (m *MyModule) Example(ctx context.Context, redpandaImageTag string)   {
	return dag.
			Tests().
			Redpandaclustertlsroundtrip(ctx, redpandaImageTag)
}

@function
async def example(redpandaimagetag: str) -> None:
	return await (
		dag.tests()
		.redpandaclustertlsroundtrip(redpandaimagetag)
	)

@func()
async example(redpandaImageTag: string): Promise<void> {
	return dag
		.tests()
		.redpandaClusterTlsRoundTrip(redpandaImageTag)
}

redpandaSchemaRegistryBundledStopIsNoOp() 🔗

RedpandaSchemaRegistryBundledStopIsNoOp pins the bundled-registry lifecycle contract: a bundled *SchemaRegistry shares the broker service, so sr.Stop must be a no-op — the cluster owns teardown via cluster.Stop. This runs a register/lookup round-trip, calls sr.Stop, then exercises the registry again through the same handle. If sr.Stop had torn down the shared broker service, that follow-up call would fail.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
redpandaImageTag	String !	"v26.1.7"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 redpanda-schema-registry-bundled-stop-is-no-op --redpanda-image-tag string

func (m *MyModule) Example(ctx context.Context, redpandaImageTag string)   {
	return dag.
			Tests().
			Redpandaschemaregistrybundledstopisnoop(ctx, redpandaImageTag)
}

@function
async def example(redpandaimagetag: str) -> None:
	return await (
		dag.tests()
		.redpandaschemaregistrybundledstopisnoop(redpandaimagetag)
	)

@func()
async example(redpandaImageTag: string): Promise<void> {
	return dag
		.tests()
		.redpandaSchemaRegistryBundledStopIsNoOp(redpandaImageTag)
}

redpandaSchemaRegistryRegisterLookupRoundTrip() 🔗

RedpandaSchemaRegistryRegisterLookupRoundTrip is the PLAINTEXT happy-path test for RedpandaCluster.SchemaRegistry: rpk redpanda start runs a Schema Registry inside the broker process on :8081, so this registers a schema against cluster.SchemaRegistry() and asserts the lookup-by-id round-trip — proving the bundled SR is reachable and interchangeable with the separate-container ConfluentSchemaRegistry. The SR service is the broker itself, so cluster.Stop tears it down — sr.Stop is a no-op for a bundled registry.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
redpandaImageTag	String !	"v26.1.7"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 redpanda-schema-registry-register-lookup-round-trip --redpanda-image-tag string

func (m *MyModule) Example(ctx context.Context, redpandaImageTag string)   {
	return dag.
			Tests().
			Redpandaschemaregistryregisterlookuproundtrip(ctx, redpandaImageTag)
}

@function
async def example(redpandaimagetag: str) -> None:
	return await (
		dag.tests()
		.redpandaschemaregistryregisterlookuproundtrip(redpandaimagetag)
	)

@func()
async example(redpandaImageTag: string): Promise<void> {
	return dag
		.tests()
		.redpandaSchemaRegistryRegisterLookupRoundTrip(redpandaImageTag)
}

redpandaSchemaRegistryTlsRegisterLookupRoundTrip() 🔗

RedpandaSchemaRegistryTlsRegisterLookupRoundTrip is the TLS-cluster counterpart of RedpandaSchemaRegistryRegisterLookupRoundTrip. A TLS Redpanda cluster terminates HTTPS on its bundled Schema Registry REST endpoint, reusing the broker’s server leaf (configured through the separately-rendered redpanda.yaml schema_registry_api_tls block), so this exercises that YAML path end-to-end and drives register/lookup over HTTPS, verifying the SR cert against the cluster CA truststore.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
redpandaImageTag	String !	"v26.1.7"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 redpanda-schema-registry-tls-register-lookup-round-trip --redpanda-image-tag string

func (m *MyModule) Example(ctx context.Context, redpandaImageTag string)   {
	return dag.
			Tests().
			Redpandaschemaregistrytlsregisterlookuproundtrip(ctx, redpandaImageTag)
}

@function
async def example(redpandaimagetag: str) -> None:
	return await (
		dag.tests()
		.redpandaschemaregistrytlsregisterlookuproundtrip(redpandaimagetag)
	)

@func()
async example(redpandaImageTag: string): Promise<void> {
	return dag
		.tests()
		.redpandaSchemaRegistryTlsRegisterLookupRoundTrip(redpandaImageTag)
}

schemaRegistry() 🔗

SchemaRegistry runs the Schema Registry tests — ConfluentSchemaRegistry, ApicurioSchemaRegistry, and KarapaceSchemaRegistry — as one group. Each test owns the cluster (and, for the round-trips, the registry service) it boots, so the group’s only lifetime guarantee is that both are torn down once it returns.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided
parallel	Integer !	0	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 schema-registry --kafka-image-tag string --parallel integer

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string, parallel int)   {
	return dag.
			Tests().
			Schemaregistry(ctx, kafkaImageTag, parallel)
}

@function
async def example(kafkaimagetag: str, parallel: int) -> None:
	return await (
		dag.tests()
		.schemaregistry(kafkaimagetag, parallel)
	)

@func()
async example(kafkaImageTag: string, parallel: number): Promise<void> {
	return dag
		.tests()
		.schemaRegistry(kafkaImageTag, parallel)
}

schemaRegistryFramedProduceConsumeRoundTrip() 🔗

SchemaRegistryFramedProduceConsumeRoundTrip exercises the data path of the Client with Confluent wire-format framing: register a schema to get an ID, produce a record whose value is framed with that ID, then consume with schemaRegistryAware=true and assert the parsed ID matches and the value bytes are stripped back to the original payload.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 schema-registry-framed-produce-consume-round-trip --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Schemaregistryframedproduceconsumeroundtrip(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.schemaregistryframedproduceconsumeroundtrip(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.schemaRegistryFramedProduceConsumeRoundTrip(kafkaImageTag)
}

schemaRegistryJsonframedProduceConsumeRoundTrip() 🔗

SchemaRegistryJSONFramedProduceConsumeRoundTrip composes the framing primitive (valueSchemaID) with the JSON serde (valueSerializeAs / valueDeserializeAs). A JSON document is registered as a JSON-schema-typed subject, produced with both opts on, then consumed with both opts on; the asserted invariant is byte-equality of the consumed value to the canonical JSON form of the original input — proving frame strip and JSON validation compose without corrupting the payload.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 schema-registry-jsonframed-produce-consume-round-trip --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Schemaregistryjsonframedproduceconsumeroundtrip(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.schemaregistryjsonframedproduceconsumeroundtrip(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.schemaRegistryJsonframedProduceConsumeRoundTrip(kafkaImageTag)
}

schemaRegistryJsonserializeRejectsMalformedInput() 🔗

SchemaRegistryJSONSerializeRejectsMalformedInput pins the up-front validation contract of valueSerializeAs=“JSON”: Produce must reject a malformed JSON payload before any broker I/O. dag.Kafka().Client(…) builds without I/O, so no cluster boots — the failure is purely a payload-validation failure on the canonicalising serializer.

Return Type

Void !

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 schema-registry-jsonserialize-rejects-malformed-input

func (m *MyModule) Example(ctx context.Context)   {
	return dag.
			Tests().
			Schemaregistryjsonserializerejectsmalformedinput(ctx)
}

@function
async def example() -> None:
	return await (
		dag.tests()
		.schemaregistryjsonserializerejectsmalformedinput()
	)

@func()
async example(): Promise<void> {
	return dag
		.tests()
		.schemaRegistryJsonserializeRejectsMalformedInput()
}

schemaRegistryPlaintextConsumeUnframed() 🔗

SchemaRegistryPlaintextConsumeUnframed verifies the negative path: a record produced without framing, consumed with schemaRegistryAware=true, must surface ValueSchemaID=0 and pass the value bytes through unchanged.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 schema-registry-plaintext-consume-unframed --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Schemaregistryplaintextconsumeunframed(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.schemaregistryplaintextconsumeunframed(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.schemaRegistryPlaintextConsumeUnframed(kafkaImageTag)
}

schemaRegistryRegisterLookupRoundTrip() 🔗

SchemaRegistryRegisterLookupRoundTrip is the PLAINTEXT happy-path test for Kafka.ConfluentSchemaRegistry: stand a cp-schema-registry up next to a fresh cluster, then exercise register → lookup-by-id → lookup-latest-by-subject → list-subjects → set/get-compatibility → delete against it — covering every SchemaRegistryClient operation.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 schema-registry-register-lookup-round-trip --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Schemaregistryregisterlookuproundtrip(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.schemaregistryregisterlookuproundtrip(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.schemaRegistryRegisterLookupRoundTrip(kafkaImageTag)
}

schemaRegistryRejectsClusterModeMismatch() 🔗

SchemaRegistryRejectsClusterModeMismatch pins the mode-match contract: a registry’s security profile must match its backing cluster’s client-listener mode, so the registry’s kafka-storage connection authenticates against the broker. Here a PLAINTEXT registry security is paired with a TLS cluster and must be rejected with an error naming both modes. The constructor errors before any service boots, so the TLS cluster never has to start.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 schema-registry-rejects-cluster-mode-mismatch --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Schemaregistryrejectsclustermodemismatch(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.schemaregistryrejectsclustermodemismatch(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.schemaRegistryRejectsClusterModeMismatch(kafkaImageTag)
}

singleNodeClusterStarts() 🔗

SingleNodeClusterStarts spins up the smallest split-role cluster (one controller + one broker) and forces the server-side Cluster constructor to run by resolving BootstrapServers, asserting only that the broker hostname is non-empty. End-to-end reachability is covered by sibling tests that exercise ListTopics / produce / consume.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 single-node-cluster-starts --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Singlenodeclusterstarts(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.singlenodeclusterstarts(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.singleNodeClusterStarts(kafkaImageTag)
}

threeControllerQuorumProduceConsume() 🔗

ThreeControllerQuorumProduceConsume stands up a real three-node KRaft controller quorum (plus one broker) and drives a produce → consume round-trip through it, mirroring the single-controller round-trip. A successful round-trip proves the three controllers formed a quorum and elected a leader over their (always-mTLS) CONTROLLER listeners — which in turn proves the internal CA minted a verifying leaf for every controller host and that all three discovered each other over session-wide DNS with no controller-to-controller WithServiceBinding. Cluster.Stop then tears down all three controller services plus the broker.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 three-controller-quorum-produce-consume --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Threecontrollerquorumproduceconsume(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.threecontrollerquorumproduceconsume(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.threeControllerQuorumProduceConsume(kafkaImageTag)
}

tlsClientWithWrongCaFails() 🔗

TlsClientWithWrongCaFails verifies that pointing the client at a truststore for an unrelated CA fails the handshake — i.e. the broker is genuinely presenting a cert chained to its own CA, not skipping verification.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 tls-client-with-wrong-ca-fails --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Tlsclientwithwrongcafails(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.tlsclientwithwrongcafails(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.tlsClientWithWrongCaFails(kafkaImageTag)
}

tlsClusterStarts() 🔗

TlsClusterStarts forces the lazy Cluster construction to run under TlsServerSecurity and confirms BootstrapServers reports a non-empty, non-zero-port broker address. No client connection attempted — this proves caller’s CA loads, leaf signing succeeds, the keystore mounts, and the broker doesn’t crash on startup.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 tls-cluster-starts --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Tlsclusterstarts(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.tlsclusterstarts(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.tlsClusterStarts(kafkaImageTag)
}

tlsRoundTrip() 🔗

TlsRoundTrip produces and consumes a single record over a TLS-only external listener with TlsClientSecurity holding the CA’s truststore. Exercises: SAN matching the bootstrap address, kgo dialer + TLS, broker SSL listener, end-to-end encryption.

Return Type

Void !

Arguments

Name	Type	Default Value	Description
kafkaImageTag	String !	"4.2.0"	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@a17bf411228ae3d544c7c18bcf08dd04c295d453 call \
 tls-round-trip --kafka-image-tag string

func (m *MyModule) Example(ctx context.Context, kafkaImageTag string)   {
	return dag.
			Tests().
			Tlsroundtrip(ctx, kafkaImageTag)
}

@function
async def example(kafkaimagetag: str) -> None:
	return await (
		dag.tests()
		.tlsroundtrip(kafkaimagetag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.tlsRoundTrip(kafkaImageTag)
}