kafka

Oct 6, 2023

Changes in this version

type ConsumerConfig

+ DistributedTracingConfiguration DistributedTracingConfiguration

+ DistributedTracingEnabled bool

+ type DistributedTracingConfiguration struct

+ Propagator propagation.TextMapPropagator

+ TracerProvider trace.TracerProvider

type Message

+ Context context.Context

+ Headers []kafka.Header

+ HighWaterMark int64

+ Key []byte

+ Offset int64

+ Partition int

+ Time time.Time

+ Topic string

+ Value []byte

+ WriterData interface{}

+ type OtelKafkaKonsumerWriter interface

+ Close func() error

+ WriteMessage func(ctx context.Context, msg segmentio.Message) error

+ WriteMessages func(ctx context.Context, msgs []segmentio.Message) error

type ProducerConfig

+ DistributedTracingConfiguration DistributedTracingConfiguration

+ DistributedTracingEnabled bool

type Reader

+ func NewOtelReaderWrapper(cfg *ConsumerConfig, reader *segmentio.Reader) (Reader, error)

type Writer

+ func NewOtelProducer(cfg *ProducerConfig, writer *segmentio.Writer) (Writer, error)

Sep 17, 2023

Changes in this version

+ type Reader interface

+ Close func() error

+ ReadMessage func(ctx context.Context) (*kafka.Message, error)

+ func NewReaderWrapper(reader *segmentio.Reader) Reader

+ type Writer interface

+ Close func() error

+ WriteMessages func(context.Context, ...kafka.Message) error

Sep 11, 2023

Changes in this version

type RetryConfiguration

+ DeadLetterTopic string

Sep 6, 2023

Changes in this version

type ConsumerConfig

+ CommitInterval time.Duration

Sep 1, 2023

Changes in this version

type ConsumerConfig

+ Dial *DialConfig

+ type DialConfig struct

+ KeepAlive time.Duration

+ Timeout time.Duration

type ProducerConfig

+ Transport *TransportConfig

+ type TransportConfig struct

+ DialTimeout time.Duration

+ IdleTimeout time.Duration

+ MetadataTTL time.Duration

+ MetadataTopics []string

Aug 24, 2023

Changes in this version

type Producer

+ ProduceBatch func(ctx context.Context, messages []Message) error

Aug 20, 2023

Changes in this version

+ const Name

+ func NewMetricMiddleware(cfg *ConsumerConfig, app *fiber.App, consumerMetric *ConsumerMetric, ...) (func(ctx *fiber.Ctx) error, error)

+ type ConsumerMetric struct

+ TotalProcessedBatchMessagesCounter int64

+ TotalProcessedMessagesCounter int64

+ TotalUnprocessedBatchMessagesCounter int64

+ TotalUnprocessedMessagesCounter int64

Aug 8, 2023

Changes in this version

+ const MechanismPlain

+ const MechanismScram

+ type API interface

+ Start func()

+ Stop func()

+ func NewAPI(cfg *ConsumerConfig) API

+ type APIConfiguration struct

+ HealthCheckPath *string

+ Port *int

+ type BatchConfiguration struct

+ BatchConsumeFn BatchConsumeFn

+ MessageGroupDuration time.Duration

+ MessageGroupLimit int

+ type BatchConsumeFn func([]Message) error

+ type ConsumeFn func(Message) error

+ type Consumer interface

+ Consume func()

+ Stop func() error

+ WithLogger func(logger LoggerInterface)

+ func NewConsumer(cfg *ConsumerConfig) (Consumer, error)

+ type ConsumerConfig struct

+ APIConfiguration APIConfiguration

+ APIEnabled bool

+ BatchConfiguration *BatchConfiguration

+ ClientID string

+ Concurrency int

+ ConsumeFn ConsumeFn

+ LogLevel LogLevel

+ Logger LoggerInterface

+ MetricConfiguration MetricConfiguration

+ Rack string

+ Reader ReaderConfig

+ RetryConfiguration RetryConfiguration

+ RetryEnabled bool

+ SASL *SASLConfig

+ TLS *TLSConfig

+ type Dialer struct

+ func (t *Dialer) SetSASL(mechanism sasl.Mechanism)

+ func (t *Dialer) SetTLSConfig(config *tls.Config)

+ type Layer interface

+ SetSASL func(mechanism sasl.Mechanism)

+ SetTLSConfig func(config *tls.Config)

+ type LogLevel string

+ const LogLevelDebug

+ const LogLevelError

+ const LogLevelInfo

+ const LogLevelWarn

+ type LoggerInterface interface

+ Debug func(args ...interface{})

+ Debugf func(format string, args ...interface{})

+ Error func(args ...interface{})

+ Errorf func(format string, args ...interface{})

+ Errorw func(msg string, keysAndValues ...interface{})

+ Info func(args ...interface{})

+ Infof func(format string, args ...interface{})

+ Infow func(msg string, keysAndValues ...interface{})

+ Warn func(args ...interface{})

+ Warnf func(format string, args ...interface{})

+ Warnw func(msg string, keysAndValues ...interface{})

+ With func(args ...interface{}) LoggerInterface

+ func NewZapLogger(level LogLevel) LoggerInterface

+ type Mechanism string

+ type Message kafka.Message

+ func (m *Message) AddHeader(header ...kafka.Header)

+ func (m *Message) Header(key string) *kafka.Header

+ func (m *Message) RemoveHeader(header kafka.Header)

+ type MetricConfiguration struct

+ Path *string

+ type Producer interface

+ Close func() error

+ Produce func(ctx context.Context, message Message) error

+ func NewProducer(cfg ProducerConfig) (Producer, error)

+ type ProducerConfig struct

+ ClientID string

+ SASL *SASLConfig

+ TLS *TLSConfig

+ Writer WriterConfig

+ type ReaderConfig kafka.ReaderConfig

+ type RetryConfiguration struct

+ Brokers []string

+ ClientID string

+ LogLevel LogLevel

+ MaxRetry int

+ Rack string

+ SASL *SASLConfig

+ StartTimeCron string

+ TLS *TLSConfig

+ Topic string

+ WorkDuration time.Duration

+ type SASLConfig struct

+ Password string

+ Type Mechanism

+ Username string

+ func (s *SASLConfig) IsEmpty() bool

+ func (s *SASLConfig) Mechanism() (sasl.Mechanism, error)

+ type TLSConfig struct

+ IntermediateCAPath string

+ RootCAPath string

+ func (c *TLSConfig) IsEmpty() bool

+ func (c *TLSConfig) TLSConfig() (*tls.Config, error)

+ type Transport struct

+ func (t *Transport) SetSASL(mechanism sasl.Mechanism)

+ func (t *Transport) SetTLSConfig(config *tls.Config)

+ type WriterConfig struct

+ AllowAutoTopicCreation bool

+ Async bool

+ Balancer kafka.Balancer

+ BatchBytes int64

+ BatchSize int

+ BatchTimeout time.Duration

+ Brokers []string

+ Completion func(messages []kafka.Message, err error)

+ Compression kafka.Compression

+ ErrorLogger kafka.Logger

+ Logger kafka.Logger

+ MaxAttempts int

+ ReadTimeout time.Duration

+ RequiredAcks kafka.RequiredAcks

+ Topic string

+ WriteBackoffMax time.Duration

+ WriteBackoffMin time.Duration

+ WriteTimeout time.Duration