Docs / FAQ

FAQ

Common questions about backends, delivery guarantees, configuration and testing

General

Why use mqutils instead of the broker’s SDK?

Because the parts that are the same across brokers (settling a message from a handler’s return value, retry budgets, batching, reconnect, graceful shutdown, health checks) are written once in the shared runtime, and the parts that differ live behind a URL scheme. You learn one handler contract and one configuration vocabulary. See Why mqutils for what it does and, just as important, what it does not do.

Which brokers are supported?

Seven backend modules, each imported separately; RabbitMQ Streams additionally has an optional Redis deduplication adapter:

Consumers and producers register identical scheme sets.

Which backends retry and dead-letter?

max_retries is honored everywhere the broker can redeliver. What happens when the budget is exhausted depends on the broker:

• AMQP: routed to the dead_letter_exchange. On RabbitMQ 4.3 and later the retry queue uses the broker’s native delayed retry.
• Kafka: retry topic, then the dead-letter topic.
• RabbitMQ Streams: bounded local retry, then confirmed forwarding to a configured dead-letter stream. Without one, stop without advancing unless discard is explicitly enabled.
• SQS: the queue’s RedrivePolicy moves it to the dead-letter queue.
• Pub/Sub: the subscription’s DeadLetterPolicy, which needs IAM grants to the Pub/Sub service account.
• JetStream and Redis Streams: dropped and logged. There is no dead-letter republish.
• NATS Core and Redis Pub/Sub: at-most-once. Nothing is redelivered, so there is nothing to retry.

Does Pub/Sub give exactly-once delivery?

Not by default. exactly_once_delivery is an opt-in flag passed through to the Pub/Sub subscription. mqutils itself makes no exactly-once guarantee on any backend; treat handlers as at-least-once and make them idempotent.

Can I move between brokers?

Yes, with two caveats. The handler and the consumer runtime are unchanged. The configuration block is not: you change the URL scheme, import the matching module, and replace any backend-native keys (exchange, routing_key, subject, stream_name) with the new backend’s equivalents. The canonical keys destination, consumer_group and max_retries work everywhere.

// RabbitMQ
config.Set("url", "amqp://localhost:5672/")
config.Set("destination", "orders")

// Kafka: same handler, new scheme, new module imported
config.Set("url", "kafka://localhost:9092")
config.Set("destination", "orders")
config.Set("consumer_group", "billing")

Is it production-ready?

It has unit tests on every module, broker-backed integration tests (docker-compose.test.yml), and a public-API acceptance suite in acceptance/. Read what it does not do before deciding; the honest limits are delivery semantics, application idempotency, and no cross-broker failover. Kafka SASL and atomic producer transactions are covered in the Kafka guide; the Stream guide explains its separate backend and deployment prerequisites.

Do Kafka transactions or Stream deduplication make processing exactly once?

No. Kafka transactions atomically publish records but do not include this runtime’s consumer-offset commits or arbitrary database effects. Stream broker deduplication suppresses repeated writes for an explicit producer/sequence identity; application deduplication suppresses handler repeats for a retained logical key. A crash between an external effect and saving completion remains possible. Use a transactional inbox to commit its receipt and database effect together. See Kafka transactions and Stream deduplication.

Can I use a GCP Pub/Sub schema?

An existing topic with an externally configured Avro or Protobuf schema can already validate the raw bytes published through mqutils. Google rejects nonconforming publication. The library does not currently provision schema settings, create schemas, or serialize/locally validate payloads; this release adds no GCP schema API. Pub/Sub schemas are not general JSON Schema. See Google’s schema documentation.

Handlers

What does the handler return value do?

• nil: the message is acknowledged (queue delete, offset commit, or explicit ack, depending on the broker).
• an error: the message is rejected and redelivered through the broker’s retry path, counted against max_retries.

There is no manual Ack() or Nack() to call in the normal path. Message still exposes both for the rare case where you must settle mid-handler; the runtime treats a duplicate settlement as success.

How does batch processing work?

Register a batch handler and enable it in the configuration. It works on every backend.

types.RegisterBatchHandler("bulk", func(ctx context.Context, msgs []types.Message) error {
    return store.WriteAll(ctx, msgs)
})

