kafka

Documentation

Index

• Constants
• func AdminConfig() *sarama.Config
• func BootstrapServersArray(bootstrapServers string) []string
• func BootstrapServersCommaSeparated(bootstrapServers []string) string
• func BrokerClassFilter() func(interface{}) bool
• func CreateTopic(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) CreateTopic(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 CreateTopic

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

CreateTopic 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.

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) CreateTopic

func (f NewClusterAdminFunc) CreateTopic(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.