kafka

Documentation

Index

• Constants
• func AdminConfig() *sarama.Config
• func BootstrapServersArray(bootstrapServers string) []string
• func BootstrapServersCommaSeparated(bootstrapServers []string) string
• func BrokerClassFilter() func(interface{}) bool
• func CreateTopicIfDoesntExist(admin sarama.ClusterAdmin, logger *zap.Logger, topic string, ...) (string, error)
• func DeleteTopic(admin sarama.ClusterAdmin, topic string) (string, error)
• func GetClusterAdmin(adminFunc NewClusterAdminFunc, bootstrapServers []string, ...) (sarama.ClusterAdmin, error)
• func GetClusterAdminFromConfig(adminFunc NewClusterAdminFunc, config *sarama.Config, ...) (sarama.ClusterAdmin, error)
• func Topic(prefix string, obj metav1.Object) string
• type NewClusterAdminFunc
  • func (f NewClusterAdminFunc) CreateTopicIfDoesntExist(logger *zap.Logger, topic string, config *TopicConfig, ...) (string, error)
  • func (f NewClusterAdminFunc) DeleteTopic(topic string, bootstrapServers []string, secOptions security.ConfigOption) (string, error)
  • func (f NewClusterAdminFunc) IsTopicPresentAndValid(topic string, bootstrapServers []string, secOptions security.ConfigOption) (bool, error)
• type TopicConfig
  • func (c TopicConfig) GetBootstrapServers() string

Constants

const (
	// Kafka broker class annotation value.
	BrokerClass = "Kafka"
)

Functions

func AdminConfig

func AdminConfig() *sarama.Config

AdminConfig returns Kafka Admin configurations.

func BootstrapServersArray

func BootstrapServersArray(bootstrapServers string) []string

func BootstrapServersCommaSeparated

func BootstrapServersCommaSeparated(bootstrapServers []string) string

func BrokerClassFilter

func BrokerClassFilter() func(interface{}) bool

func CreateTopicIfDoesntExist

func CreateTopicIfDoesntExist(admin sarama.ClusterAdmin, logger *zap.Logger, topic string, config *TopicConfig) (string, error)

CreateTopicIfDoesntExist creates a topic with name 'topic' following the TopicConfig configuration passed as parameter.

It returns the topic name or an error.

If the topic already exists, it will return no errors. TODO: what happens if the topic exists but it has a different config?

func DeleteTopic

func DeleteTopic(admin sarama.ClusterAdmin, topic string) (string, error)

func GetClusterAdmin

func GetClusterAdmin(adminFunc NewClusterAdminFunc, bootstrapServers []string, secOptions security.ConfigOption) (sarama.ClusterAdmin, error)

GetClusterAdmin creates a new sarama.ClusterAdmin.

The caller is responsible for closing the sarama.ClusterAdmin.

func GetClusterAdminFromConfig

func GetClusterAdminFromConfig(adminFunc NewClusterAdminFunc, config *sarama.Config, bootstrapServers []string, secOptions security.ConfigOption) (sarama.ClusterAdmin, error)

GetClusterAdminFromConfig creates a new sarama.ClusterAdmin.

The caller is responsible for closing the sarama.ClusterAdmin.

func Topic

func Topic(prefix string, obj metav1.Object) string

Topic returns a topic name given a topic prefix and a generic object.

Types

type NewClusterAdminFunc

type NewClusterAdminFunc func(addrs []string, config *sarama.Config) (sarama.ClusterAdmin, error)

NewClusterAdminFunc creates new sarama.ClusterAdmin.

func (NewClusterAdminFunc) CreateTopicIfDoesntExist

func (f NewClusterAdminFunc) CreateTopicIfDoesntExist(logger *zap.Logger, topic string, config *TopicConfig, secOptions security.ConfigOption) (string, error)

func (NewClusterAdminFunc) DeleteTopic

func (f NewClusterAdminFunc) DeleteTopic(topic string, bootstrapServers []string, secOptions security.ConfigOption) (string, error)

func (NewClusterAdminFunc) IsTopicPresentAndValid

func (f NewClusterAdminFunc) IsTopicPresentAndValid(topic string, bootstrapServers []string, secOptions security.ConfigOption) (bool, error)

type TopicConfig

type TopicConfig struct {
	TopicDetail      sarama.TopicDetail
	BootstrapServers []string
}

TopicConfig contains configurations for creating a topic.

func (TopicConfig) GetBootstrapServers

func (c TopicConfig) GetBootstrapServers() string

GetBootstrapServers returns TopicConfig.BootstrapServers as a comma separated list of bootstrap servers.