kafka

Documentation

Index

• Variables
• func CreateKafkaConsumer(triggerObjName, funcName, ns, topic string, clientset kubernetes.Interface) error
• func DeleteKafkaConsumer(triggerObjName, funcName, ns, topic string) error
• func GetSASLConfiguration(username string, password string) (string, string, error)
• func GetTLSConfiguration(caFile string, certFile string, keyFile string, insecure string) (*tls.Config, error)
• type Consumer
  • func NewConsumer(funcName string, funcPort int, ns string, clientset kubernetes.Interface, ...) *Consumer
  • func (c *Consumer) Cleanup(sarama.ConsumerGroupSession) error
  • func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (c *Consumer) Reset()
  • func (c *Consumer) Setup(sarama.ConsumerGroupSession) error

Variables

var (
	// ErrorCrtFileMandatory when private key is provided but certificate not.
	ErrorCrtFileMandatory = errors.New("client certificate is mandatory when private key file is provided")
	// ErrorKeyFileMandatory when certificate is provided but private not.
	ErrorKeyFileMandatory = errors.New("client private key is mandatory when certificate file is provided")
	// ErrorUsernameOrPasswordMandatory when username or password is not provided
	ErrorUsernameOrPasswordMandatory = errors.New("username and password is mandatory")
)

Functions

func CreateKafkaConsumer

func CreateKafkaConsumer(triggerObjName, funcName, ns, topic string, clientset kubernetes.Interface) error

CreateKafkaConsumer creates a goroutine that subscribes to Kafka topic

func DeleteKafkaConsumer

func DeleteKafkaConsumer(triggerObjName, funcName, ns, topic string) error

DeleteKafkaConsumer deletes goroutine created by CreateKafkaConsumer

func GetSASLConfiguration

func GetSASLConfiguration(username string, password string) (string, string, error)

GetSASLConfiguration build SASL configuration for kafka.

func GetTLSConfiguration

func GetTLSConfiguration(caFile string, certFile string, keyFile string, insecure string) (*tls.Config, error)

GetTLSConfiguration build TLS configuration for kafka, return tlsConfig associated with kafka connection.

Types

type Consumer

type Consumer struct {
	// contains filtered or unexported fields
}

Consumer represents a Sarama consumer group consumer.

func NewConsumer

func NewConsumer(funcName string, funcPort int, ns string, clientset kubernetes.Interface, ready chan struct{}, backoff time.Duration) *Consumer

NewConsumer returns new consumer.

func (*Consumer) Cleanup

func (c *Consumer) Cleanup(sarama.ConsumerGroupSession) error

Cleanup is run at the end of a session, once all ConsumeClaim goroutines have exited.

func (*Consumer) ConsumeClaim

func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages().

func (*Consumer) Reset

func (c *Consumer) Reset()

Reset resets the consumer for new session.

func (*Consumer) Setup

func (c *Consumer) Setup(sarama.ConsumerGroupSession) error

Setup is run at the beginning of a new session, before ConsumeClaim.