internal

Documentation

Index

• Constants
• func FromHeaders(sh []segmentio.Header) []kafka.Header
• func Mechanism(sasl kafka.SASLConfig) sasl.Mechanism
• func NewCronsumer(cfg *kafka.Config, fn kafka.ConsumeFn) kafka.Cronsumer
• func NewKafkaClient(cfg *kafka.Config) (kafkaClient, error)
• func NewTLSConfig(sasl kafka.SASLConfig) *tls.Config
• func ToHeaders(h []kafka.Header) []segmentio.Header
• func VerifyTopics(client kafkaClient, topics ...string) (bool, error)
• type Collector
  • func NewCollector(metricPrefix string, cronsumerMetric *CronsumerMetric) *Collector
  • func (s *Collector) Collect(ch chan<- prometheus.Metric)
  • func (s *Collector) Describe(ch chan<- *prometheus.Desc)
• type Consumer
• type CronsumerMetric
• type MessageWrapper
  • func NewMessageWrapper(msg segmentio.Message, strategyName string) *MessageWrapper
  • func (m *MessageWrapper) GetHeaders() map[string][]byte
  • func (m *MessageWrapper) IncreaseRetryAttemptCount()
  • func (m *MessageWrapper) IncreaseRetryCount()
  • func (m *MessageWrapper) IsGteMaxRetryCount(maxRetry int) bool
  • func (m *MessageWrapper) NewProduceTime()
  • func (m *MessageWrapper) ResetRetryAttempt()
  • func (m *MessageWrapper) RouteMessageToTopic(topic string)
  • func (m *MessageWrapper) To(increaseRetry bool, increaseRetryAttempt bool) segmentio.Message
• type Producer

Constants

const (
	RetryHeaderKey              = "x-retry-count"
	RetryAttemptHeaderKey       = "x-retry-attempt-count"
	MessageProduceTimeHeaderKey = "x-produce-time"
	MessageErrHeaderKey         = "x-error-message"
)

const Name = "kafka_cronsumer"

Functions

func FromHeaders

func FromHeaders(sh []segmentio.Header) []kafka.Header

func Mechanism

func Mechanism(sasl kafka.SASLConfig) sasl.Mechanism

TODO: we can support `plain` authentication type link: https://github.com/segmentio/kafka-go#plain

func NewCronsumer

func NewCronsumer(cfg *kafka.Config, fn kafka.ConsumeFn) kafka.Cronsumer

func NewKafkaClient

func NewKafkaClient(cfg *kafka.Config) (kafkaClient, error)

func NewTLSConfig

func NewTLSConfig(sasl kafka.SASLConfig) *tls.Config

func ToHeaders

func ToHeaders(h []kafka.Header) []segmentio.Header

func VerifyTopics

func VerifyTopics(client kafkaClient, topics ...string) (bool, error)

Types

type Collector

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

func NewCollector

func NewCollector(metricPrefix string, cronsumerMetric *CronsumerMetric) *Collector

func (*Collector) Collect

func (s *Collector) Collect(ch chan<- prometheus.Metric)

func (*Collector) Describe

func (s *Collector) Describe(ch chan<- *prometheus.Desc)

type Consumer

type Consumer interface {
	ReadMessage(ctx context.Context) (*segmentio.Message, error)
	Stop()
}

type CronsumerMetric

type CronsumerMetric struct {
	TotalRetriedMessagesCounter   int64
	TotalDiscardedMessagesCounter int64
}

type MessageWrapper

type MessageWrapper struct {
	kafka.Message
	RetryCount        int
	ProduceTime       int64 // Nano time
	RetryAttemptCount int
}

func NewMessageWrapper

func NewMessageWrapper(msg segmentio.Message, strategyName string) *MessageWrapper

func (*MessageWrapper) GetHeaders

func (m *MessageWrapper) GetHeaders() map[string][]byte

func (*MessageWrapper) IncreaseRetryAttemptCount

func (m *MessageWrapper) IncreaseRetryAttemptCount()

func (*MessageWrapper) IncreaseRetryCount

func (m *MessageWrapper) IncreaseRetryCount()

func (*MessageWrapper) IsGteMaxRetryCount

func (m *MessageWrapper) IsGteMaxRetryCount(maxRetry int) bool

func (*MessageWrapper) NewProduceTime

func (m *MessageWrapper) NewProduceTime()

func (*MessageWrapper) ResetRetryAttempt

func (m *MessageWrapper) ResetRetryAttempt()

func (*MessageWrapper) RouteMessageToTopic

func (m *MessageWrapper) RouteMessageToTopic(topic string)

func (*MessageWrapper) To

func (m *MessageWrapper) To(increaseRetry bool, increaseRetryAttempt bool) segmentio.Message

type Producer

type Producer interface {
	ProduceWithRetryOption(message MessageWrapper, increaseRetry bool, increaseRetryAttempt bool) error
	Produce(message kafka.Message) error
	ProduceBatch(messages []kafka.Message) error
	Close()
}