kafka

package module

Go to main page

Versions in this module

v4

v4.1.0

Feb 8, 2024

Changes in this version

+ func SafeSendProducerMessageWithOtel(ctx context.Context, ch chan<- *sarama.ProducerMessage, ...) (err error)

type ConsumerGroupConfig

+ OtelEnabled *bool

type ProducerConfig

+ OtelEnabled *bool

v4.0.1

Jan 11, 2024

v4.0.0

Dec 15, 2023

Changes in this version

+ const Closed

+ const Closer

+ const Consume

+ const ErrorChanBufferSize

+ const Errors

+ const Initialised

+ const MsgHealthyConsumerGroup

+ const MsgHealthyProducer

+ const OffsetNewest

+ const OffsetOldest

+ const Output

+ const ServiceName

+ const TraceIDHeaderKey

+ const Upstream

+ var ErrTLSCannotLoadCACerts = errors.New("cannot load CA Certs")

+ var SaramaNewBroker interfaces.BrokerGenerator = func(addr string) interfaces.SaramaBroker

+ func DrainTopic(ctx context.Context, cfg *DrainTopicConfig, in *DrainTopicInput, ...) error

+ func DrainTopics(ctx context.Context, cfg *DrainTopicConfig, in ...*DrainTopicInput)

+ func GetPrincipal(app, subnet, domain string) string

+ func GetRetryTime(attempt int, retryTime, maxRetryTime time.Duration) time.Duration

+ func NewAdmin(brokerAddrs []string, adminCfg *AdminConfig) (sarama.ClusterAdmin, error)

+ func SafeClose(ch chan struct{}) (justClosed bool)

+ func SafeCloseBool(ch chan bool) (justClosed bool)

+ func SafeCloseBytes(ch chan BytesMessage) (justClosed bool)

+ func SafeCloseErr(ch chan error) (justClosed bool)

+ func SafeCloseMessage(ch chan Message) (justClosed bool)

+ func SafeSendBool(ch chan bool, val bool) (err error)

+ func SafeSendBytes(ch chan BytesMessage, val BytesMessage) (err error)

+ func SafeSendConsumerMessage(ch chan<- *sarama.ConsumerMessage, val *sarama.ConsumerMessage) (err error)

+ func SafeSendErr(ch chan error, val error) (err error)

+ func SafeSendProducerMessage(ctx context.Context, ch chan<- *sarama.ProducerMessage, ...) (err error)

+ func SetMaxMessageSize(maxSize int32)

+ func UnwrapLogData(err error) log.Data

+ func WaitWithTimeout(wg *sync.WaitGroup, tout time.Duration) bool

+ type Acls []*sarama.AclCreation

+ func (t Acls) Apply(adm sarama.ClusterAdmin) error

+ type AdminConfig struct

+ KafkaVersion *string

+ KeepAlive *time.Duration

+ RetryBackoff *time.Duration

+ RetryMax *int

+ SecurityConfig *SecurityConfig

+ func (a *AdminConfig) Get() (*sarama.Config, error)

+ type Batch struct

+ func NewBatch(batchSize int) *Batch

+ func (batch *Batch) Add(message Message)

+ func (batch *Batch) Clear()

+ func (batch *Batch) Commit()

+ func (batch *Batch) IsEmpty() bool

+ func (batch *Batch) IsFull() bool

+ func (batch *Batch) Size() int

+ type BatchHandler func(ctx context.Context, batch []Message) error

+ type BytesMessage struct

+ Context context.Context

+ Value []byte

+ type Commiter interface

+ Commit func() bool

+ type ConsumerGroup struct

+ func NewConsumerGroup(ctx context.Context, cgConfig *ConsumerGroupConfig) (*ConsumerGroup, error)

+ func NewConsumerGroupWithGenerators(ctx context.Context, cgConfig *ConsumerGroupConfig, ...) (*ConsumerGroup, error)

+ func (cg *ConsumerGroup) Channels() *ConsumerGroupChannels

+ func (cg *ConsumerGroup) Checker(ctx context.Context, state *healthcheck.CheckState) error

+ func (cg *ConsumerGroup) Close(ctx context.Context, optFuncs ...OptFunc) (err error)

+ func (cg *ConsumerGroup) GroupName() string

+ func (cg *ConsumerGroup) Initialise(ctx context.Context) error

