sarama

Documentation

Index

• Constants
• func AuthFromSarama(config *sarama.Config) *client.KafkaAuthConfig
• func EnableSaramaLogging(enable bool)
• func GetOffsets(client sarama.Client, topicPartitions map[string][]int32, time int64) (map[string]map[int32]int64, error)
• func LoadAuthConfig(ctx context.Context, name string, namespace string) *client.KafkaAuthConfig
• func LoadEventingKafkaSettings(configMap map[string]string) (*commonconfig.EventingKafkaConfig, error)
• func LoadSettings(ctx context.Context, clientId string, configMap map[string]string, ...) (*commonconfig.EventingKafkaConfig, error)
• func StringifyHeaderPtrs(headers []*sarama.RecordHeader) map[string][]string
• func StringifyHeaders(headers []sarama.RecordHeader) map[string][]string
• type GetAuth

Constants

const DefaultAuthSecretName = "kafka-cluster"

Functions

func AuthFromSarama

func AuthFromSarama(config *sarama.Config) *client.KafkaAuthConfig

AuthFromSarama creates a KafkaAuthConfig using the SASL settings from a given Sarama config, or nil if there is no SASL user in that config

func EnableSaramaLogging

func EnableSaramaLogging(enable bool)

EnableSaramaLogging Is A Utility Function For Enabling Sarama Logging (Debugging)

func GetOffsets

func GetOffsets(client sarama.Client, topicPartitions map[string][]int32, time int64) (map[string]map[int32]int64, error)

GetOffsets queries the cluster to get the most recent available offset at the given time (in milliseconds) for the given topics and partition Time should be OffsetOldest for the earliest available offset, OffsetNewest for the offset of the message that will be produced next, or a time.

See sarama.Client.GetOffset for getting the offset of a single topic/partition combination

func LoadAuthConfig

func LoadAuthConfig(ctx context.Context, name string, namespace string) *client.KafkaAuthConfig

LoadAuthConfig Creates A Sarama-Safe KafkaAuthConfig From The Specified Secret Name/Namespace

func LoadEventingKafkaSettings

func LoadEventingKafkaSettings(configMap map[string]string) (*commonconfig.EventingKafkaConfig, error)

func LoadSettings

func LoadSettings(ctx context.Context, clientId string, configMap map[string]string, getAuthConfig GetAuth) (*commonconfig.EventingKafkaConfig, error)

LoadSettings Loads The Sarama & EventingKafka Configuration From The ConfigMap The Provided Context Must Have A Kubernetes Client Associated With It

func StringifyHeaderPtrs

func StringifyHeaderPtrs(headers []*sarama.RecordHeader) map[string][]string

StringifyHeaderPtrs Is A Pointer-version of the StringifyHeaders function

func StringifyHeaders

func StringifyHeaders(headers []sarama.RecordHeader) map[string][]string

StringifyHeaders Is A Utility function to convert []byte headers to string ones for logging purposes

Types

type GetAuth

type GetAuth func(ctx context.Context, authSecretName string, authSecretNamespace string) *client.KafkaAuthConfig

GetAuth Is The Function Type Used To Delay Loading Auth Config Until The Secret Name/Namespace Are Known