keystone

README

keystone

A core Kafka library for Go, providing reliable messaging primitives and stream-processing foundations.

Example

• Consumer Group

// Avro schema registry configurations.
// Need to register new schemas here.
sr := avro.NewCachedSchemaRegistry([]string{"foobar"}, 1)

// Kafka configurations
cfg := keystone.Config{
	Brokers:          []string{"localhost:9092"},
	Group:            "sampleConsumer",
	Version:          keystone.Version_2_1_1,
	Topics:           []string{"foobar"},
	BalanceStrategy:  keystone.RoundRobin,
	Offset:           keystone.Oldest,
	ConsumerCallback: Callback,
	Decoder:          keystone.GetDefaultDecoder(),
	EncoderBuilder:   keystone.DefaultEncoderBuilder(),
}

cg := keystone.NewConsumerGroup(cfg)

// Function is called when message recieved.
func Callback(key interface{}, value interface{}) (err error) {
	return nil
}

Documentation

Index

• Constants
• func NewSaramaLogger(logger Logger) *customSaramaLogger
• type BalanceStrategy
• type Config
• type ConsumerCallback
• type ConsumerErrorHandler
• type ConsumerGroup
  • func NewConsumerGroup(config Config) ConsumerGroup
• type ConsumerGroupHandler
  • func (c *ConsumerGroupHandler) Cleanup(sarama.ConsumerGroupSession) error
  • func (c *ConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (c *ConsumerGroupHandler) Setup(sarama.ConsumerGroupSession) error
• type Decoder
  • func GetAvroDecoder(ss avro.SchemaStore) Decoder
  • func GetDefaultDecoder() Decoder
• type Encoder
• type EncoderBuilder
  • func DefaultEncoderBuilder() EncoderBuilder
  • func NewAvroEncoderBuilder(schemaStore avro.SchemaStore) EncoderBuilder
• type Logger
• type Offset
• type Producer
  • func NewProducer(cfg Config) (Producer, error)
• type Version

Constants

const (
	Version_2_1_1 Version = "2.1.1"

	Sticky     BalanceStrategy = "sticky"
	RoundRobin BalanceStrategy = "roundrobin"
	Range      BalanceStrategy = "range"

	Newest Offset = "newest"
	Oldest Offset = "oldest"
)

Functions

func NewSaramaLogger

func NewSaramaLogger(logger Logger) *customSaramaLogger

Types

type BalanceStrategy

type BalanceStrategy string

BalanceStrategy consumer group balancing strategy.

type Config

type Config struct {
	Brokers              []string
	Group                string
	Version              Version
	Topics               []string
	BalanceStrategy      BalanceStrategy
	Offset               Offset
	DebugLogger          Logger
	EncoderBuilder       EncoderBuilder
	Decoder              Decoder
	ConsumerCallback     ConsumerCallback
	ConsumerErrorHandler ConsumerErrorHandler
}

Config General configurations.

type ConsumerCallback

type ConsumerCallback func(key, value interface{}) (err error)

ConsumerCallback hanler function for the consumer.

type ConsumerErrorHandler

type ConsumerErrorHandler func(err error) (commitMsg bool)

ConsumerErrorHandler hanlde the error returning from the ConsumerCallback.

type ConsumerGroup

type ConsumerGroup interface {
	// Init initialize the consumer group.
	Init() (err error)
	// Run start the consumer group.
	Run() (err error)
}

ConsumerGroup the interface for manage consumer group

func NewConsumerGroup

func NewConsumerGroup(config Config) ConsumerGroup

type ConsumerGroupHandler

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

ConsumerGroupHandler implementation for ConsumerGroupHandler.

func (*ConsumerGroupHandler) Cleanup

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

Cleanup cleanup the consumer group session.

func (*ConsumerGroupHandler) ConsumeClaim

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

ConsumeClaim decode messages and call the consumer callback function configured.

func (*ConsumerGroupHandler) Setup

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

Setup setup the consumer group session.

type Decoder

type Decoder func(consumerMessage *sarama.ConsumerMessage) (key, value interface{}, err error)

Decoder decode the consumer message to according to the given implementation

func GetAvroDecoder

func GetAvroDecoder(ss avro.SchemaStore) Decoder

GetAvroDecoder creates avro decoder with avro.SchemaStore.

func GetDefaultDecoder

func GetDefaultDecoder() Decoder

GetDefaultDecoder creates a default decoder (string decoder).

type Encoder

type Encoder func(schemaStore avro.SchemaStore, schema string, data []byte) sarama.Encoder

type EncoderBuilder

type EncoderBuilder interface {
	// Build build the sarama.Encoders with the given subject and data.
	Build(subject string, data interface{}) sarama.Encoder
}

EncoderBuilder resposible for creating encoders.

func DefaultEncoderBuilder

func DefaultEncoderBuilder() EncoderBuilder

DefaultEncoderBuilder creates instance of EncoderBuilder.

func NewAvroEncoderBuilder

func NewAvroEncoderBuilder(schemaStore avro.SchemaStore) EncoderBuilder

NewAvroEncoderBuilder create instance of EncoderBuilder.

type Logger

type Logger interface {
	Print(v ...interface{})
	Printf(format string, v ...interface{})
	Println(v ...interface{})
}

var (
	DefaultLogger Logger = log.New(io.Discard, "[keystone]", log.LstdFlags)
)

type Offset

type Offset string

Offset message offset of the consumer group.

type Producer

type Producer interface {
	// Produce produce the kafka message to the given topic.
	Produce(topic string, schema string, key interface{}, value interface{}) (partition int32, offset int64, err error)
}

Producer kafka message producer interface.

func NewProducer

func NewProducer(cfg Config) (Producer, error)

NewProducer creates a producer instance.

type Version

type Version string

Version kafka version.