config.Set("handler", "bulk")
config.Set("enable_batch_processing", true)
config.Set("batch_size", 50)
config.Set("batch_timeout", "500ms") // flush a partial batch after this long

A nil return settles the whole batch. An error rejects every message the handler did not settle itself. For partial success in the ordinary handler path, call msg.Nack() on the messages to reject and return nil; the runtime skips the ones the handler already settled. With Stream application deduplication enabled, manual settlement is gated on durable completion: a manual Nack fails the owned batch, while already-completed duplicates remain settled.

Can I reply to a message?

If the message carries a ReplyTo() destination, publish to it through msg.Publisher():

types.RegisterHandler("rpc", func(ctx context.Context, msg types.Message) error {
    if msg.ReplyTo() == "" {
        return nil
    }
    return msg.Publisher().Publish(ctx, msg.CorrelationId(), "", msg.ReplyTo(), "text/plain", []byte("ok"))
})

Configuration

How do I pick and configure a backend?

The URL scheme selects the backend. The rest is viper keys or builder methods. Each backend page under Docs lists its keys with defaults.

// AMQP: exchange and routing key are AMQP concepts
config.Set("url", "amqp://user:pass@localhost:5672/")
config.Set("queue", "orders")
config.Set("exchange", "events")
config.Set("routing_key", "order.created")

// Kafka: consumer group
config.Set("url", "kafka://localhost:9092")
config.Set("topic", "orders")
config.Set("consumer_group", "billing")

// SQS: visibility timeout in seconds
config.Set("url", "sqs://us-east-1/orders")
config.Set("visibility_timeout", 300)

// JetStream: stream and durable consumer
config.Set("url", "jetstream://localhost:4222")
config.Set("subject", "orders.created")
config.Set("stream_name", "orders")
config.Set("consumer_name", "billing")

Can I configure from environment variables?

Yes. Viper reads them natively, or set the values yourself:

config := viper.New()
config.Set("url", os.Getenv("MQ_URL"))
config.Set("destination", os.Getenv("MQ_DESTINATION"))
config.Set("handler", "process")

How do I enable TLS?

Use the secure scheme: amqps://, rabbitmq-stream+tls://, kafkas://, natss:// or tls://, sqss://, pubsubs://, rediss://, or redisstreams://. Check each backend reference for supported CA, client-certificate, and server-name settings; these options are not uniform across adapters. skip_verify relaxes certificate verification only; it does not turn TLS off. Kafka supports PLAIN and both SCRAM mechanisms over TLS; see Kafka authentication.

What happens when the connection drops?

With auto_reconnect enabled the runtime reconnects with exponential backoff and resumes. With it disabled, Run returns an error and it is up to you to restart. During a RabbitMQ memory or disk alarm, fire-and-forget publishes are queued in order (bounded by publish_queue_size) and flushed when the broker unblocks; confirmed publishes return types.ErrConnectionBlocked immediately.

Operations

How do I check health?

Consumers and producers implement types.HealthChecker:

health, err := consumer.HealthCheck(ctx)
if err != nil || health.Status() != types.HealthStatusHealthy {
    log.Printf("unhealthy: %v %s", err, health.Message())
}

How do I shut down cleanly?

Cancel the context passed to Run. With enable_graceful_shutdown set, in-flight messages finish (bounded by graceful_shutdown_timeout) before Run returns.

My consumer receives nothing. What should I check?

1. HealthCheck reports healthy.
2. The URL scheme matches the module you imported. A scheme with no registered backend fails at NewConsumer, not silently.
3. The handler name in handler matches a registered handler, or you used WithHandler.
4. The queue, topic or subject exists, or auto_declare is enabled where the backend supports it.
5. For Kafka and JetStream, the consumer group or durable name is not stuck on an old offset.

How do I test code that uses mqutils?

Handlers are plain functions, so unit-test them directly. For consumer and producer behavior against a real broker, docker-compose.test.yml in the repository starts every supported broker locally, and the queue_testing module holds the helpers the project’s own tests use.

Project

Where do I report a bug?

GitLab issues. Merge requests are welcome; run the unit tests and, for backend changes, the integration tests for that backend.

What is the license?

MIT.

Is v1 still supported?

v2 is the current line. Import paths gained a /v2 suffix and handlers now return an error; see Upgrading from v1.