+ func (cg *ConsumerGroup) IsInitialised() bool

+ func (cg *ConsumerGroup) LogErrors(ctx context.Context)

+ func (cg *ConsumerGroup) OnHealthUpdate(status string)

+ func (cg *ConsumerGroup) RegisterBatchHandler(ctx context.Context, batchHandler BatchHandler) error

+ func (cg *ConsumerGroup) RegisterHandler(ctx context.Context, h Handler) error

+ func (cg *ConsumerGroup) SaramaCgHandler() *SaramaHandler

+ func (cg *ConsumerGroup) Start() error

+ func (cg *ConsumerGroup) State() State

+ func (cg *ConsumerGroup) StateWait(state State)

+ func (cg *ConsumerGroup) Stop() error

+ func (cg *ConsumerGroup) StopAndWait() error

+ func (cg *ConsumerGroup) Topic() string

+ type ConsumerGroupChannels struct

+ Closed chan struct{}

+ Closer chan struct{}

+ Consume chan bool

+ Errors chan error

+ Initialised chan struct{}

+ State *ConsumerStateChannels

+ Upstream chan Message

+ func CreateConsumerGroupChannels(upstreamBufferSize, errorsBufferSize int) *ConsumerGroupChannels

+ func (consumerChannels *ConsumerGroupChannels) Validate() error

+ type ConsumerGroupConfig struct

+ BatchSize *int

+ BatchWaitTime *time.Duration

+ BrokerAddrs []string

+ GroupName string

+ KafkaVersion *string

+ KeepAlive *time.Duration

+ MaxRetryPeriod *time.Duration

+ MessageConsumeTimeout *time.Duration

+ MinBrokersHealthy *int

+ MinRetryPeriod *time.Duration

+ NumWorkers *int

+ Offset *int64

+ RetryBackoff *time.Duration

+ RetryBackoffFunc *func(retries int) time.Duration

+ SecurityConfig *SecurityConfig

+ Topic string

+ func (c *ConsumerGroupConfig) Get() (*sarama.Config, error)

+ func (c *ConsumerGroupConfig) Validate() error

+ type ConsumerStateChannels struct

+ Closing *StateChan

+ Consuming *StateChan

+ Initialising *StateChan

+ Starting *StateChan

+ Stopped *StateChan

+ Stopping *StateChan

+ type DrainTopicConfig struct

+ BatchSize int

+ BrokerAddrs []string

+ KafkaVersion *string

+ Timeout time.Duration

+ type DrainTopicInput struct

+ GroupName string

+ Topic string

+ type Error struct

+ func NewError(err error, logData map[string]interface{}) *Error

+ func (e *Error) Error() string

+ func (e *Error) LogData() map[string]interface{}

+ func (e *Error) Unwrap() error

+ type Handler func(ctx context.Context, workerID int, msg Message) error

+ type HealthInfo struct

+ HasTopic bool

+ Reachable bool

+ type HealthInfoMap struct

+ func Healthcheck(ctx context.Context, brokers []interfaces.SaramaBroker, topic string, ...) HealthInfoMap

+ func (h *HealthInfoMap) Set(broker interfaces.SaramaBroker, healthInfo HealthInfo)

+ func (h *HealthInfoMap) UpdateStatus(state *health.CheckState, minHealthyThreshold int, msgHealthy string) error

+ type IConsumerGroup interface

+ Channels func() *ConsumerGroupChannels

+ Checker func(ctx context.Context, state *healthcheck.CheckState) error

+ Close func(ctx context.Context, optFuncs ...OptFunc) error

+ Initialise func(ctx context.Context) error

+ IsInitialised func() bool

+ LogErrors func(ctx context.Context)

+ OnHealthUpdate func(status string)

+ RegisterBatchHandler func(ctx context.Context, batchHandler BatchHandler) error

+ RegisterHandler func(ctx context.Context, h Handler) error

+ Start func() error

+ State func() State

+ StateWait func(state State)

+ Stop func() error

+ StopAndWait func() error

+ type IProducer interface

+ AddHeader func(key, value string)

+ Channels func() *ProducerChannels

+ Checker func(ctx context.Context, state *healthcheck.CheckState) error

+ Close func(ctx context.Context) (err error)

+ Initialise func(ctx context.Context) error

+ IsInitialised func() bool

+ LogErrors func(ctx context.Context)

+ Send func(ctx context.Context, schema *avro.Schema, event interface{}) error

