kafka

package module

Go to main page

Versions in this module

v2

v2.4.6

Oct 31, 2024

v2.3.6

Oct 30, 2024

Changes in this version

type RetryConfiguration

+ Concurrency int

+ ProducerBatchSize int

+ ProducerBatchTimeout time.Duration

+ QueueCapacity int

v2.3.5

Sep 27, 2024

Changes in this version

+ type ProducerInterceptor interface

+ OnProduce func(ctx ProducerInterceptorContext)

+ type ProducerInterceptorContext struct

+ Context context.Context

+ Message *Message

v2.3.4

Sep 3, 2024

Changes in this version

+ func GetBalancerString(balancer Balancer) string

type BatchConfiguration

+ func (cfg *BatchConfiguration) JSON() string

type ConsumerConfig

+ func (cfg *ConsumerConfig) JSON() string

+ func (cfg *ConsumerConfig) JSONPretty() string

+ func (cfg *ConsumerConfig) String() string

+ type Headers []Header

+ func (hs Headers) Pretty() string

type ProducerConfig

+ func (cfg *ProducerConfig) JSON() string

+ func (cfg *ProducerConfig) JSONPretty() string

+ func (cfg *ProducerConfig) String() string

type ReaderConfig

+ func (cfg ReaderConfig) JSON() string

type RetryConfiguration

+ func (cfg RetryConfiguration) JSON() string

type SASLConfig

+ func (s *SASLConfig) JSON() string

type TLSConfig

+ func (c *TLSConfig) JSON() string

type WriterConfig

+ func (cfg WriterConfig) JSON() string

v2.3.3

Jun 4, 2024

Changes in this version

type BatchConfiguration

+ MessageGroupByteSizeLimit any

type ConsumerConfig

+ VerifyTopicOnStartup bool

type ConsumerMetric

+ TotalErrorCountDuringFetchingMessage int64

type Message

+ func (m *Message) TotalSize() int

type RetryConfiguration

+ VerifyTopicOnStartup bool

v2.3.2

Apr 2, 2024

v2.3.1

Mar 29, 2024

Changes in this version

+ type Balancer kafka.Balancer

+ func GetBalancerCRC32() Balancer

+ func GetBalancerHash() Balancer

+ func GetBalancerLeastBytes() Balancer

+ func GetBalancerMurmur2Balancer() Balancer

+ func GetBalancerReferenceHash() Balancer

+ func GetBalancerRoundRobin() Balancer

type Message

+ ErrDescription string

type RetryConfiguration

+ Balancer Balancer

v2.3.0

Feb 18, 2024

Changes in this version

type Consumer

+ GetMetricCollectors func() []prometheus.Collector

type ConsumerConfig

+ MetricPrefix string

+ type MetricCollector struct

+ func NewMetricCollector(metricPrefix string, consumerMetric *ConsumerMetric) *MetricCollector

+ func (s *MetricCollector) Collect(ch chan<- prometheus.Metric)

+ func (s *MetricCollector) Describe(ch chan<- *prometheus.Desc)

type RetryConfiguration

+ MetricPrefix string

v2.2.9

Feb 15, 2024

Changes in this version

type ConsumerConfig

+ SkipMessageByHeaderFn SkipMessageByHeaderFn

type RetryConfiguration

+ SkipMessageByHeaderFn SkipMessageByHeaderFn

+ type SkipMessageByHeaderFn func(header []kafka.Header) bool

v2.1.9

Jan 15, 2024

Changes in this version

type Consumer

+ Pause func()

+ Resume func()

v2.1.8

Jan 8, 2024

Changes in this version

+ type IncomingMessage struct

v2.0.8

Jan 2, 2024

v2.0.7

Dec 26, 2023

v2.0.7-segmentio-degrade

Jan 2, 2024

v2.0.6

Dec 18, 2023

Changes in this version

type BatchConfiguration

+ PreBatchFn PreBatchFn

+ type PreBatchFn func([]*Message) []*Message

v2.0.5

Dec 10, 2023

v2.0.4

Dec 10, 2023

v2.0.3

Dec 10, 2023

v2.0.2

Dec 10, 2023

Changes in this version

+ const MechanismPlain

+ const MechanismScram

+ const Name

+ func NewBoolPtr(value bool) *bool

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

+ type API interface

+ Start func()

+ Stop func()

+ func NewAPI(cfg *ConsumerConfig, consumerMetric *ConsumerMetric, ...) API

+ type APIConfiguration struct

+ HealthCheckPath *string

+ Port *int

+ type BatchConfiguration struct

+ BatchConsumeFn BatchConsumeFn

+ 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

+ CommitInterval time.Duration

+ Concurrency int

+ ConsumeFn ConsumeFn

+ Dial *DialConfig

+ DistributedTracingConfiguration DistributedTracingConfiguration

+ DistributedTracingEnabled bool

+ LogLevel LogLevel

+ Logger LoggerInterface

+ MessageGroupDuration time.Duration

+ MetricConfiguration MetricConfiguration

+ Rack string

+ Reader ReaderConfig

+ RetryConfiguration RetryConfiguration

+ RetryEnabled bool

+ SASL *SASLConfig

+ TLS *TLSConfig

+ TransactionalRetry *bool

+ type ConsumerMetric struct

+ TotalProcessedMessagesCounter int64

+ TotalUnprocessedMessagesCounter int64

+ type DialConfig struct

+ KeepAlive time.Duration

+ Timeout time.Duration

+ type Dialer struct

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

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

+ type DistributedTracingConfiguration struct

+ Propagator propagation.TextMapPropagator

+ TracerProvider trace.TracerProvider

+ type Header = protocol.Header

+ 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 struct

+ Context context.Context

+ Headers []Header

+ HighWaterMark int64

+ IsFailed bool

+ Key []byte

+ Offset int64

+ Partition int

+ Time time.Time

+ Topic string

+ Value []byte

+ WriterData interface{}

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

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

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

+ type MetricConfiguration struct

+ Path *string

+ 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 Producer interface

+ Close func() error

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

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

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

+ type ProducerConfig struct

+ ClientID string

+ DistributedTracingConfiguration DistributedTracingConfiguration

+ DistributedTracingEnabled bool

+ SASL *SASLConfig

+ TLS *TLSConfig

+ Transport *TransportConfig

+ Writer WriterConfig

+ type Reader interface

+ Close func() error

+ CommitMessages func(messages []kafka.Message) error

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

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

+ func NewReaderWrapper(reader *segmentio.Reader) Reader

+ type ReaderConfig kafka.ReaderConfig

+ type RetryConfiguration struct

+ Brokers []string

+ ClientID string

+ DeadLetterTopic 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 TransportConfig struct

+ DialTimeout time.Duration

+ IdleTimeout time.Duration

+ MetadataTTL time.Duration

+ MetadataTopics []string

+ type Writer interface

+ Close func() error

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

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

+ 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

Other modules containing this package

github.com/Trendyol/kafka-konsumer

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