Frequently Asked Questions

General Questions

Q: Why should I use mqutils instead of individual SDKs?

A: mqutils provides a unified API across 6 major message queue systems, eliminating the need to learn and maintain different APIs for each system. Key benefits include:

• Single learning curve - Learn once, use everywhere
• Consistent error handling - Structured validation across all systems
• Built-in production features - Health monitoring, batch processing, graceful shutdown
• Easy migration - Switch between systems without code changes
• Reduced complexity - One library instead of 6+ different SDKs

Q: Is mqutils production-ready?

A: Yes. mqutils is designed for production use with:

• Comprehensive error handling and validation
• Retry budgets and dead-lettering where the broker supports them (AMQP DLX, Kafka retry/DLT, SQS RedrivePolicy, GCP DeadLetterPolicy, Redis Streams drop). NATS Core and Redis Pub/Sub are at-most-once and cannot dead-letter.
• Built-in health monitoring for all systems
• Graceful shutdown when enable_graceful_shutdown is set
• Thread-safe operations with proper synchronization
• Unit tests, broker-backed integration tests, and a public-API acceptance suite (acceptance/)

Q: What message queue systems are supported?

A: mqutils supports 6 major message queue systems:

1. AMQP/RabbitMQ - Enterprise messaging with exchanges and routing
2. Apache Kafka - Event streaming with consumer groups and partitions
3. NATS Core/JetStream - Cloud-native messaging with clustering
4. AWS SQS - Cloud-managed queues with FIFO support
5. GCP Pub/Sub - Google Cloud messaging with exactly-once delivery
6. Redis Pub/Sub & Streams - In-memory messaging with persistence

Q: Can I migrate between different message queue systems?

A: Yes! This is one of mqutils’ key strengths. Migration typically requires only changing the connection URL and destination:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

// From RabbitMQ
config.Set("url", "amqp://localhost:5672/")
config.Set("destination", "queue") // canonical key, works on every backend

// To Kafka (same code, different URL)
config.Set("url", "kafka://localhost:9092")
config.Set("destination", "topic")

// To AWS SQS
config.Set("url", "sqs://us-east-1/queue-name")

consumer, err := mqutils.NewConsumer(ctx, config)

Your message handlers and application logic remain unchanged.

Performance Questions

Q: What kind of performance can I expect?

A: mqutils is optimized for high performance (see the benchmarks/ module to measure on your own hardware):

• Configurable buffers (message_channel_buffer)
• AMQP channel pooling (channel_pool_size) for resource efficiency
• Batch processing on every backend with configurable size and flush timeout

Q: How does batch processing work?

A: Batch processing allows you to process multiple messages together for higher throughput, and works on every backend:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

// Register the batch handler globally
types.RegisterBatchHandler("batch_processor", func(ctx context.Context, msgs []types.Message) error {
    log.Printf("Processing batch of %d messages", len(msgs))
    
    for _, msg := range msgs {
        // Process each message
    }
    
    return nil // All messages acknowledged on success
})

// Enable it in the consumer configuration
config.Set("enable_batch_processing", true)
config.Set("batch_size", 50)
config.Set("batch_timeout", "500ms") // duration string
config.Set("handler", "batch_processor")

Q: Is mqutils thread-safe?

A: Yes, mqutils is designed with thread safety in mind:

• Atomic operations for critical sections
• Proper synchronization mechanisms
• Race condition prevention
• Safe concurrent access to shared resources

Configuration Questions

Q: How do I configure different message queue systems?

A: The URL scheme selects the backend automatically; the rest of the configuration is passed via viper (or the typed builder). Every backend also accepts the canonical keys destination, max_retries, and consumer_group:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

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

// Kafka with consumer group
config.Set("url", "kafka://localhost:9092")
config.Set("topic", "topic")
config.Set("consumer_group", "my-service")

// AWS SQS with visibility timeout
config.Set("url", "sqs://us-east-1/my-queue")
config.Set("visibility_timeout", 300)

// NATS JetStream with durable consumer
config.Set("url", "jetstream://localhost:4222")
config.Set("subject", "orders.created")
config.Set("use_jetstream", true)
config.Set("stream_name", "orders")
config.Set("consumer_name", "processor")
config.Set("durable", true)

Q: Can I use environment variables for configuration?

A: Yes, you can use environment variables or any configuration management system (viper supports both natively):

1
2
3
4
5

config := viper.New()
config.Set("url", os.Getenv("MESSAGE_QUEUE_URL"))
config.Set("destination", os.Getenv("MESSAGE_QUEUE_DESTINATION"))
config.Set("handler", "process")
consumer, err := mqutils.NewConsumer(ctx, config)

Q: How do I configure SSL/TLS connections?

A: Use the secure URL schemes:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

// AMQP with TLS
"amqps://user:pass@secure-host:5671/queue"

// Kafka with TLS (SASL is not supported)
"kafkas://secure-broker:9093/topic"

// NATS with TLS
"natss://secure-nats:4222/subject"

// Redis with TLS
"rediss://secure-redis:6380/channel"

