tests
Installation
Entrypoint
Types
Tests

tests

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

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

- main.go — Tests struct + the All() orchestrator.
- 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 round-trip tests.

Installation

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

Entrypoint

Return Type

Tests

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@d38cd46573b89a2722fb888c91ced6495d6825e9 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 in parallel. Each test owns its own cluster lifecycle: it builds a cluster on entry and tears it down via defer cluster.Stop(ctx) so the broker Container.asService spans close the moment the test work is done rather than running out to the parent parallel group’s lifetime. Reusing clusters across tests within a parallel run is a follow-up — early attempts amplified Dagger’s service-binding propagation race into intermittent “lookup broker-… no such host” failures, so this PR keeps the per-test isolation that’s already proven.

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) — so callers can verify the module against a newer Kafka release without first changing main.go. The default matches the Apache constructors’ own default.

confluentImageTag is the independent knob for the cp-kafka tests (ConfluentCluster). Confluent Platform versioning is not aligned with Apache’s release numbering (CP 8.2.0 bundles Kafka 4.2.0), so this gets its own argument and its own default.

redpandaImageTag is the independent knob for the redpandadata/redpanda tests (RedpandaCluster). Redpanda has its own release cadence with no alignment to Apache or Confluent numbering, so it gets its own argument and its own default.

parallel caps how many tests run concurrently inside this suite. Defaults to 1 (sequential) to mirror go test package-level semantics; pass 0 to fan out every test with no limit, or any positive integer to opt into a specific level of concurrency.

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 !	1	No description provided

Example

