connection lifecycle
channel lifecycle
thread ownership
publisher confirm strategy
mandatory/unroutable handling
consumer ack/nack discipline
prefetch policy
serialization
metadata propagation
idempotency hooks
retry/DLQ behavior
metrics/logging/tracing
graceful shutdown
automatic recovery
topology declaration policy
test strategy

create connection
create channel
declare exchange/queue/binding
publish message
consume delivery
ack/nack/reject delivery
register return listener
use publisher confirms
handle connection/channel shutdown
recover connections/topology depending configuration

message contract
idempotency
outbox/inbox
business transaction boundary
retry classification
DLQ policy
schema compatibility
security classification
trace context
operational dashboard

Publish OrderSubmitted event
Handle PriceCalculated command
Process FulfillmentRequested task

which channel to use
how long to wait for confirms
which headers to include
how to nack poison messages
how to emit metrics

one or small number of connections per application instance
connection created on startup
connection reused across publishers/consumers
connection closed on graceful shutdown
blocked/unblocked listener attached
shutdown listener attached
metrics emitted

public final class RabbitConnectionManager implements AutoCloseable {
    private final ConnectionFactory factory;
    private volatile Connection connection;

    public RabbitConnectionManager(RabbitConfig config) {
        this.factory = new ConnectionFactory();
        this.factory.setHost(config.host());
        this.factory.setPort(config.port());
        this.factory.setVirtualHost(config.vhost());
        this.factory.setUsername(config.username());
        this.factory.setPassword(config.password());
        this.factory.setAutomaticRecoveryEnabled(true);
        this.factory.setTopologyRecoveryEnabled(false); // decide intentionally
        this.factory.setRequestedHeartbeat(config.heartbeatSeconds());
        this.factory.setConnectionTimeout(config.connectionTimeoutMs());
    }

    public synchronized Connection start() throws Exception {
        if (connection != null && connection.isOpen()) {
            return connection;
        }

        connection = factory.newConnection("quote-order-service");
        connection.addBlockedListener(new BrokerBlockedListener());
        connection.addShutdownListener(new BrokerShutdownListener());
        return connection;
    }

    @Override
    public synchronized void close() throws Exception {
        if (connection != null && connection.isOpen()) {
            connection.close();
        }
    }
}

Use meaningful connection names if supported.
Expose connection status via health/metrics.
Do not create one connection per message.
Do not create one connection per HTTP request.
Separate publisher and consumer connections if operationally useful.
Handle blocked/unblocked events.

Do not share Channel instances concurrently across publisher threads.

// Bad: shared channel used concurrently by request threads.
class UnsafePublisher {
    private final Channel sharedChannel;

    public void publish(byte[] body) throws Exception {
        sharedChannel.basicPublish("orders.exchange", "order.created", null, body);
    }
}

// Pseudocode: channel ownership is explicit.
class PublisherWorker implements Runnable {
    private final Connection connection;
    private final BlockingQueue<OutboundMessage> queue;

    public void run() {
        try (Channel channel = connection.createChannel()) {
            channel.confirmSelect();

            while (!Thread.currentThread().isInterrupted()) {
                OutboundMessage message = queue.take();
                publishAndWaitForConfirm(channel, message);
            }
        } catch (Exception ex) {
            // reconnect/restart worker according to policy
        }
    }
}

A channel should have a clear owner.

public interface MessagePublisher {
    PublishResult publish(MessageEnvelope envelope);
}

message type
schema version
payload bytes
content type
message id
correlation id
causation id
trace context
tenant id
routing key
exchange
mandatory flag
persistence flag
headers
created timestamp

confirmed
returned/unroutable
timeout
blocked
failed before publish
failed after uncertain publish

sync wait vs async confirm listener
single publish vs batch publish
confirm timeout
retry behavior
duplicate publish tolerance
outbox mark-published timing
metrics by outcome

use mandatory flag
register return listener
consider alternate exchange
alert on returned messages
store returned message state if using outbox

channel.addReturnListener(returned -> {
    metrics.increment("rabbitmq.publisher.returned", Tags.of(
        "exchange", returned.getExchange(),
        "routingKey", returned.getRoutingKey()
    ));

    log.error("RabbitMQ message returned: exchange={}, routingKey={}, replyCode={}, replyText={}",
        returned.getExchange(),
        returned.getRoutingKey(),
        returned.getReplyCode(),
        returned.getReplyText());

    // Mark outbox row as ROUTING_FAILED if message id can be correlated.
});

public interface MessageHandler<T> {
    HandlerResult handle(MessageContext context, T payload) throws Exception;
}

