The highest tagged major version is
README
¶
kafka-client-go
Description
Library for producing and consuming messages directly from Kafka.
Only perseverant connections (for both producer & consumer) are provided i.e. establishing connections to Kafka is retried until successful.
Attempting to consume or produce messages or to check connectivity will fail if connection hasn't been established yet.
The library is NOT using Zookeeper to connect to Kafka under the hood.
Usage
Importing:
import "github.com/Financial-Times/kafka-client-go/v3"
Producer
Creating a producer:
config := kafka.ProducerConfig{
BrokersConnectionString: "", // Comma-separated list of Kafka brokers
Topic: "", // Topic to publish to
}
initialDelay := time.Second // Duration to await before trying to establish connection
retryInterval := time.Second // Duration between each retry for establishing connection
logger := logger.NewUPPLogger(...)
producer := kafka.NewProducer(config, logger, initialDelay, retryInterval)
The connection to Kafka is started in a separate go routine when creating the producer.
Sending a message:
headers := map[string]string{}
body := ""
message := kafka.NewFTMessage(headers, body)
err := producer.SendMessage(message)
// Error handling
Connection should be closed by the client:
err := producer.Close()
// Error handling
Consumer
Creating a consumer:
config := kafka.ConsumerConfig{
BrokersConnectionString: "", // Comma-separated list of Kafka brokers
ConsumerGroup: "", // Unique name of a consumer group
}
topics := []*kafka.Topic{
kafka.NewTopic(""), // Topic to consume from
kafka.NewTopic("", kafka.WithLagTolerance(50)), // Topic to consume from with custom lag tolerance used for monitoring
}
retryInterval := time.Second // Duration between each retry for establishing connection
logger := logger.NewUPPLogger(...)
consumer := kafka.NewConsumer(config, topics, logger, retryInterval)
Consuming messages:
Consumer groups are lazily initialized i.e. establishing connection to Kafka is done within StartListening().
handler := func(message kafka.FTMessage) {
// Message handling
}
go consumer.StartListening(handler) // Blocking until connection is established
Connections should be closed by the client:
err := consumer.Close()
// Error handling
Testing
go test --race -v ./...
NB: Some tests in this project require a local Kafka (port 29092). Use the -short flag in order to omit those.
Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func DefaultConsumerOptions ¶
DefaultConsumerOptions returns a new sarama configuration with predefined default settings.
func DefaultProducerOptions ¶
DefaultProducerOptions creates a new Sarama producer configuration with default values.
Types ¶
type Consumer ¶
type Consumer struct {
// contains filtered or unexported fields
}
Consumer which will keep trying to reconnect to Kafka on a specified interval. The underlying consumer group is created lazily when message listening is started.
func NewConsumer ¶
func NewConsumer(config ConsumerConfig, topics []*Topic, log *logger.UPPLogger, connectionRetryInterval time.Duration) *Consumer
func (*Consumer) Close ¶
Close closes the consumer connection to Kafka if the consumer is connected.
func (*Consumer) ConnectivityCheck ¶
ConnectivityCheck checks whether a connection to Kafka can be established.
func (*Consumer) MonitorCheck ¶
MonitorCheck checks whether the consumer group is lagging behind when reading messages.
func (*Consumer) StartListening ¶
StartListening is a blocking call that tries to establish a connection to Kafka and then starts listening for messages.
type ConsumerConfig ¶
type ConsumerConfig struct {
BrokersConnectionString string
ConsumerGroup string
// Time interval between each offset fetching request. Default is 3 minutes.
OffsetFetchInterval time.Duration
// Total count of offset fetching request failures until consumer status is marked as unknown.
// Default is 5. Note: A single failure will result in follow-up requests to be sent on
// shorter interval than the value of OffsetFetchInterval until successful.
OffsetFetchMaxFailureCount int
Options *sarama.Config
}
type Producer ¶
type Producer struct {
// contains filtered or unexported fields
}
Producer which will keep trying to reconnect to Kafka on a specified interval. The underlying producer is created in a separate go-routine when the Producer is initialized.
func NewProducer ¶
func NewProducer(config ProducerConfig, logger *logger.UPPLogger, initialDelay, retryInterval time.Duration) *Producer
func (*Producer) ConnectivityCheck ¶
ConnectivityCheck checks whether a connection to Kafka can be established.
func (*Producer) SendMessage ¶
SendMessage checks if the producer is connected, then sends a message to Kafka.
type ProducerConfig ¶
type Topic ¶
type Topic struct {
Name string
// contains filtered or unexported fields
}
func NewTopic ¶
func NewTopic(name string, opts ...TopicOption) *Topic
type TopicOption ¶
type TopicOption func(topic *Topic)
func WithLagTolerance ¶
func WithLagTolerance(tolerance int64) TopicOption
WithLagTolerance sets custom lag tolerance threshold used for monitoring. Consumer lagging behind with more messages than the configured tolerance will be reported as unhealthy. Default is 500 messages.
Source Files