kafka

Documentation

Index

• func WithClusterID(clusterID string) testcontainers.CustomizeRequestOption
• type KafkaContainer
  • func RunContainer(ctx context.Context, opts ...testcontainers.ContainerCustomizer) (*KafkaContainer, error)
  • func (kc *KafkaContainer) Brokers(ctx context.Context) ([]string, error)

Examples

• RunContainer

Functions

func WithClusterID

func WithClusterID(clusterID string) testcontainers.CustomizeRequestOption

Types

type KafkaContainer

type KafkaContainer struct {
	testcontainers.Container
	ClusterID string
}

KafkaContainer represents the Kafka container type used in the module

func RunContainer

func RunContainer(ctx context.Context, opts ...testcontainers.ContainerCustomizer) (*KafkaContainer, error)

RunContainer creates an instance of the Kafka container type

Example

// runKafkaContainer {
ctx := context.Background()

kafkaContainer, err := kafka.RunContainer(ctx,
	kafka.WithClusterID("test-cluster"),
	testcontainers.WithImage("confluentinc/confluent-local:7.5.0"),
)
if err != nil {
	log.Fatalf("failed to start container: %s", err)
}

// Clean up the container after
defer func() {
	if err := kafkaContainer.Terminate(ctx); err != nil {
		log.Fatalf("failed to terminate container: %s", err)
	}
}()
// }

state, err := kafkaContainer.State(ctx)
if err != nil {
	log.Fatalf("failed to get container state: %s", err) // nolint:gocritic
}

fmt.Println(kafkaContainer.ClusterID)
fmt.Println(state.Running)

Output:

test-cluster
true

func (*KafkaContainer) Brokers

func (kc *KafkaContainer) Brokers(ctx context.Context) ([]string, error)

Brokers retrieves the broker connection strings from Kafka with only one entry, defined by the exposed public port.