kafka

Documentation

Overview

Package kafka contains data types, interfaces, and methods related to Kafka that can be used to configure brokers, as well as consume/produce messages.

Index

• func GetBrokersAddresses(brokersCfg BrokersConfig) []string
• func SaramaConfigFromBrokerConfig(cfg *BrokerConfiguration) (*sarama.Config, error)
• type BrokerConfiguration
• type BrokersConfig
• type SASLConfiguration
• type SCRAMClient
  • func (x *SCRAMClient) Begin(userName, password, authzID string) (err error)
  • func (x *SCRAMClient) Done() bool
  • func (x *SCRAMClient) Step(challenge string) (response string, err error)

Functions

func GetBrokersAddresses

func GetBrokersAddresses(brokersCfg BrokersConfig) []string

GetBrokersAddresses returns array of addresses of the configured brokers

func SaramaConfigFromBrokerConfig

func SaramaConfigFromBrokerConfig(cfg *BrokerConfiguration) (*sarama.Config, error)

SaramaConfigFromBrokerConfig returns a Config struct from broker.Configuration parameters

Types

type BrokerConfiguration

type BrokerConfiguration struct {
	Address          string        `mapstructure:"address" toml:"address"`
	SecurityProtocol string        `mapstructure:"security_protocol" toml:"security_protocol"`
	CertPath         string        `mapstructure:"cert_path" toml:"cert_path"`
	SaslMechanism    string        `mapstructure:"sasl_mechanism" toml:"sasl_mechanism"`
	SaslUsername     string        `mapstructure:"sasl_username" toml:"sasl_username"`
	SaslPassword     string        `mapstructure:"sasl_password" toml:"sasl_password"`
	Topic            string        `mapstructure:"topic" toml:"topic"`
	Timeout          time.Duration `mapstructure:"timeout" toml:"timeout"`
	Group            string        `mapstructure:"group" toml:"group"`
	ClientID         string        `mapstructure:"client_id" toml:"client_id"`
	Enabled          bool          `mapstructure:"enabled" toml:"enabled"`
}

BrokerConfiguration represents configuration of a single-instance Kafka broker

type BrokersConfig

type BrokersConfig []*BrokerConfiguration

BrokersConfig represents configuration of Kafka broker with multiple instances running on different hosts (kafka cluster)

type SASLConfiguration

type SASLConfiguration struct {
	SecurityProtocol string `mapstructure:"security_protocol" toml:"security_protocol"`
	CertPath         string `mapstructure:"cert_path" toml:"cert_path"`
	SaslMechanism    string `mapstructure:"sasl_mechanism" toml:"sasl_mechanism"`
	SaslUsername     string `mapstructure:"sasl_username" toml:"sasl_username"`
	SaslPassword     string `mapstructure:"sasl_password" toml:"sasl_password"`
}

SASLConfiguration represents configuration of SASL authentication for a given Kafka broker

type SCRAMClient

type SCRAMClient struct {
	*scram.Client
	*scram.ClientConversation
	scram.HashGeneratorFcn
}

SCRAMClient implementation for the SCRAM authentication

func (*SCRAMClient) Begin

func (x *SCRAMClient) Begin(userName, password, authzID string) (err error)

Begin prepares the client for the SCRAM exchange

func (*SCRAMClient) Done

func (x *SCRAMClient) Done() bool

Done should return true when the SCRAM conversation is over.

func (*SCRAMClient) Step

func (x *SCRAMClient) Step(challenge string) (response string, err error)

Step steps client through the SCRAM exchange