kafkatest

Documentation

Index

• Constants
• Variables
• type KafkaCluster
  • func NewKafkaCluster(ctx context.Context, setters ...Option) (*KafkaCluster, error)
  • func NewKafkaClusterWithOptions(ctx context.Context, origOptions *Options) (*KafkaCluster, error)
  • func (kc *KafkaCluster) ClientTLSConfig() *tls.Config
  • func (kc *KafkaCluster) GetKafkaHost(ctx context.Context) (string, error)
  • func (kc *KafkaCluster) GetKafkaHostAndPort(ctx context.Context) (string, error)
  • func (kc *KafkaCluster) StopCluster(ctx context.Context)
• type Option
  • func WithContainerShutdownTimeout(timeout time.Duration) Option
  • func WithContainerStartTimeout(timeout time.Duration) Option
  • func WithKafkaClientPort(port int) Option
  • func WithKafkaImage(image string) Option
  • func WithKafkaWaitStrategy(strategy wait.Strategy) Option
  • func WithLogger(logger testcontainers.Logging) Option
  • func WithNetworkCreateTimeout(timeout time.Duration) Option
  • func WithNetworkPrefix(prefix string) Option
  • func WithNetworkShutdownTimeout(timeout time.Duration) Option
  • func WithTLS(tls bool) Option
  • func WithZookeeperImage(image string) Option
  • func WithZookeeperWaitStrategy(strategy wait.Strategy) Option
• type Options

Constants

const (
	// DefaultZookeeperImage is the default image for the zookeeper container.
	DefaultZookeeperImage = "confluentinc/cp-zookeeper:7.2.2"

	// DefaultKafkaImage is the default image for the kafka container.
	DefaultKafkaImage = "confluentinc/cp-kafka:7.2.2"

	// DefaultKafkaClientPort is the default exposed port of the kafka container.
	DefaultKafkaClientPort = 9093

	// DefaultNetworkPrefix is the default prefix for the network used to connect containers.
	DefaultNetworkPrefix = "kafka-cluster"

	// DefaultTLS is the default for weather or not the cluster should use encrypted connections.
	DefaultTLS = false

	// DefaultContainerStartTimeout is the default timeout for starting containers
	DefaultContainerStartTimeout = time.Minute

	// DefaultContainerShutdownTimeout is the default timeout for shutting down containers
	DefaultContainerShutdownTimeout = time.Minute

	// DefaultNetworkCreateTimeout is the default timeout for creating the container network
	DefaultNetworkCreateTimeout = time.Minute

	// DefaultNetworkShutdownTimeout is the default timeout for shutting down the container network
	DefaultNetworkShutdownTimeout = time.Minute
)

Variables

var (
	// DefaultZookeeperWaitStrategy is the default wait.Strategy for the zookeeper container.
	DefaultZookeeperWaitStrategy = wait.ForLog("binding to port")

	// DefaultKafkaWaitStrategy is the default wait.Strategy for the kafka container.
	DefaultKafkaWaitStrategy = wait.ForLog("Initialized broker")

	// DefaultLogger is the default logger.
	DefaultLogger = testcontainers.Logger
)

Types

type KafkaCluster

type KafkaCluster struct {
	// contains filtered or unexported fields
}

func NewKafkaCluster

func NewKafkaCluster(ctx context.Context, setters ...Option) (*KafkaCluster, error)

NewKafkaCluster creates and starts a new kafka cluster via the given Option objects.

func NewKafkaClusterWithOptions

func NewKafkaClusterWithOptions(ctx context.Context, origOptions *Options) (*KafkaCluster, error)

NewKafkaClusterWithOptions creates and starts a new kafka cluster via the given Options object.

func (*KafkaCluster) ClientTLSConfig

func (kc *KafkaCluster) ClientTLSConfig() *tls.Config

ClientTLSConfig returns the tls.Config which can be used by clients to enable WithTLS encryption with the cluster, if configured.

func (*KafkaCluster) GetKafkaHost

func (kc *KafkaCluster) GetKafkaHost(ctx context.Context) (string, error)

GetKafkaHost gets the kafka host, so it can be accessed from outside the container.

func (*KafkaCluster) GetKafkaHostAndPort

func (kc *KafkaCluster) GetKafkaHostAndPort(ctx context.Context) (string, error)

GetKafkaHostAndPort gets the kafka host:port, so it can be accessed from outside the container.

func (*KafkaCluster) StopCluster

func (kc *KafkaCluster) StopCluster(ctx context.Context)

StopCluster gracefully stops the cluster.

type Option

type Option func(*Options)

Option represents an option for a Kafka cluster.

func WithContainerShutdownTimeout

func WithContainerShutdownTimeout(timeout time.Duration) Option

WithContainerShutdownTimeout sets the timeout for shutting down containers. The default is DefaultContainerShutdownTimeout.

func WithContainerStartTimeout

func WithContainerStartTimeout(timeout time.Duration) Option

WithContainerStartTimeout sets the timeout for starting containers. The default is DefaultContainerStartTimeout.

func WithKafkaClientPort

func WithKafkaClientPort(port int) Option

WithKafkaClientPort sets exposed port of the kafka container. The default is DefaultKafkaClientPort.

func WithKafkaImage

func WithKafkaImage(image string) Option

WithKafkaImage sets image for the kafka container. The default is DefaultKafkaImage.

func WithKafkaWaitStrategy

func WithKafkaWaitStrategy(strategy wait.Strategy) Option

WithKafkaWaitStrategy sets the wait.Strategy for the kafka container. The default is DefaultKafkaWaitStrategy.

func WithLogger

func WithLogger(logger testcontainers.Logging) Option

WithLogger sets the logger. The default is DefaultLogger.

func WithNetworkCreateTimeout

func WithNetworkCreateTimeout(timeout time.Duration) Option

WithNetworkCreateTimeout sets the timeout for creating the container network. The default is DefaultNetworkCreateTimeout.

func WithNetworkPrefix

func WithNetworkPrefix(prefix string) Option

WithNetworkPrefix sets prefix for the network used to connect containers. The default is DefaultNetworkPrefix.

func WithNetworkShutdownTimeout

func WithNetworkShutdownTimeout(timeout time.Duration) Option

WithNetworkShutdownTimeout sets the timeout for shutting down the container network. The default is DefaultNetworkShutdownTimeout.

func WithTLS

func WithTLS(tls bool) Option

WithTLS sets weather or not the cluster should use encrypted connections. The default is DefaultTLS.

func WithZookeeperImage

func WithZookeeperImage(image string) Option

WithZookeeperImage sets image for the zookeeper container. The default is DefaultZookeeperImage.

func WithZookeeperWaitStrategy

func WithZookeeperWaitStrategy(strategy wait.Strategy) Option

WithZookeeperWaitStrategy sets the wait.Strategy for the zookeeper container. The default is DefaultZookeeperWaitStrategy.

type Options

type Options struct {
	ZookeeperImage  string
	KafkaImage      string
	KafkaClientPort int

	NetworkPrefix string
	TLS           bool

	ZookeeperWaitStrategy wait.Strategy
	KafkaWaitStrategy     wait.Strategy

	ContainerStartTimeout    time.Duration
	ContainerShutdownTimeout time.Duration
	NetworkCreateTimeout     time.Duration
	NetworkShutdownTimeout   time.Duration

	Logger testcontainers.Logging
}

Options bundles all available configuration properties for a KafkaCluster.