+ type Message interfaces.Message

+ type OptFunc func()

+ type Producer struct

+ func NewProducer(ctx context.Context, pConfig *ProducerConfig) (producer *Producer, err error)

+ func NewProducerWithGenerators(ctx context.Context, pConfig *ProducerConfig, ...) (*Producer, error)

+ func (p *Producer) AddHeader(key, value string)

+ func (p *Producer) Channels() *ProducerChannels

+ func (p *Producer) Checker(ctx context.Context, state *healthcheck.CheckState) error

+ func (p *Producer) Close(ctx context.Context) (err error)

+ func (p *Producer) Initialise(ctx context.Context) error

+ func (p *Producer) IsInitialised() bool

+ func (p *Producer) LogErrors(ctx context.Context)

+ func (p *Producer) Send(ctx context.Context, schema *avro.Schema, event interface{}) error

+ type ProducerChannels struct

+ Closed chan struct{}

+ Closer chan struct{}

+ Errors chan error

+ Initialised chan struct{}

+ Output chan BytesMessage

+ func CreateProducerChannels() *ProducerChannels

+ func (producerChannels *ProducerChannels) Validate() error

+ type ProducerConfig struct

+ BrokerAddrs []string

+ KafkaVersion *string

+ KeepAlive *time.Duration

+ MaxMessageBytes *int

+ MaxRetryPeriod *time.Duration

+ MinBrokersHealthy *int

+ MinRetryPeriod *time.Duration

+ RetryBackoff *time.Duration

+ RetryBackoffFunc *func(retries, maxRetries int) time.Duration

+ RetryMax *int

+ SecurityConfig *SecurityConfig

+ Topic string

+ func (p *ProducerConfig) Get() (*sarama.Config, error)

+ func (p *ProducerConfig) Validate() error

+ type SaramaHandler struct

+ func (sh *SaramaHandler) Cleanup(session sarama.ConsumerGroupSession) error

+ func (sh *SaramaHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

+ func (sh *SaramaHandler) Setup(session sarama.ConsumerGroupSession) error

+ type SaramaMessage struct

+ func NewSaramaMessage(m *sarama.ConsumerMessage, s sarama.ConsumerGroupSession, ud chan struct{}) *SaramaMessage

+ func (m SaramaMessage) Commit()

+ func (m SaramaMessage) CommitAndRelease()

+ func (m SaramaMessage) Context() context.Context

+ func (m SaramaMessage) GetData() []byte

+ func (m SaramaMessage) GetHeader(key string) string

+ func (m SaramaMessage) Mark()

+ func (m SaramaMessage) Offset() int64

+ func (m SaramaMessage) Release()

+ func (m SaramaMessage) UpstreamDone() chan struct{}

+ type SecurityConfig struct

+ ClientCert string

+ ClientKey string

+ InsecureSkipVerify bool

+ RootCACerts string

+ func GetSecurityConfig(caCerts, clientCert, clientKey string, skipVerify bool) *SecurityConfig

+ type State int

+ const Closing

+ const Consuming

+ const Initialising

+ const Starting

+ const Stopped

+ const Stopping

+ func (s State) String() string

+ type StateChan struct

+ func NewStateChan() *StateChan

+ func (sc *StateChan) Channel() chan struct{}

+ func (sc *StateChan) RWMutex() *sync.RWMutex

+ func (sc *StateChan) Wait()

+ type StateMachine struct

+ func NewConsumerStateMachine() *StateMachine

+ func (sm *StateMachine) Get() State

+ func (sm *StateMachine) GetChan(s State) *StateChan

+ func (sm *StateMachine) Set(newState State)

+ func (sm *StateMachine) SetIf(allowed []State, newState State) error

+ func (sm *StateMachine) String() string

+ type TopicAuth struct

+ App string

+ Hosts []string

+ Operations []sarama.AclOperation

+ Subnets []string

+ Topic string

+ func (t TopicAuth) GetAcls(domain string) Acls

+ type TopicAuthList struct

+ Acls []TopicAuth

+ Brokers []string

+ Domain string

+ func (t TopicAuthList) Apply(adm sarama.ClusterAdmin) error

Other modules containing this package

github.com/ONSdigital/dp-kafka

github.com/ONSdigital/dp-kafka/v2

github.com/ONSdigital/dp-kafka/v3

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL