Docs / Upgrading from v1

Upgrading from v1

What changed in v2: /v2 import paths, error-returning handlers, and the shared consumer runtime

This page is rendered from UPGRADING.md in the repository.

v2 is a breaking release. The import paths, the handler contract, and several transport behaviors changed. This guide covers everything a v1 consumer needs.

Import paths

Every module gained the Go-required /v2 suffix:

Handlers return an error (the headline change)

// v1: handlers acked/nacked manually and returned nothing.
func handler(ctx context.Context, msg types.Message) {
    if err := process(ctx, msg); err != nil {
        _ = msg.Nack()
        return
    }
    _ = msg.Ack()
}

// v2: return nil to acknowledge, an error to reject. The shared consumer
// runtime settles the message (with retry) based on the return value.
func handler(ctx context.Context, msg types.Message) error {
    return process(ctx, msg)
}

• Manual msg.Ack()/msg.Nack() inside a handler still works; the runtime treats the duplicate settlement (types.ErrAlreadyAcknowledged) as success.
• A handler that returns nil without acking is now acknowledged by the runtime; return an error when processing did not succeed.
• Batch handlers (BatchHandlerFunc) follow the same contract: nil settles the whole batch, an error rejects every message the handler did not settle itself.
• Double ack/nack on a message now uniformly returns types.ErrAlreadyAcknowledged on every backend (some previously returned nil).

Behavior changes to plan for

• Connection loss with auto_reconnect: false returns an error from Run instead of idling forever. Enable auto_reconnect if you relied on the old hang.
• Kafka has real at-least-once semantics. Acks commit offsets (the old transport committed everything on receipt regardless of handler outcome). Nacks republish to retry_topic with a retry_count header and land on dead_letter_topic once retry_max_retries is exhausted; without a retry_topic failures are retried locally with backoff. Handlers must be idempotent — rebalances can redeliver in-flight messages.
• Kafka skip_verify no longer disables TLS. TLS turns on via kafkas:// or tls_enabled: true; skip_verify only relaxes certificate verification.
• NATS no longer drops messages under backpressure — delivery blocks (context-aware) when the internal channel is full, matching every other backend.
• Batch processing works on every backend (enable_batch_processing, batch_size, batch_timeout), not just AMQP. batch_timeout is a real duration string ("250ms"); the v1 AMQP implementation misapplied it by a factor of a million and never flushed partial batches.
• Previously dead configuration is now live. If you set any of these in v1 they silently did nothing and WILL take effect after upgrading: AMQP channel_pool_size/mandatory/producer durable+auto_delete; NATS max_reconnects/reconnect_wait/queue_group/consumer_name; SQS dead_letter_queue_url (applies a RedrivePolicy), consumer tuning and FIFO group/dedup fields; GCP dead_letter_topic+max_delivery_attempts (DeadLetterPolicy — requires IAM grants to the Pub/Sub service account), publish batching thresholds, subscription tuning; Redis auth/pool/timeout fields, stream trimming, and claim_idle_time/pending_message_max_age (XAutoClaim pending recovery, Redis ≥ 6.2).

Removed API

• amqp producer immediate config key (RabbitMQ ≥ 3.0 rejects the flag).
• aws consumer message_retention_days (was never applied; honoring it would clobber externally provisioned retention).
• GCP skip_verify config keys (meaningless for the gRPC client; the emulator path uses PUBSUB_EMULATOR_HOST).
• Exported AcknowledgeMessage/NacknowledgeMessage helpers in the kafka and nats packages — use runner.AckWithRetry/runner.NackWithRetry.
• Transport.DeclareExchange/Transport.BindQueue now take a context.Context, and BindQueue’s parameters are exchange-first with a durable flag (matching what every implementation always did).

New in v2 (non-breaking once you’re on the new paths)