dagger -m github.com/z5labs/devex/daggerverse/kafka/tests@d38cd46573b89a2722fb888c91ced6495d6825e9 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(kafka_image_tag: str, confluent_image_tag: str, redpanda_image_tag: str, parallel: int) -> None:
	return await (
		dag.tests()
		.all(kafka_image_tag, confluent_image_tag, redpanda_image_tag, 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@d38cd46573b89a2722fb888c91ced6495d6825e9 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(kafka_image_tag: str) -> None:
	return await (
		dag.tests()
		.apache_cluster_mtls_round_trip(kafka_image_tag)
	)

@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@d38cd46573b89a2722fb888c91ced6495d6825e9 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(kafka_image_tag: str) -> None:
	return await (
		dag.tests()
		.apache_cluster_produce_list_topics_round_trip(kafka_image_tag)
	)

@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@d38cd46573b89a2722fb888c91ced6495d6825e9 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(kafka_image_tag: str) -> None:
	return await (
		dag.tests()
		.apache_cluster_tls_round_trip(kafka_image_tag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.apacheClusterTlsRoundTrip(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@d38cd46573b89a2722fb888c91ced6495d6825e9 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(kafka_image_tag: str) -> None:
	return await (
		dag.tests()
		.auto_create_topics_disabled(kafka_image_tag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.autoCreateTopicsDisabled(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@d38cd46573b89a2722fb888c91ced6495d6825e9 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(kafka_image_tag: str) -> None:
	return await (
		dag.tests()
		.bind_brokers_exposes_both_listeners(kafka_image_tag)
	)

@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@d38cd46573b89a2722fb888c91ced6495d6825e9 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(kafka_image_tag: str) -> None:
	return await (
		dag.tests()
		.cluster_client_can_list_topics_on_fresh_cluster(kafka_image_tag)
	)

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

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@d38cd46573b89a2722fb888c91ced6495d6825e9 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(confluent_image_tag: str) -> None:
	return await (
		dag.tests()
		.confluent_cluster_mtls_round_trip(confluent_image_tag)
	)

@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@d38cd46573b89a2722fb888c91ced6495d6825e9 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(confluent_image_tag: str) -> None:
	return await (
		dag.tests()
		.confluent_cluster_produce_list_topics_round_trip(confluent_image_tag)
	)

@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@d38cd46573b89a2722fb888c91ced6495d6825e9 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(confluent_image_tag: str) -> None:
	return await (
		dag.tests()
		.confluent_cluster_tls_round_trip(confluent_image_tag)
	)

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

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@d38cd46573b89a2722fb888c91ced6495d6825e9 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(kafka_image_tag: str) -> None:
	return await (
		dag.tests()
		.consumer_group_on_single_broker_works(kafka_image_tag)
	)

@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@d38cd46573b89a2722fb888c91ced6495d6825e9 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(kafka_image_tag: str) -> None:
	return await (
		dag.tests()
		.create_and_delete_topic_round_trip(kafka_image_tag)
	)

@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@d38cd46573b89a2722fb888c91ced6495d6825e9 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(kafka_image_tag: str) -> None:
	return await (
		dag.tests()
		.dedicated_controller_and_broker_produce_consume(kafka_image_tag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.dedicatedControllerAndBrokerProduceConsume(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@d38cd46573b89a2722fb888c91ced6495d6825e9 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(kafka_image_tag: str) -> None:
	return await (
		dag.tests()
		.internal_listeners_are_encrypted(kafka_image_tag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.internalListenersAreEncrypted(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@d38cd46573b89a2722fb888c91ced6495d6825e9 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(kafka_image_tag: str) -> None:
	return await (
		dag.tests()
		.mtls_requires_client_cert(kafka_image_tag)
	)

@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@d38cd46573b89a2722fb888c91ced6495d6825e9 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(kafka_image_tag: str) -> None:
	return await (
		dag.tests()
		.mtls_round_trip(kafka_image_tag)
	)

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

multiControllerIsRejected() 🔗

MultiControllerIsRejected pins the current contract: this story only supports a single-controller quorum (controllers=1), and the constructor must reject any larger value with a clear error rather than silently spinning up a broken topology. Multi-controller HA is gated behind a follow-up story; see daggerverse/kafka/README.md.

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@d38cd46573b89a2722fb888c91ced6495d6825e9 call \
 multi-controller-is-rejected --kafka-image-tag string

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

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

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

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@d38cd46573b89a2722fb888c91ced6495d6825e9 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(kafka_image_tag: str) -> None:
	return await (
		dag.tests()
		.one_controller_two_brokers_replication_factor_two(kafka_image_tag)
	)

@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@d38cd46573b89a2722fb888c91ced6495d6825e9 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()
		.plaintext_security_profiles_are_non_nil()
	)

@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@d38cd46573b89a2722fb888c91ced6495d6825e9 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(kafka_image_tag: str) -> None:
	return await (
		dag.tests()
		.produce_consume_round_trip_base64(kafka_image_tag)
	)

@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@d38cd46573b89a2722fb888c91ced6495d6825e9 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(kafka_image_tag: str) -> None:
	return await (
		dag.tests()
		.produce_consume_round_trip_hex(kafka_image_tag)
	)

@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@d38cd46573b89a2722fb888c91ced6495d6825e9 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(kafka_image_tag: str) -> None:
	return await (
		dag.tests()
		.produce_consume_round_trip_raw(kafka_image_tag)
	)

@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@d38cd46573b89a2722fb888c91ced6495d6825e9 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(kafka_image_tag: str) -> None:
	return await (
		dag.tests()
		.produce_rejects_unknown_encoding(kafka_image_tag)
	)

@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@d38cd46573b89a2722fb888c91ced6495d6825e9 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(kafka_image_tag: str) -> None:
	return await (
		dag.tests()
		.properties_file_contains_bootstrap_and_security_protocol(kafka_image_tag)
	)

@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@d38cd46573b89a2722fb888c91ced6495d6825e9 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(kafka_image_tag: str) -> None:
	return await (
		dag.tests()
		.properties_file_contains_mtls_settings(kafka_image_tag)
	)

@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@d38cd46573b89a2722fb888c91ced6495d6825e9 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(kafka_image_tag: str) -> None:
	return await (
		dag.tests()
		.properties_file_contains_tls_settings(kafka_image_tag)
	)

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

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@d38cd46573b89a2722fb888c91ced6495d6825e9 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(redpanda_image_tag: str) -> None:
	return await (
		dag.tests()
		.redpanda_cluster_produce_list_topics_round_trip(redpanda_image_tag)
	)

@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@d38cd46573b89a2722fb888c91ced6495d6825e9 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(redpanda_image_tag: str) -> None:
	return await (
		dag.tests()
		.redpanda_cluster_tls_round_trip(redpanda_image_tag)
	)

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

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@d38cd46573b89a2722fb888c91ced6495d6825e9 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(kafka_image_tag: str) -> None:
	return await (
		dag.tests()
		.single_node_cluster_starts(kafka_image_tag)
	)

@func()
async example(kafkaImageTag: string): Promise<void> {
	return dag
		.tests()
		.singleNodeClusterStarts(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@d38cd46573b89a2722fb888c91ced6495d6825e9 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(kafka_image_tag: str) -> None:
	return await (
		dag.tests()
		.tls_client_with_wrong_ca_fails(kafka_image_tag)
	)

@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@d38cd46573b89a2722fb888c91ced6495d6825e9 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(kafka_image_tag: str) -> None:
	return await (
		dag.tests()
		.tls_cluster_starts(kafka_image_tag)
	)

@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@d38cd46573b89a2722fb888c91ced6495d6825e9 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(kafka_image_tag: str) -> None:
	return await (
		dag.tests()
		.tls_round_trip(kafka_image_tag)
	)

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