AMQP (amqps://) also supports sni_hostname when the TLS ServerName differs from the dial host, and tls_cert/tls_key (plus optional tls_ca) for mTLS. skip_verify only relaxes certificate verification; it does not disable TLS or SNI.

Troubleshooting

Q: My consumer isn’t receiving messages. What should I check?

A: Follow this troubleshooting checklist:

1. Verify connection - Check if the consumer can connect to the message broker

   1 2 3 4

   health, err := consumer.HealthCheck(ctx) if err == nil { log.Printf("Health: %s - %s", health.Status(), health.Message()) }

2. Check URL format - Ensure the connection URL is correct for your system

3. Verify handler registration - Make sure your handler is registered before calling Run()

4. Check queue/topic exists - Ensure the destination exists in your message broker

5. Review logs - Look for connection errors or authentication issues

Q: How do I handle connection failures?

A: mqutils provides built-in error handling and reconnection. Set auto_reconnect: true to reconnect with backoff on connection loss (with it disabled, Run returns an error):

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

// Built-in health monitoring
go func() {
    ticker := time.NewTicker(30 * time.Second)
    for range ticker.C {
        health, err := consumer.HealthCheck(ctx)
        if err != nil || health.Status() != types.HealthStatusHealthy {
            log.Printf("Connection issue: %v", err)
            // Implement alerting here
        }
    }
}()

// Graceful shutdown handling
ctx, cancel := context.WithCancel(context.Background())
defer cancel()

// Run with context for proper shutdown
if err := consumer.Run(ctx); err != nil {
    log.Printf("Consumer error: %v", err)
}

Q: Messages are being processed slowly. How can I improve performance?

A: Try these performance optimizations:

1. Enable batch processing for higher throughput
2. Increase message buffer size (message_channel_buffer)
3. Use appropriate batch sizes for your use case
4. Configure proper timeouts for your message broker
5. Monitor resource usage (CPU, memory, network)

1
2
3
4
5

// Example optimization
types.RegisterBatchHandler("optimized", func(ctx context.Context, msgs []types.Message) error {
    // Process messages in batches for better performance
    return processBatch(ctx, msgs)
})

Q: How do I debug message processing issues?

A: Use comprehensive logging and error handling:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

types.RegisterHandler("debug_handler", func(ctx context.Context, msg types.Message) error {
    log.Printf("Processing message ID: %s", msg.MessageId())
    log.Printf("Headers: %+v", msg.Headers())
    log.Printf("Body: %s", string(msg.Body()))
    
    if err := processMessage(ctx, msg); err != nil {
        log.Printf("Processing failed: %v", err)
        return err // Will trigger retry or dead letter
    }
    
    log.Printf("Message processed successfully")
    return nil
})

Integration Questions

Q: Can I use mqutils with Docker?

A: Yes, mqutils works perfectly in containerized environments:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

FROM golang:1.24-alpine AS builder
WORKDIR /app
COPY go.mod go.sum ./
RUN go mod download
COPY . .
RUN go build -o app ./cmd/consumer

FROM alpine:latest
RUN apk --no-cache add ca-certificates
WORKDIR /root/
COPY --from=builder /app/app .
CMD ["./app"]

Q: How do I test applications using mqutils?

A: mqutils supports comprehensive testing (see the queue_testing module for fakes and helpers):

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

func TestMessageProcessing(t *testing.T) {
    var processedMsg types.Message
    types.RegisterHandler("test", func(ctx context.Context, msg types.Message) error {
        processedMsg = msg
        return nil
    })

    // Use test containers or a broker from docker-compose.test.yml
    config := viper.New()
    config.Set("url", "amqp://test-rabbit:5672/")
    config.Set("queue", "test-queue")
    config.Set("handler", "test")

    consumer, err := mqutils.NewConsumer(context.Background(), config)
    require.NoError(t, err)
    _ = consumer

    // Test your message processing logic
}

Q: Can I use mqutils with Kubernetes?

A: Absolutely! mqutils is cloud-native friendly:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

apiVersion: apps/v1
kind: Deployment
metadata:
  name: mqutils-consumer
spec:
  replicas: 3
  selector:
    matchLabels:
      app: mqutils-consumer
  template:
    metadata:
      labels:
        app: mqutils-consumer
    spec:
      containers:
      - name: consumer
        image: myapp:latest
        env:
        - name: MESSAGE_QUEUE_URL
          value: "kafka://kafka-cluster:9092/events"
        resources:
          requests:
            memory: "128Mi"
            cpu: "100m"
          limits:
            memory: "512Mi"
            cpu: "500m"

Support and Community

Q: Where can I get help?

A: Multiple support channels are available:

• GitLab Issues: Report bugs and request features
• Documentation: Comprehensive guides and examples
• API Reference: Complete interface documentation

Q: How can I contribute to mqutils?

A: We welcome contributions! Please:

1. Fork the repository
2. Create a feature branch
3. Add tests for your changes
4. Ensure all tests pass
5. Submit a merge request

See our contributing guidelines for details.

Q: Is mqutils open source?

A: Yes! mqutils is licensed under the MIT License, making it free for both personal and commercial use with no vendor lock-in.

Still Have Questions?

If you can’t find the answer you’re looking for, please:

• Open an issue on GitLab
• Check our examples directory for more use cases
• Review the API documentation for detailed interface information