proxy

Documentation

Index

• Variables
• type Ack
  • func AutoAck() Ack
  • func NewAck(partition int32, offset int64) (Ack, error)
  • func NoAck() Ack
• type Set
  • func NewSet(proxies map[string]*T, defaultPxy *T) *Set
  • func (s *Set) Get(cluster string) (*T, error)
• type T
  • func Spawn(parentActDesc *actor.Descriptor, name string, cfg *config.Proxy) (*T, error)
  • func (p *T) Ack(group, topic string, ack Ack) error
  • func (p *T) AsyncProduce(topic string, key, message sarama.Encoder, headers []sarama.RecordHeader)
  • func (p *T) Consume(group, topic string, ack Ack) (consumer.Message, error)
  • func (p *T) GetAllTopicConsumers(topic string) (map[string]map[string][]int32, error)
  • func (p *T) GetGroupOffsets(group, topic string) ([]admin.PartitionOffset, error)
  • func (p *T) GetTopicConsumers(group, topic string) (map[string][]int32, error)
  • func (p *T) GetTopicMetadata(topic string, withPartitions, withConfig bool) (admin.TopicMetadata, error)
  • func (p *T) ListTopics(withPartitions, withConfig bool) ([]admin.TopicMetadata, error)
  • func (p *T) Produce(topic string, key, message sarama.Encoder, headers []sarama.RecordHeader) (*sarama.ProducerMessage, error)
  • func (p *T) SetGroupOffsets(group, topic string, offsets []admin.PartitionOffset) error
  • func (p *T) Stop()

Variables

var (
	ErrUnavailable        = errors.New("service is shutting down")
	ErrDisabled           = errors.New("service is disabled by configuration")
	ErrHeadersUnsupported = errors.New("headers are not supported with this version of Kafka. Consider changing `kafka.version` (https://github.com/mailgun/kafka-pixy/blob/master/default.yaml#L35)")
)

Types

type Ack

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

func AutoAck

func AutoAck() Ack

AutoAck returns an ack value that should be passed to proxy.Consume function when a caller wants the consumed message to be acknowledged immediately.

func NewAck

func NewAck(partition int32, offset int64) (Ack, error)

NewAck creates an acknowledgement instance from a partition and an offset. Note that group and topic are not included. Respective values that are passed to proxy.Consume function along with the ack are gonna be used.

func NoAck

func NoAck() Ack

NoAck returns an ack value that should be passed to proxy.Consume function when a caller does not want to acknowledge anything.

type Set

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

Set represents a collection of proxy.T instances with a default value.

func NewSet

func NewSet(proxies map[string]*T, defaultPxy *T) *Set

NewSet creates a proxy.Set from a cluster-to-proxy map and a default proxy.

func (*Set) Get

func (s *Set) Get(cluster string) (*T, error)

Get returns a proxy for a cluster name. If there is no proxy configured for the cluster name, then the default proxy is returned.

type T

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

T implements a proxy to a particular Kafka/ZooKeeper cluster.

func Spawn

func Spawn(parentActDesc *actor.Descriptor, name string, cfg *config.Proxy) (*T, error)

Spawn creates a proxy instance and starts its internal goroutines.

func (*T) Ack

func (p *T) Ack(group, topic string, ack Ack) error

func (*T) AsyncProduce

func (p *T) AsyncProduce(topic string, key, message sarama.Encoder, headers []sarama.RecordHeader)

AsyncProduce is an asynchronously counterpart of the `Produce` function. Errors are silently ignored.

func (*T) Consume

func (p *T) Consume(group, topic string, ack Ack) (consumer.Message, error)

Consume consumes a message from the specified topic on behalf of the specified consumer group. If there are no more new messages in the topic at the time of the request then it will block for `Config.Consumer.LongPollingTimeout`. If no new message is produced during that time, then `ErrRequestTimeout` is returned.

Note that during state transitions topic subscribe<->unsubscribe and consumer group register<->deregister the method may return either `ErrBufferOverflow` or `ErrRequestTimeout` even when there are messages available for consumption. In that case the user should back off a bit and then repeat the request.

func (*T) GetAllTopicConsumers

func (p *T) GetAllTopicConsumers(topic string) (map[string]map[string][]int32, error)

GetAllTopicConsumers returns group -> client-id -> consumed-partitions-list mapping for a particular topic. Warning, the function performs scan of all consumer groups registered in ZooKeeper and therefore can take a lot of time.

func (*T) GetGroupOffsets

func (p *T) GetGroupOffsets(group, topic string) ([]admin.PartitionOffset, error)

GetGroupOffsets for every partition of the specified topic it returns the current offset range along with the latest offset and metadata committed by the specified consumer group.

func (*T) GetTopicConsumers

func (p *T) GetTopicConsumers(group, topic string) (map[string][]int32, error)

GetTopicConsumers returns client-id -> consumed-partitions-list mapping for a clients from a particular consumer group and a particular topic.

func (*T) GetTopicMetadata

func (p *T) GetTopicMetadata(topic string, withPartitions, withConfig bool) (admin.TopicMetadata, error)

GetTopicMetadata returns a topic metadata. An optional partition metadata can be requested and/or detailed topic configuration can be requested.

func (*T) ListTopics

func (p *T) ListTopics(withPartitions, withConfig bool) ([]admin.TopicMetadata, error)

ListTopics returns a list of all topics existing in the Kafka cluster.

func (*T) Produce

func (p *T) Produce(topic string, key, message sarama.Encoder, headers []sarama.RecordHeader) (*sarama.ProducerMessage, error)

Produce submits a message to the specified `topic` of the Kafka cluster using `key` to identify a destination partition. The exact algorithm used to map keys to partitions is implementation specific but it is guaranteed that it returns consistent results. If `key` is `nil`, then the message is placed into a random partition.

Errors usually indicate a catastrophic failure of the Kafka cluster, or missing topic if there cluster is not configured to auto create topics.

func (*T) SetGroupOffsets

func (p *T) SetGroupOffsets(group, topic string, offsets []admin.PartitionOffset) error

SetGroupOffsets commits specific offset values along with metadata for a list of partitions of a particular topic on behalf of the specified group.

func (*T) Stop

func (p *T) Stop()

Stop terminates the proxy instances synchronously.