kafka

package module
v0.0.0-...-2165892 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Mar 13, 2019 License: Apache-2.0 Imports: 15 Imported by: 0

README

codecov CircleCI

go-kafka-avro

A library provides consumer/producer to work with kafka, avro and schema registry

Installation

$ go get github.com/dangkaka/go-kafka-avro
Usage

Examples

  • Setup kafka, schema-registry

    docker-compose up -d

  • Add test messages

    go run producer/main.go -n 10

  • Run consumer

    go run consumer.main.go
References

Documentation

Index

Constants

View Source 
const (
	COMMIT_ERROR_CTX = "commit"
)

Variables

This section is empty.

Functions

This section is empty.

Types

type AvroConsumer 

type AvroConsumer struct {
	Consumer             *cluster.Consumer
	SchemaRegistryClient *CachedSchemaRegistryClient
	// contains filtered or unexported fields
}

func NewAvroConsumer 

func NewAvroConsumer(kafkaServers []string, schemaRegistryServers []string,
	topic string, groupId string, callbacks ConsumerCallbacks) (*AvroConsumer, error)

avroConsumer is a basic consumer to interact with schema registry, avro and kafka

func NewAvroConsumerCustomConfig 

func NewAvroConsumerCustomConfig(kafkaServers []string, schemaRegistryServers []string,
	topic string, groupId string, callbacks ConsumerCallbacks, config *cluster.Config) (*AvroConsumer, error)

func (*AvroConsumer) Close 

func (ac *AvroConsumer) Close()

func (*AvroConsumer) Consume 

func (ac *AvroConsumer) Consume() error

func (*AvroConsumer) GetSchema 

func (ac *AvroConsumer) GetSchema(id int) (*goavro.Codec, error)

GetSchemaId get schema id from schema-registry service

func (*AvroConsumer) ProcessAvroMsg 

func (ac *AvroConsumer) ProcessAvroMsg(m *sarama.ConsumerMessage) (Message, error)

type AvroProducer 

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

func NewAvroProducer 

func NewAvroProducer(kafkaServers []string, schemaRegistryServers []string) (*AvroProducer, error)

NewAvroProducer is a basic producer to interact with schema registry, avro and kafka

func NewAvroProducerCustomConfig 

func NewAvroProducerCustomConfig(kafkaServers []string, schemaRegistryServers []string, config *sarama.Config) (*AvroProducer, error)

func (*AvroProducer) Add 

func (ap *AvroProducer) Add(topic string, schema string, key []byte, value []byte) error

func (*AvroProducer) AddWithResponse 

func (ap *AvroProducer) AddWithResponse(topic string, schema string, key []byte, value []byte) (int32, int64, error)

func (*AvroProducer) Close 

func (ac *AvroProducer) Close()

func (*AvroProducer) GetSchemaId 

func (ap *AvroProducer) GetSchemaId(topic string, avroCodec *goavro.Codec) (int, error)

GetSchemaId get schema id from schema-registry service

type CachedSchemaRegistryClient 

type CachedSchemaRegistryClient struct {
	SchemaRegistryClient *SchemaRegistryClient
	// contains filtered or unexported fields
}

CachedSchemaRegistryClient is a schema registry client that will cache some data to improve performance

func NewCachedSchemaRegistryClient 

func NewCachedSchemaRegistryClient(connect []string) *CachedSchemaRegistryClient

func NewCachedSchemaRegistryClientWithRetries 

func NewCachedSchemaRegistryClientWithRetries(connect []string, retries int) *CachedSchemaRegistryClient

func (*CachedSchemaRegistryClient) CreateSubject 

func (client *CachedSchemaRegistryClient) CreateSubject(subject string, codec *goavro.Codec) (int, error)

CreateSubject will return and cache the id with the given codec

func (*CachedSchemaRegistryClient) DeleteSubject 

func (client *CachedSchemaRegistryClient) DeleteSubject(subject string) error

DeleteSubject deletes the subject, should only be used in development

func (*CachedSchemaRegistryClient) DeleteVersion 

func (client *CachedSchemaRegistryClient) DeleteVersion(subject string, version int) error

DeleteVersion deletes the a specific version of a subject, should only be used in development.

func (*CachedSchemaRegistryClient) GetLatestSchema 

func (client *CachedSchemaRegistryClient) GetLatestSchema(subject string) (*goavro.Codec, error)

GetLatestSchema returns the highest version schema for a subject

func (*CachedSchemaRegistryClient) GetSchema 

func (client *CachedSchemaRegistryClient) GetSchema(id int) (*goavro.Codec, error)

GetSchema will return and cache the codec with the given id

func (*CachedSchemaRegistryClient) GetSchemaByVersion 

func (client *CachedSchemaRegistryClient) GetSchemaByVersion(subject string, version int) (*goavro.Codec, error)

