kafka

Documentation

Index

• Constants
• func AreTopicsPresentAndValid(kafkaClusterAdmin sarama.ClusterAdmin, topics ...string) (bool, error)
• func BootstrapServersArray(bootstrapServers string) []string
• func BootstrapServersCommaSeparated(bootstrapServers []string) string
• func BootstrapServersFromConfigMap(logger *zap.Logger, cm *corev1.ConfigMap) ([]string, error)
• func BrokerClassFilter() func(interface{}) bool
• func BrokerTopic(prefix string, obj metav1.Object) string
• func ChannelTopic(prefix string, obj metav1.Object) string
• 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
• type ConfigOption
• type InitOffsetsFunc
• type InvalidOrNotPresentTopic
  • func (it InvalidOrNotPresentTopic) Error() string
• type NewClientFunc
• type NewClusterAdminClientFunc
• type TopicConfig
  • func TopicConfigFromConfigMap(logger *zap.Logger, cm *corev1.ConfigMap) (*TopicConfig, error)
  • func (c TopicConfig) GetBootstrapServers() string

Constants

const (
	DefaultTopicNumPartitionConfigMapKey      = "default.topic.partitions"
	DefaultTopicReplicationFactorConfigMapKey = "default.topic.replication.factor"
	BootstrapServersConfigMapKey              = "bootstrap.servers"
)

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 BootstrapServersFromConfigMap

func BootstrapServersFromConfigMap(logger *zap.Logger, cm *corev1.ConfigMap) ([]string, error)

func BrokerClassFilter

func BrokerClassFilter() func(interface{}) bool

func BrokerTopic

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

BrokerTopic returns a topic name given a topic prefix and a Broker.

func ChannelTopic

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

ChannelTopic returns a topic name given a topic prefix and a KafkaChannel.

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

Types

type ConfigOption

type ConfigOption func(config *sarama.Config) error

type InitOffsetsFunc

type InitOffsetsFunc func(ctx context.Context, kafkaClient sarama.Client, kafkaAdminClient sarama.ClusterAdmin, topics []string, consumerGroup string) (int32, error)

InitOffsetsFunc initialize offsets for a provided set of topics and a provided consumer group id.

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 NewClusterAdminClientFunc

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

NewClusterAdminClientFunc creates new sarama.ClusterAdmin.

type TopicConfig

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

TopicConfig contains configurations for creating a topic.

func TopicConfigFromConfigMap

func TopicConfigFromConfigMap(logger *zap.Logger, cm *corev1.ConfigMap) (*TopicConfig, error)

func (TopicConfig) GetBootstrapServers

func (c TopicConfig) GetBootstrapServers() string

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