kafka

Dec 14, 2021

Changes in this version

+ const BootstrapServersConfigMapKey

+ const DefaultTopicNumPartitionConfigMapKey

+ const DefaultTopicReplicationFactorConfigMapKey

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

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

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

+ type InitOffsetsFunc func(ctx context.Context, kafkaClient sarama.Client, ...) (int32, error)

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

type TopicConfig

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

Nov 2, 2021

Changes in this version

+ func AreTopicsPresentAndValid(kafkaClusterAdmin sarama.ClusterAdmin, topics ...string) (bool, 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 func(config *sarama.Config) error

+ type InvalidOrNotPresentTopic struct

+ Topic string

+ func (it InvalidOrNotPresentTopic) Error() string

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

Sep 25, 2021

Changes in this version

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

type NewClusterAdminFunc

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

Sep 30, 2020

Changes in this version

+ func AdminConfig() *sarama.Config

+ func BootstrapServersArray(bootstrapServers string) []string

+ func BootstrapServersCommaSeparated(bootstrapServers []string) string

+ 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(addrs []string, config *sarama.Config) (sarama.ClusterAdmin, error)

+ func (f NewClusterAdminFunc) CreateTopic(logger *zap.Logger, topic string, config *TopicConfig) (string, error)

+ func (f NewClusterAdminFunc) DeleteTopic(topic string, bootstrapServers []string) (string, error)

+ func (f NewClusterAdminFunc) IsTopicPresentAndValid(topic string, bootstrapServers []string) (bool, error)

+ type TopicConfig struct

+ BootstrapServers []string

+ TopicDetail sarama.TopicDetail

+ func (c TopicConfig) GetBootstrapServers() string

Aug 19, 2020

Changes in this version

+ const BrokerClass

+ func BrokerClassFilter() func(interface{}) bool