kafka

Documentation

Index

• Constants
• Variables
• func Dynamic(opts *cli.Options) error
• func Lag(opts *cli.Options) error
• func Read(opts *cli.Options, md *desc.MessageDescriptor) error
• func Relay(opts *cli.Options, relayCh chan interface{}, shutdownCtx context.Context) (relay.IRelayBackend, error)
• func Write(opts *cli.Options, md *desc.MessageDescriptor) error
• type Kafka
  • func (k *Kafka) Read() error
  • func (k *Kafka) Write(key, value []byte) error
• type KafkaLag
  • func NewKafkaLagConnection(opts *cli.Options) (*KafkaLag, error)
  • func (kLag *KafkaLag) GetLastOffsetPerPartition(topic string, groupId string, part int, opts *cli.Options) (int64, error)
  • func (kLag *KafkaLag) LagCalculationForConsumerGroup(groupId string, opts *cli.Options) error
  • func (kLag *KafkaLag) LagCalculationPerPartition(topic string, groupId string, part int, opts *cli.Options) (int64, error)
• type KafkaReader
  • func NewReader(opts *cli.Options) (*KafkaReader, error)
• type KafkaWriter
  • func NewWriter(opts *cli.Options) (*KafkaWriter, error)
• type Relayer
  • func (r *Relayer) Relay() error

Constants

const (
	DefaultBatchSize = 1
)

const (
	RetryReadInterval = 5 * time.Second
)

Variables

var (
	ErrMissingTopic = errors.New("You must specify at least one topic")
)

Functions

func Dynamic

func Dynamic(opts *cli.Options) error

Dynamic starts up a new GRPC client connected to the dProxy service and receives a stream of outbound replay messages which are then written to the message bus.

func Lag

func Lag(opts *cli.Options) error

validate cli options and init connection

func Read

func Read(opts *cli.Options, md *desc.MessageDescriptor) error

Read is the entry point function for performing read operations in Kafka.

This is where we verify that the provided arguments and flag combination makes sense/are valid; this is also where we will perform our initial conn.

func Relay

func Relay(opts *cli.Options, relayCh chan interface{}, shutdownCtx context.Context) (relay.IRelayBackend, error)

Relay sets up a new Kafka relayer

func Write

func Write(opts *cli.Options, md *desc.MessageDescriptor) error

Write is the entry point function for performing write operations in Kafka.

This is where we verify that the passed args and flags combo makes sense, attempt to establish a connection, parse protobuf before finally attempting to perform the write.

Types

type Kafka

type Kafka struct {
	Id      string
	Reader  *skafka.Reader
	Writer  *skafka.Writer
	Options *cli.Options
	MsgDesc *desc.MessageDescriptor
	// contains filtered or unexported fields
}

Kafka holds all attributes required for performing a write to Kafka. This struct should be instantiated via the kafka.Read(..) func.

func (*Kafka) Read

func (k *Kafka) Read() error

Read will attempt to consume one or more messages from a given topic, optionally decode it and/or convert the returned output.

This method SHOULD be able to recover from network hiccups.

func (*Kafka) Write

func (k *Kafka) Write(key, value []byte) error

Write writes a message to a kafka topic. It is a wrapper for WriteMessages.

type KafkaLag

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

func NewKafkaLagConnection

func NewKafkaLagConnection(opts *cli.Options) (*KafkaLag, error)

func (*KafkaLag) GetLastOffsetPerPartition

func (kLag *KafkaLag) GetLastOffsetPerPartition(topic string, groupId string, part int, opts *cli.Options) (int64, error)

func (*KafkaLag) LagCalculationForConsumerGroup

func (kLag *KafkaLag) LagCalculationForConsumerGroup(groupId string, opts *cli.Options) error

calculate lag with a given connection

func (*KafkaLag) LagCalculationPerPartition

func (kLag *KafkaLag) LagCalculationPerPartition(topic string, groupId string, part int, opts *cli.Options) (int64, error)

create new connection per topic and address

type KafkaReader

type KafkaReader struct {
	Reader *skafka.Reader
	Conn   *skafka.Conn
}

func NewReader

func NewReader(opts *cli.Options) (*KafkaReader, error)

type KafkaWriter

type KafkaWriter struct {
	Writer *skafka.Writer
	Conn   *skafka.Conn
}

func NewWriter

func NewWriter(opts *cli.Options) (*KafkaWriter, error)

type Relayer

type Relayer struct {
	Options *cli.Options
	RelayCh chan interface{}

	Looper      *director.FreeLooper
	ShutdownCtx context.Context
	// contains filtered or unexported fields
}

func (*Relayer) Relay

func (r *Relayer) Relay() error

Relay reads messages from Kafka and sends them to RelayCh which is then read by relay.Run()