kafka

Documentation

Index

• type Client
  • func NewClient(config *config.Connector) Client
• type RejectionLog
• type RejectionLogSinkResponseHandler
  • func (r *RejectionLogSinkResponseHandler) OnError(ctx *SinkResponseHandlerContext)
  • func (r *RejectionLogSinkResponseHandler) OnInit(ctx *SinkResponseHandlerInitContext)
  • func (r *RejectionLogSinkResponseHandler) OnSuccess(_ *SinkResponseHandlerContext)
• type SinkResponseHandler
  • func NewRejectionLogSinkResponseHandler() SinkResponseHandler
• type SinkResponseHandlerContext
• type SinkResponseHandlerInitContext

Types

type Client

type Client interface {
	GetEndOffsets(topic string, partitions []int) ([]kafka.PartitionOffsets, error)
	GetPartitions(topic string) ([]int, error)
	CreateCompactedTopic(topic string, partition int, replicationFactor int) error
	Producer() *kafka.Writer
	Consumer(topic string, partition int, startOffset int64) *kafka.Reader
	CheckTopicIsCompacted(topic string) error
	CheckTopics(topics []string) error
}

func NewClient

func NewClient(config *config.Connector) Client

type RejectionLog

type RejectionLog struct {
	Topic string
	Value string
	Error string
	Key   []byte
}

type RejectionLogSinkResponseHandler

type RejectionLogSinkResponseHandler struct {
	KafkaClient Client
	Writer      *kafka.Writer
	Topic       string
	Config      config.Kafka
}

func (*RejectionLogSinkResponseHandler) OnError

func (r *RejectionLogSinkResponseHandler) OnError(ctx *SinkResponseHandlerContext)

func (*RejectionLogSinkResponseHandler) OnInit

func (r *RejectionLogSinkResponseHandler) OnInit(ctx *SinkResponseHandlerInitContext)

func (*RejectionLogSinkResponseHandler) OnSuccess

func (r *RejectionLogSinkResponseHandler) OnSuccess(_ *SinkResponseHandlerContext)

type SinkResponseHandler

type SinkResponseHandler interface {
	OnSuccess(ctx *SinkResponseHandlerContext)
	OnError(ctx *SinkResponseHandlerContext)
	OnInit(ctx *SinkResponseHandlerInitContext)
}

func NewRejectionLogSinkResponseHandler

func NewRejectionLogSinkResponseHandler() SinkResponseHandler

type SinkResponseHandlerContext

type SinkResponseHandlerContext struct {
	Message *message.KafkaMessage
	Err     error
}

type SinkResponseHandlerInitContext

type SinkResponseHandlerInitContext struct {
	KafkaClient Client
	Writer      *kafka.Writer
	Config      config.Kafka
}