kafka

May 5, 2020

Changes in this version

+ type ContainerConnectionConfig struct

+ Brokers []string

+ KafkaVersion string

+ func StartKafkaContainer(options ContainerOptions) (testcontainers.Container, *ContainerConnectionConfig, testcontainers.Container, ...)

+ func StartStandaloneKafkaContainer(options ContainerOptions) (testcontainers.Container, *ContainerConnectionConfig, error)

+ type ContainerOptions struct

+ StartZookeeper bool

Apr 25, 2020

Changes in this version

+ type Consumer struct

+ Messages chan *sarama.ConsumerMessage

+ func ConsumeGroup(ctx context.Context, options ConsumerOptions) (*Consumer, *sync.WaitGroup, error)

+ func (consumer *Consumer) Cleanup(sarama.ConsumerGroupSession) error

+ func (consumer *Consumer) Close() error

+ func (consumer *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

+ func (consumer *Consumer) Setup(sarama.ConsumerGroupSession) error

+ type ConsumerOptions struct

+ Brokers []string

+ Group string

+ Topics []string

+ Version string

+ type Producer struct

+ AccessLogProducer sarama.AsyncProducer

+ DataCollector sarama.SyncProducer

+ func CreateProducer(ctx context.Context, options ProducerOptions) (*Producer, error)

+ func (p *Producer) Close() error

+ func (p *Producer) Send(topic string, pkey string, entry sarama.Encoder)

+ type ProducerOptions struct

+ Brokers []string

+ Group string

+ Topics []string

+ Version string