message id
correlation id
causation id
trace context
tenant id
message type
schema version
routing key
exchange
redelivered flag
delivery tag
x-death headers if present
received timestamp

Ack only after durable side effects are complete.

DeliverCallback callback = (consumerTag, delivery) -> {
    long tag = delivery.getEnvelope().getDeliveryTag();

    try {
        MessageEnvelope envelope = deserialize(delivery);
        ProcessingDecision decision = dispatcher.dispatch(envelope);

        if (decision.success()) {
            channel.basicAck(tag, false);
        } else if (decision.retryable()) {
            channel.basicNack(tag, false, false); // route through DLX/retry topology
        } else {
            channel.basicReject(tag, false);      // permanent failure -> DLQ/parking lot
        }
    } catch (Throwable ex) {
        // Avoid infinite immediate requeue loops unless explicitly intended.
        channel.basicNack(tag, false, false);
    }
};

channel.basicNack(tag, false, true); // immediate requeue can create a hot redelivery loop

Is this safe to retry?
Will retry change the result?
Could retry duplicate an external side effect?
Should this be visible to operations immediately?

catch (Exception ex) {
    Thread.sleep(60_000); // Bad: holds consumer thread and delivery unacked
    channel.basicNack(tag, false, true);
}

consumer rejects/nacks message without requeue
broker DLX routes to retry queue
retry queue TTL expires
message returns to main exchange/queue
x-death / retry headers track attempts
after threshold message goes to parking lot

success -> ack
retryable -> nack/reject false into retry/DLX topology
permanent -> reject false into DLQ/parking lot

exception class
error code
business state
handler version
payload hash
schema version
tenant id
first failure time
last failure time
retry count
operator note

consumer nack/reject false
RabbitMQ dead-letters message
x-death header records history

simple
standard
less app code
works with broker policies

limited business error context
harder to attach structured failure reason

consumer catches permanent failure
publishes FailureEnvelope to parking exchange
acks original only after failure publish confirmed

rich metadata
strong audit story
custom replay workflow

more code
requires confirm discipline
risk if failure publish succeeds but ack fails
must be idempotent

public interface MessageSerializer<T> {
    SerializedMessage serialize(T value, MessageSchema schema);
}

public interface MessageDeserializer<T> {
    T deserialize(byte[] body, MessageHeaders headers);
}

content type
content encoding
schema name
schema version
payload bytes
payload hash if needed

Never infer schema only from queue name.
Include message type/version.
Keep headers small.
Do not put large business objects in headers.
Avoid Java-native serialization.
Treat payload as language-neutral contract.

message_id
message_type
message_version
correlation_id
causation_id
traceparent
tenant_id if applicable
source_service
created_at
published_at
idempotency_key

actor_id
request_id
schema_name
schema_version
retry_count
business_key
aggregate_id
payload_hash

PII leakage
credential leakage
header size explosion
trace cardinality issues
tenant boundary leak

handle SIGTERM/application shutdown hook
stop accepting new work
cancel consumer subscription
wait bounded time for in-flight processing
ack successful work
nack/reject failed or incomplete work according policy
close channel/connection
emit shutdown metrics

automatic connection recovery enabled?
topology recovery enabled?
who declares exchanges/queues/bindings?
what happens during recovery window?
are publisher confirms reset correctly?
are consumers re-registered?
are duplicate consumers possible?
are metrics emitted?

Should application declare exchange/queue/binding?
Should platform GitOps declare them?
Should declaration be passive in production?
Should mismatch fail fast?
Should queue arguments be mutable?
How are policy/operator policy applied?

critical topology declared as code
application may passively verify existence
startup fails if required topology missing
queue arguments are not silently changed by app
breaking topology changes go through review

// Pseudocode only.
channel.exchangeDeclarePassive("orders.command.x");
channel.queueDeclarePassive("quote-order.pricing.work.q");

publish_attempt_total
publish_confirmed_total
publish_failed_total
publish_returned_total
publish_timeout_total
publish_latency_ms
publisher_confirm_latency_ms
connection_blocked_total
connection_recovered_total
outbox_backlog

message_received_total
message_processed_total
message_failed_total
message_ack_total
message_nack_total
message_reject_total
message_redelivered_total
consumer_processing_latency_ms
consumer_db_latency_ms
consumer_external_call_latency_ms
inflight_messages

service
vhost
exchange
queue
message_type
result
failure_class

message_id
tenant_id if too many tenants
correlation_id
raw routing key if unbounded
exception message

message_id
message_type
correlation_id
causation_id
exchange
routing_key
outbox_id if present
confirm_result
returned_flag
latency_ms

