redpanda

Documentation

Index

• type Consumer
  • func NewConsumer(kafkaBrokers, subscribeRegexes []string, groupId, instanceId string, ...) (*Consumer, error)
  • func (c *Consumer) GetAllLastMessages() (map[string]*shared.KafkaMessage, error)
  • func (c *Consumer) GetAllTopics() ([]string, error)
  • func (c *Consumer) GetLastMessageFromTopic(topic string) (*shared.KafkaMessage, error)
  • func (c *Consumer) GetMessage() *shared.KafkaMessage
  • func (c *Consumer) GetMessages() <-chan *shared.KafkaMessage
  • func (c *Consumer) GetStats() (uint64, uint64)
  • func (c *Consumer) GetTopics() []string
  • func (c *Consumer) IsReady() bool
  • func (c *Consumer) MarkMessage(message *shared.KafkaMessage)
  • func (c *Consumer) MarkMessages(messages []*shared.KafkaMessage)
  • func (c *Consumer) UnsafeGetClient() *sarama.Client
• type ConsumerGroupHandler
  • func (c *ConsumerGroupHandler) Cleanup(session sarama.ConsumerGroupSession) error
  • func (c *ConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (c *ConsumerGroupHandler) Setup(session sarama.ConsumerGroupSession) error

Types

type Consumer

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

Consumer represents a Kafka consumer.

func NewConsumer

func NewConsumer(kafkaBrokers, subscribeRegexes []string, groupId, instanceId string, initialOffset int64) (*Consumer, error)

NewConsumer initializes and returns a new Consumer instance.

func (*Consumer) GetAllLastMessages

func (c *Consumer) GetAllLastMessages() (map[string]*shared.KafkaMessage, error)

func (*Consumer) GetAllTopics

func (c *Consumer) GetAllTopics() ([]string, error)

GetAllTopics returns all topics available in the Kafka cluster.

func (*Consumer) GetLastMessageFromTopic

func (c *Consumer) GetLastMessageFromTopic(topic string) (*shared.KafkaMessage, error)

GetLastMessageFromTopic returns the latest message from a topic.

func (*Consumer) GetMessage

func (c *Consumer) GetMessage() *shared.KafkaMessage

GetMessage returns the next message from the consumer.

func (*Consumer) GetMessages

func (c *Consumer) GetMessages() <-chan *shared.KafkaMessage

GetMessages returns the channel of messages from the consumer.

func (*Consumer) GetStats

func (c *Consumer) GetStats() (uint64, uint64)

GetStats returns consumed message counts.

func (*Consumer) GetTopics

func (c *Consumer) GetTopics() []string

GetTopics returns the topics that the consumer is subscribed to.

func (*Consumer) IsReady

func (c *Consumer) IsReady() bool

IsReady returns whether the consumer is ready to consume messages.

func (*Consumer) MarkMessage

func (c *Consumer) MarkMessage(message *shared.KafkaMessage)

MarkMessage marks a message as processed.

func (*Consumer) MarkMessages

func (c *Consumer) MarkMessages(messages []*shared.KafkaMessage)

MarkMessages marks a slice of messages as processed.

func (*Consumer) UnsafeGetClient

func (c *Consumer) UnsafeGetClient() *sarama.Client

UnsafeGetClient should only be used as a last resort.

type ConsumerGroupHandler

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

ConsumerGroupHandler represents a Sarama consumer group consumer

func (*ConsumerGroupHandler) Cleanup

func (c *ConsumerGroupHandler) Cleanup(session sarama.ConsumerGroupSession) error

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

func (*ConsumerGroupHandler) ConsumeClaim

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

ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages(). Once the Messages() channel is closed, the Handler must finish its processing loop and exit.

func (*ConsumerGroupHandler) Setup

func (c *ConsumerGroupHandler) Setup(session sarama.ConsumerGroupSession) error

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