kafka

Documentation

Index

• Constants
• func AreTopicsPresentAndValid(kafkaClusterAdmin sarama.ClusterAdmin, topics ...string) (bool, error)
• 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 DisableOffsetAutoCommitConfigOption(config *sarama.Config) error
• func GetSaramaConfig(configOptions ...ConfigOption) (*sarama.Config, error)
• func NoOpConfigOption(*sarama.Config) error
• func Options(config *sarama.Config, options ...ConfigOption) error
• func Topic(prefix string, obj metav1.Object) string
• type ConfigOption
• type InvalidOrNotPresentTopic
  • func (it InvalidOrNotPresentTopic) Error() string
• type NewClientFunc
• type NewClusterAdminFunc
• type TopicConfig
  • func (c TopicConfig) GetBootstrapServers() string

Constants

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

Functions

func AreTopicsPresentAndValid

func AreTopicsPresentAndValid(kafkaClusterAdmin sarama.ClusterAdmin, topics ...string) (bool, error)

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 DisableOffsetAutoCommitConfigOption

func DisableOffsetAutoCommitConfigOption(config *sarama.Config) error

func GetSaramaConfig

func GetSaramaConfig(configOptions ...ConfigOption) (*sarama.Config, error)

GetSaramaConfig returns Kafka Client configuration with the given options applied.

func NoOpConfigOption

func NoOpConfigOption(*sarama.Config) error

NoOpConfigOption is a no-op ConfigOption.

func Options

func Options(config *sarama.Config, options ...ConfigOption) error

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 ConfigOption

type ConfigOption func(config *sarama.Config) error

type InvalidOrNotPresentTopic

type InvalidOrNotPresentTopic struct {
	Topic string
}

func (InvalidOrNotPresentTopic) Error

func (it InvalidOrNotPresentTopic) Error() string

type NewClientFunc

type NewClientFunc func(addrs []string, config *sarama.Config) (sarama.Client, error)

NewClientFunc creates new sarama.Client.

type NewClusterAdminFunc

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

NewClusterAdminFunc creates new sarama.ClusterAdmin.

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.