• Typed builders: mqutils.NewConsumerBuilder() / NewProducerBuilder() configure consumers without viper and take handler funcs directly.
• Canonical config keys accepted by every backend alongside native keys: destination, max_retries, consumer_group.
• Symmetric URL schemes: consumers and producers register identical scheme sets (sqss://, gcp://, redisstreams://, … all work on both sides now).
• runner package: the shared consumer loop (ack-retry helpers, batching, graceful shutdown, reconnect backoff, retry-budget drops with a dead-letter hook) is exported for building custom backends.
• AMQP blocked-connection handling: when RabbitMQ raises a memory/disk alarm (connection.blocked), fire-and-forget publishes are accepted into a bounded internal queue (publish_queue_size, default 1000; overflow returns types.ErrPublishQueueFull) and flushed in order on unblock — v1 wrote into the stalled socket and hung. New publisher-confirms mode (publisher_confirms: true, confirm_timeout): every publish waits for the broker’s ack (types.ErrPublishNacked on nack) and publishes during a block are rejected immediately with types.ErrConnectionBlocked for the caller to handle. NATS gains reconnect_buf_size to bound the client buffer that already queues publishes while disconnected.

Reliability update for existing v2 deployments

RabbitMQ queue expiration is opt-in

transient_queue_expires is now disabled when omitted or zero. A positive integer enables automatic x-expires in milliseconds for transient queues that RabbitMQ 4.3+ requires the library to declare durable. The durability workaround remains in effect. Intentionally durable queues do not acquire an expiration automatically.

Existing queues without x-expires can be reused with the default setting without recreating them, provided all other declaration arguments still match. For queues created by the previous 30-minute default, retain their argument explicitly:

transient: true
transient_queue_expires: 1800000

Queue arguments must match on redeclaration. Removing or changing an existing x-expires can still cause PRECONDITION_FAILED; the library never deletes queues to resolve this. With automatic expiry disabled, upgraded durable queues can remain after a consumer stops. Use an explicitly chosen expiry or an operator-managed lifecycle when cleanup is required.

Retry queues have a separate lifecycle because they have no consumers: publishing into them does not renew RabbitMQ’s inactivity lease. They no longer inherit transient_queue_expires. If a retry queue was already created with expiry, retain its matching argument using the new setting:

auto_declare: true
retry_queue_expires: 1800000

Positive retry_queue_expires enables periodic lease renewal for that retry queue while its consumer session is active. It requires automatic declaration; renewal stops with the session and failures are reported through its connection error path. Omitted/zero leaves retry expiry unset. These settings accept positive millisecond values or zero; invalid/overflowing values fail validation.

Omitting retry_queue_ttl now applies its documented 1000 ms message retry delay. Previously such a retry queue could trap messages indefinitely. An existing retry queue that lacks x-message-ttl still requires a deliberate broker/configuration migration; this update does not recreate it. Large valid TTLs are encoded without narrowing to a 32-bit integer.

Handler concurrency and shutdown

All consumers default max_concurrent_handlers to the effective message_channel_buffer value, independent of whether graceful shutdown is enabled. An omitted or zero buffer resolves to its existing default of 10; a configured buffer of 100 therefore permits up to 100 active handlers or batches without a separate concurrency setting. Set an explicit positive limit to override the buffer-derived default, for example:

message_channel_buffer: 100
max_concurrent_handlers: 8
enable_graceful_shutdown: true
graceful_shutdown_timeout: 30

The typed consumer builder’s WithPrefetch(100) also supplies the buffer used for this default; WithConcurrency(8) overrides it. Each batch counts as one active unit. Explicit zero or negative concurrency is invalid. GCP accepts concurrency up to 10000, matching its buffer ceiling, and the setting bounds actual handler work rather than the number of StreamingPull streams.

The standalone shared runner has no buffer configuration and retains its own default of one active handler; backend consumers pass their resolved limit through WithConcurrency.

The graceful timeout applies to cancellation, deliberate transport closure, and terminal errors. Handlers retain a live context during the drain; that context is canceled when the drain completes or expires. Code that ignores context cancellation cannot be forcibly stopped by Go. Without graceful shutdown, handler contexts are canceled immediately; releasing a partial unhandled batch has a separate bounded one-second cleanup budget.

Delivery and configuration corrections

• Kafka commits only contiguous settled offsets within each partition. Failed records without a retry topic are retried locally with backoff; they cannot be skipped by a later record’s acknowledgment. A terminal acknowledgment or dead-letter publishing failure returns an error from Run, cancels the subscription, and leaves unresolved originals for recovery. Restart after resolving the failure; handlers must remain idempotent.
• NATS JetStream max_deliver: 1 permits the first attempt. Its budget counts total deliveries, including the initial one. Authenticated NATS URLs retain their credentials when passed to the client.
• Explicit Redis stream_max_len: 0 disables trimming and pending_message_max_age_seconds: 0 disables stale-pending drops. Omitted values retain their defaults. Pending recovery pages past locally active entries so later abandoned records can be recovered.
• GCP startup preserves existing subscription settings when they are omitted. Explicit changes are applied selectively and failures are returned. GCP’s default delivery/retry budget is five; the older documentation’s value of 50 was incorrect. SDK receive callbacks remain active through settlement, and publisher handles are reused and stopped during cleanup.
• Typed builder values override both canonical and native keys from WithConfig. Builds snapshot evaluated configuration and no longer modify the caller’s Viper instance. Direct Viper construction retains native-key precedence when both native and canonical keys are supplied.
• Unknown-backend factory errors and logs remove URL credentials and query data. Applications should avoid independently logging raw connection URLs.

Explicit producer cleanup

types.Publisher now embeds io.Closer; types.Producer inherits Close() error. Call producer.Close() directly without a type assertion. Custom publisher and producer implementations must provide Close() error. This update remains in the v2 release line. Close producers when their owning component stops. GCP aborts unconfirmed work by closing its RPC clients, then stops cached topic publishers; publishes that already returned confirmed success remain complete. Close can interrupt pending requests with an error, and a failed/canceled publish may already have reached the broker. AMQP releases the producer’s own resources while retaining shared connection-pool ownership.

Kafka’s synchronous client cannot cancel an individual in-flight send. Canceling an active publish retires that connection, interrupts its sockets, and returns an error indicating unknown delivery. Call Start again before publishing more messages; an already-canceled context supplied before a send leaves the connection usable. Closing a producer also interrupts stalled sends. NATS publish and request/reply operations honor caller cancellation and transport closure without requiring a connection restart after ordinary caller cancellation.

producer, err := mqutils.NewProducer(ctx, config)
if err != nil {
    return err
}
defer producer.Close()
if err := producer.Start(ctx); err != nil {
    return err
}

Kafka SASL and transactions

SASL is opt-in through sasl_enabled, sasl_mechanism, sasl_username, and sasl_password. PLAIN and both SCRAM-SHA-256/SCRAM-SHA-512 mechanisms require TLS unless the explicit development-only sasl_allow_insecure override is enabled. Existing unauthenticated configurations are unchanged. Credentials also apply to transport-owned retry/dead-letter and administrative clients.

Atomic producer writes use kafka.NewTransactionalProducerBuilder, whose Build connects the producer. The ordinary producer rejects transaction_id and transaction_timeout; use the transactional API when those settings are present. Only the callback-scoped publisher can send inside InTransaction. Scoped Close aborts without closing the parent; callbacks returning nil after Close still receive ErrTransactionClosed. A retained scoped publisher cannot affect a later transaction. Never automatically replay ErrTransactionOutcomeUnknown: reconcile the earlier outcome first. Failed/fenced/uncertain producers require explicit restart or replacement.

Consumer isolation_level defaults to read_uncommitted; opt into read_committed to exclude aborted records. Producer transactions do not include consumer-offset commits, retry-topic writes from consumption, or external database side effects. See the Kafka guide.

RabbitMQ Stream backend and import availability

The new modules are go.digitalxero.dev/mq-rmqstream/v2 and the optional go.digitalxero.dev/mq-rmqstream-dedupredis/v2. They require Go 1.25 and keep /v2 module paths. The optional adapter does not add Redis to the core Stream module. Their vanity mappings are prepared on the supporting go-pkgs branch, which must be delivered before the new imports resolve publicly; then the new mqutils modules must be released. Existing backend import paths remain available. Use the canonical-checkout/local-replacement instructions to evaluate the changes before those prerequisites are complete.

Native rabbitmq-stream:// and rabbitmq-stream+tls:// endpoints are independent of AMQP. Stream Ack stores progress; it does not delete a record. Retention and reconnect can cause replay. Durable consumer identity is tied to its filter contract, recorded in broker metadata topology even when application filtering is not enabled. Consumers therefore require topology declaration/discovery permission; keep metadata until retiring the identity. Change consumer identity when changing the filter contract.

Broker sequence deduplication requires exclusive application ownership of each producer-name/physical-stream pair and a durable publication ledger. Preserve the original sequence, partition, and immutable event on replay. Application MessageId deduplication requires an injected atomic claim store and explicit retention. Expired claims are fenced by ownership token, manual Ack waits for durable completion, and store failures leave unresolved source work replayable. Redis restart durability depends on persistence configuration. A transactional inbox is still required to make database effects atomic with their completion record. The Stream guide and runnable outbox/inbox example explain the boundaries.

GCP Pub/Sub schema enforcement was assessed but no provisioning or encoding API was added. Existing raw-byte publication can use a topic whose Avro/Protobuf schema is configured externally; Google enforces it at publish time. There is no new local validation, schema administration, or general JSON Schema support.