kafkaconsumer

Aug 5, 2021

Changes in this version

type Metrics

+ ConsumerLagStored *prometheus.GaugeVec

Nov 12, 2019

Changes in this version

+ type KafkaConsumer struct

+ func (k *KafkaConsumer) GetMetrics() *Metrics

+ func (k *KafkaConsumer) Receive(msg fbcontext.Message) error

+ func (k *KafkaConsumer) Setup(config map[string]string, eventchan chan firebolt.Event) error

+ func (k *KafkaConsumer) Shutdown() error

+ func (k *KafkaConsumer) Start() error

+ type Metrics struct

+ ConsumerLag *prometheus.GaugeVec

+ EventsConsumed *prometheus.CounterVec

+ HighWatermark *prometheus.GaugeVec

+ LowWatermark *prometheus.GaugeVec

+ RecoveryEvents *prometheus.CounterVec

+ RecoveryPartitions prometheus.Gauge

+ RecoveryRemaining *prometheus.GaugeVec

+ StoredOffset *prometheus.GaugeVec

+ func (m *Metrics) RegisterConsumerMetrics()

+ func (m *Metrics) UpdateConsumerMetrics(statsJSON string, topic string)

+ type PartitionStats struct

+ type RecoveryConsumer struct

+ func NewRecoveryConsumer(topic string, sendCh chan firebolt.Event, config map[string]string, ...) (*RecoveryConsumer, error)

+ func (rc *RecoveryConsumer) RefreshAssignments() error

+ func (rc *RecoveryConsumer) RequestRecovery(partitionID int32, fromOffset kafka.Offset, toOffset kafka.Offset)

+ func (rc *RecoveryConsumer) SetAssignedPartitions(partitions []kafka.TopicPartition)

+ func (rc *RecoveryConsumer) Shutdown()

+ type RecoveryRequest struct

+ Created time.Time

+ FromOffset int64

+ PartitionID int32

+ ToOffset int64

+ type RecoveryRequests struct

+ Requests []*RecoveryRequest

+ type RecoveryTracker struct

+ func NewRecoveryTracker(metrics *Metrics, ctx fbcontext.FBContext) (*RecoveryTracker, error)

+ func (rt *RecoveryTracker) AddRecoveryRequest(partitionID int32, fromOffset int64, toOffset int64) error

+ func (rt *RecoveryTracker) GetRecoveryRequest(partitionID int32) *RecoveryRequest

+ func (rt *RecoveryTracker) MarkRecoveryComplete(partitionID int32, toOffset int64) error

+ func (rt *RecoveryTracker) RecoveryRequestCount() int

+ func (rt *RecoveryTracker) Shutdown()

+ func (rt *RecoveryTracker) UpdateRecoveryRequest(partitionID int32, fromOffset int64, toOffset int64) error