Kafka

Apache Kafka event streaming implementation with consumer groups, partitions, and TLS (SASL is not supported)

kafka

v2 note: Import go.digitalxero.dev/mq-kafka/v2. Handlers return error (nil acks). AcknowledgeMessage / NacknowledgeMessage were removed. SASL is not supported; use kafkas:// or tls_enabled for TLS. Consume key is topic (not queue). initial_offset is earliest/latest (oldest/newest accepted as aliases). batch_timeout is a duration string ("100ms").

1

import "go.digitalxero.dev/mq-kafka/v2"

Index

func AcknowledgeMessage

1

func AcknowledgeMessage(ctx context.Context, message types.Message) error

AcknowledgeMessage provides a robust way for handlers to acknowledge messages with retry logic This function is exported to allow handlers to use the same acknowledgment retry mechanism

func NacknowledgeMessage

1

func NacknowledgeMessage(ctx context.Context, message types.Message) error

NacknowledgeMessage provides a robust way for handlers to nack messages with retry logic This function is exported to allow handlers to use the same nack retry mechanism

func NewKafkaConsumer

1

func NewKafkaConsumer(config *viper.Viper) (types.Consumer, error)

NewKafkaConsumer creates a new Apache Kafka consumer with the provided configuration. This function is typically called by mqutils.NewConsumer when it detects a Kafka URL.

Configuration options:

  • url: Kafka broker URLs (required, e.g., “kafka://broker1:9092,broker2:9092”)
  • topic: Topic name to consume from (required; alias: destination)
  • consumer_group: Consumer group ID (required; default: “mqutils-default-group”). This value is now applied to the Sarama group session.
  • handler: Name of registered handler function (default: “kafkaLogger”)
  • tls_enabled: Enable TLS encryption (default: false; also enabled by kafkas://). SASL is not supported.
  • skip_verify: Skip TLS certificate verification (default: false)
  • retry_topic: Topic for message retries (optional; required for nack without losing offsets)
  • retry_max_retries: Max retry attempts (default: 50; alias: max_retries)
  • session_timeout: Session timeout in ms (default: 30000)
  • heartbeat_interval: Heartbeat interval in ms (default: 3000)
  • initial_offset: Where to start reading - earliest/oldest or latest/newest (default: latest; alias: auto_offset_reset)
  • auto_commit: Auto-commit offsets (default: true; alias: enable_auto_commit)
  • batch_size: Number of messages per batch (default: 5)
  • batch_timeout: Batch collection timeout as a duration string (default: “100ms”)
  • enable_batch_processing: Enable batch message processing (default: false)

Returns an error if configuration validation fails or transport creation fails.

func NewKafkaProducer

1

func NewKafkaProducer(config *viper.Viper) (types.Producer, error)

NewKafkaProducer creates a new Apache Kafka producer with the provided configuration. This function is typically called by mqutils.NewProducer when it detects a Kafka URL.

Configuration options:

  • url: Kafka broker URLs (required, e.g., “kafka://broker1:9092,broker2:9092”)
  • topic: Default publish topic (optional; alias: destination). Used when Publish is called with an empty topic.
  • tls_enabled: Enable TLS encryption (default: false; also enabled by kafkas://). SASL is not supported.
  • skip_verify: Skip TLS certificate verification (default: false)
  • compression_type: Compression - none, gzip, snappy, lz4 (default: snappy)
  • required_acks: Ack mode - all, 1, 0 (default: all)
  • max_retries: Max retry attempts for failed sends (default: 3)
  • retry_backoff: Retry backoff in ms (default: 100)
  • flush_frequency: Flush interval in ms (default: 10)
  • flush_messages: Messages before flush (default: 100)
  • flush_bytes: Bytes before flush (default: 1048576)
  • partitioner: Partitioning strategy - hash, random, manual (default: hash)

Returns an error if configuration validation fails or transport creation fails. The producer must be started with Start() before publishing messages.

func NewKafkaTransport

1

func NewKafkaTransport() types.Transport

NewKafkaTransport creates a new Kafka transport implementation. The transport provides low-level Kafka operations including connection management, producer/consumer creation, and message operations.

The transport manages:

  • Kafka client connections with automatic broker discovery
  • Producer instances for publishing
  • Consumer group coordination
  • TLS encryption (kafkas:// or tls_enabled). SASL is not supported.
  • Connection health monitoring

Unlike AMQP, Kafka doesn’t use channels, so this transport manages Sarama client and producer instances directly.

func NewMessageBuilder

1

func NewMessageBuilder() types.MessageBuilder

NewMessageBuilder creates a new Kafka message builder. The builder provides a fluent interface for constructing Kafka messages with all supported properties and attributes.

Example:

msg := kafka.NewMessageBuilder().
    WithBody([]byte(`{"event": "order.placed"}`)).
    WithContentType("application/json").
    WithRoutingKey("order-123"). // Used as partition key
    WithHeaders(map[string]interface{}{
        "source": "order-service",
        "version": "2.0",
    }).
    Build()

type ConfigurationError

ConfigurationError represents a validation error with specific field and message

1
2
3
4

type ConfigurationError struct {
    Field   string
    Message string
}

func (*ConfigurationError) Error

1

func (e *ConfigurationError) Error() string

type ConfigurationErrors

ConfigurationErrors represents multiple validation errors

1
2
3

type ConfigurationErrors struct {
    Errors []*ConfigurationError
}

func (*ConfigurationErrors) Add

1

func (e *ConfigurationErrors) Add(field, message string)

Add adds a new configuration error to the collection

func (*ConfigurationErrors) Error

1

func (e *ConfigurationErrors) Error() string

func (*ConfigurationErrors) HasErrors

1

func (e *ConfigurationErrors) HasErrors() bool

HasErrors returns true if there are any validation errors

Generated by gomarkdoc