test

Documentation

Index

• func AssertEquals(t *testing.T, a, b []*confluent.Message)
• func AssertMessageEquals(t *testing.T, m1, m2 *confluent.Message)
• func AssertMessagesinTopic(t *testing.T, topic string, msgs []*confluent.Message)
• func ConsumeMessages(t *testing.T, topic string) []*confluent.Message
• func CreateTopics(t *testing.T, nbTopic int, nbPartitions []int, prefix string) []string
• func FullMessage(topic string, key, value interface{}, epoch int64, targetTopic string, ...) *confluent.Message
• func GetConsumerConfig(prefix string) *confluent.ConfigMap
• func GetProducerConfig() *confluent.ConfigMap
• func Message(topic string, key, value interface{}, epoch int64) *confluent.Message
• func Messages(topic string, epochs []int64) []*confluent.Message
• func NewKafkaStore(t *testing.T, nbTopic int, nbPartitions []int) (store *kafka_store.Store, topics []string)
• func NewKafkaStoreFromTopics(t *testing.T, topics []string) *kafka_store.Store
• func PrintMessages(t *testing.T, prefix string, msgs []*confluent.Message)
• func ProduceMessages(t *testing.T, messages []*confluent.Message)
• func RandomTopicName(prefix string) string

Functions

func AssertEquals

func AssertEquals(t *testing.T, a, b []*confluent.Message)

Compares 2 slices of kafka messages

func AssertMessageEquals

func AssertMessageEquals(t *testing.T, m1, m2 *confluent.Message)

Compares 2 kafka messages

func AssertMessagesinTopic

func AssertMessagesinTopic(t *testing.T, topic string, msgs []*confluent.Message)

Verifies if specified messages are in the topic

func ConsumeMessages

func ConsumeMessages(t *testing.T, topic string) []*confluent.Message

Get messages in the topic (it will return after 10s)

func CreateTopics

func CreateTopics(t *testing.T, nbTopic int, nbPartitions []int, prefix string) []string

Creates topics based on the array of nbPartitions param

func FullMessage

func FullMessage(topic string, key, value interface{}, epoch int64, targetTopic string, targetKey interface{}) *confluent.Message

FullMessage creates a kafka message with more details with scheduler headers

func GetConsumerConfig

func GetConsumerConfig(prefix string) *confluent.ConfigMap

func GetProducerConfig

func GetProducerConfig() *confluent.ConfigMap

func Message

func Message(topic string, key, value interface{}, epoch int64) *confluent.Message

Creates a kafka.Message

func Messages

func Messages(topic string, epochs []int64) []*confluent.Message

Creates multiple kafka.Messages

func NewKafkaStore

func NewKafkaStore(t *testing.T, nbTopic int, nbPartitions []int) (store *kafka_store.Store, topics []string)

func NewKafkaStoreFromTopics

func NewKafkaStoreFromTopics(t *testing.T, topics []string) *kafka_store.Store

func PrintMessages

func PrintMessages(t *testing.T, prefix string, msgs []*confluent.Message)

func ProduceMessages

func ProduceMessages(t *testing.T, messages []*confluent.Message)

func RandomTopicName

func RandomTopicName(prefix string) string

creates a random topic name, each test got a different topic name