kafkaclient

README

Kafka Client

A Go package providing Kafka consumer and producer implementations with support for batch processing and async/sync message delivery.

Features

• Sarama Integration: Built on top of github.com/IBM/sarama for reliable Kafka client implementation
• Batch Consumer: Process messages in batches with configurable size and timeout
• Sync Producer: Blocking message delivery with guaranteed ordering
• Async Producer: Non-blocking message delivery for high throughput
• Structured Logging: Built-in telemetry and metrics
• Graceful Shutdown: Proper resource cleanup on termination
• Error Recovery: Automatic retry and panic recovery mechanisms
• Telemetry Integration: Built-in metrics collection for monitoring

Installation

go get github.com/n-r-w/kafkaclient

Documentation

• Consumer Documentation
• Producer Documentation

Example

full example

Documentation

Index

• Variables
• func PrepareDefaultConfigSASL(config *sarama.Config, mechanism sarama.SASLMechanism, user string, ...)
• func SaslMechanismFromString(saslMechanism string) (sarama.SASLMechanism, error)
• func ValidateClientID(clientID string) error
• type XDGSCRAMClient
  • func NewXDGSCRAMClient(hashGeneratorFcn scram.HashGeneratorFcn) *XDGSCRAMClient
  • func (x *XDGSCRAMClient) Begin(userName, password, authzID string) (err error)
  • func (x *XDGSCRAMClient) Done() bool
  • func (x *XDGSCRAMClient) Step(challenge string) (response string, err error)

Variables

var (
	// SHA256 - hash function for SCRAM.
	SHA256 scram.HashGeneratorFcn = sha256.New //nolint:gochecknoglobals // ok
	// SHA512 - hash function for SCRAM.
	SHA512 scram.HashGeneratorFcn = sha512.New //nolint:gochecknoglobals // ok
)

var KafkaVersion = sarama.V3_4_0_0 //nolint:gochecknoglobals // ok

KafkaVersion is the default Kafka version.

Functions

func PrepareDefaultConfigSASL

func PrepareDefaultConfigSASL(
	config *sarama.Config,
	mechanism sarama.SASLMechanism,
	user string,
	password string,
)

PrepareDefaultConfigSASL prepares the configuration for creating a producer or consumer with SASL.

func SaslMechanismFromString

func SaslMechanismFromString(saslMechanism string) (sarama.SASLMechanism, error)

SaslMechanismFromString converts a config string to the corresponding SASL type in sarama. Can be helpful for config parsing.

func ValidateClientID

func ValidateClientID(clientID string) error

ValidateClientID checks the validity of ClientID for Kafka.

Types

type XDGSCRAMClient

type XDGSCRAMClient struct {
	*scram.Client
	*scram.ClientConversation
	scram.HashGeneratorFcn
}

XDGSCRAMClient implements sarama.SCRAMClient interface for SASL configuration https://github.com/IBM/sarama/blob/main/examples/sasl_scram_client/scram_client.go

func NewXDGSCRAMClient

func NewXDGSCRAMClient(hashGeneratorFcn scram.HashGeneratorFcn) *XDGSCRAMClient

NewXDGSCRAMClient creates a new instance of XDGSCRAMClient.

func (*XDGSCRAMClient) Begin

func (x *XDGSCRAMClient) Begin(userName, password, authzID string) (err error)

Begin starts SASL session.

func (*XDGSCRAMClient) Done

func (x *XDGSCRAMClient) Done() bool

Done completes SASL session.

func (*XDGSCRAMClient) Step

func (x *XDGSCRAMClient) Step(challenge string) (response string, err error)

Step performs one iteration of SASL session.