message_id
message_type
correlation_id
queue
routing_key
redelivered
retry_count
x_death_count
handler_result
processing_latency_ms
failure_class

PII exposure
secret leakage
huge log volume
compliance violation
customer data leakage through DLQ debugging

extract trace context from incoming HTTP request
inject trace context into message headers
extract trace context in consumer
create consumer span linked to producer span
include messaging system/exchange/queue/routing key attributes

Can the app open a TCP connection to RabbitMQ?

can connection be established?
is channel open?
is broker blocked?
are required exchanges/queues present?
is outbox backlog within SLA?
are consumers registered?
is DLQ growth under threshold?

Do not make readiness flap because one optional queue is delayed.
Do not declare topology from health endpoint.
Do not publish test messages to production queues casually.

liveness: is this process alive?
readiness: can it safely receive traffic?
dependency health: is RabbitMQ usable for this function?
business health: are message flows meeting SLA?

routing key selection
header construction
message envelope validation
error classification
retry decision
ack/nack decision
idempotency key extraction
serialization/deserialization

OrderSubmitted event has correct message_type/version
missing tenant id fails validation
permanent schema error maps to DLQ decision
transient DB timeout maps to retry decision
redelivered poison message after threshold maps to parking lot

exchange routing
binding correctness
mandatory return behavior
publisher confirm behavior
manual ack/redelivery
nack/reject/DLX
TTL retry
consumer prefetch behavior
connection recovery basics

// Pseudocode only.
@Container
static RabbitMQContainer rabbit = new RabbitMQContainer("rabbitmq:management")
    .withExposedPorts(5672, 15672);

publish to command exchange routes to expected queue
publish with wrong routing key is returned when mandatory=true
consumer failure sends message to DLQ
message is redelivered after consumer closes without ack
retry queue returns message after TTL
publisher does not mark outbox row until confirm

publisher confirm success
unroutable message failure
consumer ack after DB commit
consumer crash before ack redelivers
retry path
DLQ path
idempotent duplicate handling
trace/correlation header propagation

creating connection per message
creating connection per HTTP request
sharing channel across publishing threads
ack before DB commit
auto ack for business-critical messages
sleeping inside consumer for retry delay
using requeue true for all exceptions
no publisher confirms
no mandatory flag for critical routing
no return listener
no blocked listener
no graceful shutdown
no message id/correlation id
payload logging in production
topology created ad hoc by every service
mock-only tests
no DLQ replay safety

Is there an internal RabbitMQ library/wrapper?
Which RabbitMQ Java client version is used?
Is Spring AMQP used, raw Java client, or custom abstraction?
Who owns the client configuration standard?
Are connection/channel patterns documented?

connections per pod
channels per connection
channel ownership model
publisher thread model
consumer channel model
shutdown handling
blocked/unblocked listener
shutdown listener
heartbeat/timeout settings

publisher confirms
mandatory flag
return listener
alternate exchange
confirm timeout
outbox integration
retry/backoff
metrics/logging/tracing

manual ack
ack after DB commit
nack/reject policy
prefetch
consumer concurrency
idempotency/inbox
DLQ/retry topology
poison message handling
graceful shutdown

message format
content type
schema version
message id
correlation id
causation id
traceparent
tenant id
payload logging policy

unit tests for routing/header/error policy
contract tests
Testcontainers integration tests
consumer crash test
broker restart test
retry/DLQ test
idempotency duplicate test

Who owns this connection?
Who owns this channel?
Can this channel be used concurrently?
Is publisher confirm required and used?
Can unroutable messages be detected?
What happens when broker blocks publishing?
Is there a timeout?
Is publish inside HTTP request path?
Should outbox be used instead?
Does consumer use manual ack?
Is ack after durable processing?
What happens on exception?
Can this create immediate requeue loop?
Is idempotency enforced?
Are headers standardized?
Are PII payloads logged?
Does shutdown drain in-flight messages?
Are metrics enough for incident debugging?
Is topology declared intentionally?
Are Testcontainers tests covering RabbitMQ behavior?

Do not scatter raw RabbitMQ client usage through business code.
Connections are long-lived; channels need clear ownership.
Do not share publishing channels across threads.
Publisher confirms and mandatory returns are core reliability tools.
Manual ack should be centralized and tied to durable side effects.
Retry classification should be explicit.
Do not sleep inside consumers for delayed retry.
Graceful shutdown is part of correctness.
Automatic recovery is useful but not a correctness guarantee.
Topology declaration must have an ownership model.
Metrics, logs, traces, and tests are part of the client design.