kafkaclient

README

KafkaClient

KafkaClient is a Go library designed to simplify and enhance interactions with Apache Kafka. It provides robust features for message production, including connection retries, batch processing, and seamless integration with a ring buffer for efficient data handling.

Features

• Reliable Kafka Connection: Built-in retry logic to ensure stable connections.
• Batch Message Processing: Efficiently send messages in batches to improve throughput.
• Ring Buffer Support: Integrates with kafkabuff for optimized buffering and flow control.
• Customizable Logger: Plug in your logger or use the default standard logger.
• Readiness State: Track Kafka client's readiness state for better monitoring and resilience.

Installation

go get github.com/vagabundor/kafkaclient

Usage

Creating a Kafka Client

package main

import (
    "time"

    "github.com/IBM/sarama"
    "github.com/vagabundor/kafkaclient"
)

func main() {
    brokers := []string{"localhost:9092"}
    maxRetries := 5
    retryInterval := 2 * time.Second

    client, err := kafkaclient.NewKafkaClient(brokers, maxRetries, retryInterval, nil, nil)
    if err != nil {
        panic(err)
    }
    defer client.Close()

    // Check readiness
    if client.IsReady() {
        println("Kafka client is ready!")
    }
}

Sending a Batch of Messages

messages := []*sarama.ProducerMessage{
    {Topic: "example-topic", Value: sarama.StringEncoder("Message 1")},
    {Topic: "example-topic", Value: sarama.StringEncoder("Message 2")},
}

err := client.SendBatch(messages, "example-topic")
if err != nil {
    println("Failed to send messages:", err)
}

Using the Ring Buffer for Batch Sending

import "github.com/vagabundor/kafkabuff"

ringBuffer := kafkabuff.NewRingBuffer(1000) // Create a ring buffer with a capacity of 1000
ringBuffer.Add(&sarama.ProducerMessage{Topic: "example-topic", Value: sarama.StringEncoder("Message")})

client.StartBatchSender(ringBuffer, 10, time.Second, "example-topic")

Configuration Options

• Brokers: Kafka broker addresses (e.g., []string{"localhost:9092"}).
• Retries: Maximum number of retry attempts for connecting or sending messages.
• Retry Interval: Duration between retries. When set to 0, retries are infinite.
• Logger: Use a custom logger by implementing the Logger interface.

Documentation

Index

• type KafkaClient
  • func NewKafkaClient(brokers []string, maxRetries int, retryInterval time.Duration, ...) (*KafkaClient, error)
  • func (kc *KafkaClient) Close() error
  • func (kc *KafkaClient) IsReady() bool
  • func (kc *KafkaClient) SendBatch(batch []*sarama.ProducerMessage) error
  • func (kc *KafkaClient) StartBatchSender(ringBuffer *kafkabuff.RingBuffer, batchSize int, interval time.Duration)
• type Logger
• type StdLogger
  • func (l *StdLogger) Debugf(format string, args ...interface{})
  • func (l *StdLogger) Errorf(format string, args ...interface{})
  • func (l *StdLogger) Infof(format string, args ...interface{})
  • func (l *StdLogger) Warnf(format string, args ...interface{})

Types

type KafkaClient

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

func NewKafkaClient

func NewKafkaClient(brokers []string, maxRetries int, retryInterval time.Duration, config *sarama.Config, logger Logger) (*KafkaClient, error)

NewKafkaClient creates a new KafkaClient instance.

func (*KafkaClient) Close

func (kc *KafkaClient) Close() error

Close closes the connection to Kafka.

func (*KafkaClient) IsReady

func (kc *KafkaClient) IsReady() bool

IsReady returns the readiness state of the Kafka client.

func (*KafkaClient) SendBatch

func (kc *KafkaClient) SendBatch(batch []*sarama.ProducerMessage) error

SendBatch sends a batch of messages to Kafka with retries.

func (*KafkaClient) StartBatchSender

func (kc *KafkaClient) StartBatchSender(ringBuffer *kafkabuff.RingBuffer, batchSize int, interval time.Duration)

StartBatchSender starts sending messages from the ring buffer to Kafka.

type Logger

type Logger interface {
	Debugf(format string, args ...interface{})
	Infof(format string, args ...interface{})
	Warnf(format string, args ...interface{})
	Errorf(format string, args ...interface{})
}

Logger interface

type StdLogger

type StdLogger struct{}

func (*StdLogger) Debugf

func (l *StdLogger) Debugf(format string, args ...interface{})

func (*StdLogger) Errorf

func (l *StdLogger) Errorf(format string, args ...interface{})

func (*StdLogger) Infof

func (l *StdLogger) Infof(format string, args ...interface{})

func (*StdLogger) Warnf

func (l *StdLogger) Warnf(format string, args ...interface{})