mqutils
GitLab ↗

Docs / Backends / Apache Kafka

Apache Kafka

Consumer groups, TLS and SASL authentication, atomic producer transactions, and read-committed consumption

See Kafka security and transactions for SASL configuration, the scoped transaction API, and commit-outcome handling.

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

Index

Variables

go
var (
    // ErrTransactionRequired rejects ordinary publishes on a transactional producer.
    ErrTransactionRequired = errors.New("Kafka publication requires InTransaction")
    // ErrTransactionClosed means the callback publisher is no longer usable.
    ErrTransactionClosed = errors.New("Kafka transaction scope is closed or aborted")
    // ErrTransactionOutcomeUnknown means commit may have reached Kafka. Do not replay
    // the callback automatically; reconcile application state before retrying.
    ErrTransactionOutcomeUnknown = errors.New("Kafka transaction commit outcome is unknown")
    // ErrTransactionalProducerClosed indicates that Close released the producer.
    ErrTransactionalProducerClosed = errors.New("Kafka transactional producer is closed")
    // ErrTransactionalProducerFailed requires an explicit Start or a new producer.
    ErrTransactionalProducerFailed = errors.New("Kafka transactional producer must be restarted after failure")
)

funcNewKafkaConsumer

go
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)
  • consumer_group: Consumer group ID (default: “mqutils-default-group”)
  • handler: Name of registered handler function (default: “kafkaLogger”)
  • tls_enabled: Enable TLS encryption; also enabled by kafkas:// (default: false)
  • skip_verify: Skip TLS certificate verification, never disables TLS (default: false)
  • sasl_enabled: Enable SASL authentication (default: false)
  • sasl_mechanism: PLAIN (default), SCRAM-SHA-256, or SCRAM-SHA-512
  • sasl_username, sasl_password: Required credentials when SASL is enabled
  • sasl_authorization_identity: Optional authorization identity
  • sasl_allow_insecure: Explicit development override for SASL without TLS (default: false; credentials otherwise require kafkas:// or tls_enabled)
  • retry_topic: Topic nacked messages are republished to (optional; enables the retry pipeline — the consumer also subscribes to this topic)
  • dead_letter_topic: Topic messages land on once retry_max_retries is exhausted (required when retry_topic is set)
  • retry_max_retries: Retry budget (default: 50 with retry_topic; otherwise 0 means unlimited local redelivery). Exhausted messages are dead-lettered if dead_letter_topic is configured, otherwise acknowledged and dropped.
  • auto_commit: Auto-commit offsets on an interval; when false every Ack commits synchronously (default: true; alias: enable_auto_commit)
  • commit_interval: Auto-commit interval in ms (default: 1000)
  • 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)
  • isolation_level: read_uncommitted (default) or read_committed; the latter hides unfinished/aborted transactional writes and requires Kafka 0.11+
  • kafka_version: Kafka protocol version (default: “2.8.0”)
  • batch_size: Number of messages per batch (default: 5)
  • batch_timeout: Batch collection timeout as a duration (default: “100ms”)
  • enable_batch_processing: Enable batch message processing (default: false)
  • max_concurrent_handlers: Maximum in-flight handlers or batches (default: effective message_channel_buffer; must be positive)

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

funcNewKafkaProducer

go
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/PublishMsg is called with an empty topic.
  • tls_enabled: Enable TLS encryption; also enabled by kafkas:// (default: false)
  • skip_verify: Skip TLS certificate verification, never disables TLS (default: false)
  • sasl_enabled: Enable SASL authentication (default: false)
  • sasl_mechanism: PLAIN (default), SCRAM-SHA-256, or SCRAM-SHA-512
  • sasl_username, sasl_password: Required credentials when SASL is enabled
  • sasl_authorization_identity: Optional authorization identity
  • sasl_allow_insecure: Explicit development override for SASL without TLS (default: false; credentials otherwise require kafkas:// or tls_enabled)
  • compression_type: Compression - none, gzip, snappy, lz4, zstd (default: snappy)
  • required_acks: Ack mode - all, leader, none (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)
  • kafka_version: Kafka protocol version (default: “2.8.0”)

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

funcNewKafkaTransport

go
func NewKafkaTransport() types.Transport

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

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

funcNewMessageBuilder

go
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:

go
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()

typeTransactionalProducer

TransactionalProducer publishes atomic groups of records across Kafka topics and partitions. Transaction IDs must be unique among active producers and stable when replacing the same logical producer after restart.

This API does not commit consumed offsets. Ordinary Publish/PublishMsg reject calls; use the publisher supplied by InTransaction. A producer owns one active callback at a time; callbacks must not recursively call InTransaction on the same producer. Waiting callers can cancel their contexts.

go
type TransactionalProducer interface {
    types.Producer
    // InTransaction commits if fn and all its publishes succeed; otherwise it
    // aborts. The transaction timeout bounds waiting for the callback, but cannot
    // stop its arbitrary application code or external effects. Panics are
    // rethrown after bounded cleanup. Cancellation invalidates the publisher and
    // stops waiting for fn. The caller must separately propagate its context to
    // application work; fn does not receive the internal deadline. Never replay
    // a callback automatically on an error.
    //
    // The scoped publisher expires on return. Its Close requests abort, never
    // commits, and never closes the parent. A callback returning nil after Close
    // still causes ErrTransactionClosed. Close on an expired scope is harmless.
    InTransaction(context.Context, func(types.Publisher) error) error
}

typeTransactionalProducerBuilder

TransactionalProducerBuilder configures an atomic Kafka producer. The private builder also implements TransactionalProducer. Build validates, connects, and returns that same instance. Use a separate builder for each producer.

go
type TransactionalProducerBuilder interface {
    WithURL(string) TransactionalProducerBuilder
    WithConfig(*viper.Viper) TransactionalProducerBuilder
    WithTransactionID(string) TransactionalProducerBuilder
    WithTransactionTimeout(time.Duration) TransactionalProducerBuilder
    Build(context.Context) (TransactionalProducer, error)
}

funcNewTransactionalProducerBuilder

go
func NewTransactionalProducerBuilder() TransactionalProducerBuilder

NewTransactionalProducerBuilder creates a builder for a started Kafka transactional producer. WithConfig accepts the normal Kafka producer settings, plus transaction_id and transaction_timeout (a duration, default one minute).

Generated by gomarkdoc

↑ ↓ move↵ open/ opens search anywhere