GetSchemaByVersion returns the codec for a specific version of a subject

func (*CachedSchemaRegistryClient) GetSubjects 

func (client *CachedSchemaRegistryClient) GetSubjects() ([]string, error)

GetSubjects returns a list of subjects

func (*CachedSchemaRegistryClient) GetVersions 

func (client *CachedSchemaRegistryClient) GetVersions(subject string) ([]int, error)

GetVersions returns a list of all versions of a subject

func (*CachedSchemaRegistryClient) IsSchemaRegistered 

func (client *CachedSchemaRegistryClient) IsSchemaRegistered(subject string, codec *goavro.Codec) (int, error)

IsSchemaRegistered checks if a specific codec is already registered to a subject

type ConsumerCallbacks 

type ConsumerCallbacks struct {
	OnDataReceived func(msg *Message) bool
	OnError        func(err error)
	OnCommitFailed func(msg *Message, err error)
	OnNotification func(notification *cluster.Notification)
}

type Error 

type Error struct {
	ErrorCode int    `json:"error_code"`
	Message   string `json:"message"`
}

Error holds more detailed information about errors coming back from schema registry

func (*Error) Error 

func (e *Error) Error() string

type Message 

type Message struct {
	SchemaId  int
	Topic     string
	Partition int32
	Offset    int64
	Key       string
	Value     string
}

type SchemaRegistryClient 

type SchemaRegistryClient struct {
	SchemaRegistryConnect []string
	// contains filtered or unexported fields
}

SchemaRegistryClient is a basic http client to interact with schema registry

func NewSchemaRegistryClient 

func NewSchemaRegistryClient(connect []string) *SchemaRegistryClient

NewSchemaRegistryClient creates a client to talk with the schema registry at the connect string By default it will retry failed requests (5XX responses and http errors) len(connect) number of times

func NewSchemaRegistryClientWithRetries 

func NewSchemaRegistryClientWithRetries(connect []string, retries int) *SchemaRegistryClient

NewSchemaRegistryClientWithRetries creates an http client with a configurable amount of retries on 5XX responses

func (*SchemaRegistryClient) CreateSubject 

func (client *SchemaRegistryClient) CreateSubject(subject string, codec *goavro.Codec) (int, error)

CreateSubject adds a schema to the subject

func (*SchemaRegistryClient) DeleteSubject 

func (client *SchemaRegistryClient) DeleteSubject(subject string) error

DeleteSubject deletes a subject. It should only be used in development

func (*SchemaRegistryClient) DeleteVersion 

func (client *SchemaRegistryClient) DeleteVersion(subject string, version int) error

DeleteVersion deletes a subject. It should only be used in development

func (*SchemaRegistryClient) GetLatestSchema 

func (client *SchemaRegistryClient) GetLatestSchema(subject string) (*goavro.Codec, error)

GetLatestSchema returns a goavro.Codec for the latest version of the subject

func (*SchemaRegistryClient) GetSchema 

func (client *SchemaRegistryClient) GetSchema(id int) (*goavro.Codec, error)

GetSchema returns a goavro.Codec by unique id

func (*SchemaRegistryClient) GetSchemaByVersion 

func (client *SchemaRegistryClient) GetSchemaByVersion(subject string, version int) (*goavro.Codec, error)

GetSchemaByVersion returns a goavro.Codec for the version of the subject

func (*SchemaRegistryClient) GetSubjects 

func (client *SchemaRegistryClient) GetSubjects() ([]string, error)

GetSubjects returns a list of all subjects in the schema registry

func (*SchemaRegistryClient) GetVersions 

func (client *SchemaRegistryClient) GetVersions(subject string) ([]int, error)

GetVersions returns a list of the versions of a subject

func (*SchemaRegistryClient) IsSchemaRegistered 

func (client *SchemaRegistryClient) IsSchemaRegistered(subject string, codec *goavro.Codec) (int, error)

IsSchemaRegistered tests if the schema is registered, if so it returns the unique id of that schema

type SchemaRegistryClientInterface 

type SchemaRegistryClientInterface interface {
	GetSchema(int) (*goavro.Codec, error)
	GetSubjects() ([]string, error)
	GetVersions(string) ([]int, error)
	GetSchemaByVersion(string, int) (*goavro.Codec, error)
	GetLatestSchema(string) (*goavro.Codec, error)
	CreateSubject(string, *goavro.Codec) (int, error)
	IsSchemaRegistered(string, *goavro.Codec) (int, error)
	DeleteSubject(string) error
	DeleteVersion(string, int) error
}

SchemaRegistryClientInterface defines the api for all clients interfacing with schema registry

Source Files

View all Source files

Directories

Path Synopsis
examples
consumer
producer

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.