Messaging

Messaging Guidelines

MANDATORY FOR ALL AI AGENTS: Every event publish MUST go through the transactional outbox pattern — never direct broker calls inline with business writes. Every consumer MUST be idempotent. Topology (exchanges, queues, topics, partitions) MUST be externalized — no @Bean Queue/Exchange/Binding/NewTopic in microservice code. Message envelopes MUST follow the canonical EventMessage<T> shape (Header + Body). Retries follow exponential backoff bounded by maxAttempts; permanent failures land in DLT/DLQ.

Project-agnostic: code samples use placeholder package com.example.platform, service name platform-service. Replace with your project’s actual values. The canonical reference implementation for the outbox pattern is codehuntersio-sdk-event-messaging — treat it as the source of truth for the JVM outbox shape and adapt configuration as needed.

Companion guides:

• hexagonal-architecture.md — ports/adapters for publishers and consumers, outbox + chain of responsibility.
• adr.md — capture broker choice (RabbitMQ vs Kafka vs SQS/SNS), topology, schema evolution policy.
• security.md — message signing, in-transit encryption, mTLS to broker, never log payloads with PII.
• observability.md — metrics + tracing propagation (W3C traceparent in headers).
• liquibase.md — outbox + DLQ schema migrations follow expand/contract.
• arch-doc-spec.md §Event Catalog — board-grade event inventory.
• api-design.md §21 — webhook signing (related but distinct).

Companion commands/agents:

• /arch-outbox → arch-outbox-verifier (audits outbox impl correctness).
• /arch-saga → arch-saga-verifier (audits saga choreography/orchestration).
• arch-outbox-pattern skill — canonical write-side pattern.
• arch-saga-pattern skill — distributed transactions across services.
• kafka-rabbitmq skill — broker-specific patterns.

Table of Contents

1. Principles
2. Broker Selection Matrix
3. Canonical Event Envelope
4. Transactional Outbox Pattern
5. Claim-Based Dispatch
6. Retry, Backoff & Dead-Letter
7. Idempotent Consumers
8. Schema Evolution & Versioning
9. Topology Externalization
10. RabbitMQ Specifics
11. Kafka Specifics
12. Choreography vs Orchestration
13. Tracing Propagation
14. Hexagonal Placement
15. Testing
16. Operational Metrics & Alerting
17. Anti-patterns
18. Pre-Merge Checklist

1. Principles

• Reliability over freshness. Better to publish 200ms late atomically than instantly with data-loss risk.
• At-least-once delivery is the default. Exactly-once is rare, expensive, and usually a consumer-idempotency problem dressed up.
• The producer commits the business write AND the outbox row in one local transaction. No separate “send” call inline with the business change.
• Brokers are dumb pipes. Business logic lives in services; topology lives in broker config; transformations live in adapters.
• Schemas are contracts. Treat them with API-versioning rigor (see api-design.md §5).
• Consumers MUST be idempotent. Duplicates always happen — partition rebalance, retry on network blip, restart.
• Origin scoping: each microservice claims and dispatches only the outbox rows it produced. Cross-service dispatch is an anti-pattern.

2. Broker Selection Matrix

Decision is an ADR. Don’t mix brokers in one bounded context without a documented reason.

3. Canonical Event Envelope

Every cross-service event uses the same envelope. Reference shape, language-agnostic:

1
EventMessage<T>
2
├── Header
3
│   ├── eventId       // stable unique id of the physical event (UUID, generated in the factory)
4
│   ├── traceId       // distributed tracing correlation
5
│   ├── version       // envelope version, e.g. "1.0.0"
6
│   ├── schemaVersion // payload schema version, e.g. "1.0.0"
7
│   ├── origin        // producing service (spring.application.name)
8
│   ├── timestamp     // ISO-8601 instant the event was created
9
│   └── locale        // optional, e.g. "es", "en-US"
10
└── Body<T>
11
    ├── aggregate     // aggregate identifier (e.g., orderId, paymentId)
12
    ├── eventName     // event name in past tense (OrderCreated, PaymentFailed)
13
    ├── eventType     // EVENT | COMMAND
14
    └── payload       // T — typed business payload

3.1 Java reference

1
public class EventMessage<T> {
2
    private Header header;
3
    private Body<T> body;
4

5
    public static class Header {
6
        private String eventId; // stable unique id of the physical event (UUID), set in the factory
7
        private String traceId;
8
        private String version = "1.0.0";
9
        private String schemaVersion = "1.0.0";
10
        private String origin;
11
        private String timestamp; // Instant.now().toString()
12
        private String locale;
13
    }
14

15
    public static class Body<T> {
16
        private String aggregate;
17
        private String eventName;
18
        private String eventType;
19
        private T payload;
20
    }
21
}

3.2 Construction via factory

EventMessage is always built via a factory that injects origin from spring.application.name. Caller never sets it manually — prevents copy-paste bugs across services.

1
@Component
2
public class OrderEventPublisher {
3
    private final OutboxEventFactory factory;
4
    private final OutboxRepositoryPort outbox;
5

6
    public void publishOrderCreated(Order order, String traceId) {
7
        var msg = factory.create(
8
            "platform.notifications",       // topic / exchange
9
            traceId,
10
            order.getId().toString(),       // aggregate
11
            "OrderCreated",                  // eventName
12
            EventTypes.EVENT,                // eventType
13
            new OrderCreatedPayload(order.getId(), order.getAmount())
14
        );
15
        outbox.save(msg);
16
    }
17
}

3.3 Field conventions

4. Transactional Outbox Pattern

4.1 Why

Dual-write problem: writing to the DB AND the broker in two separate operations cannot be atomic — one may succeed, the other fail. Result: data corruption (DB committed, event lost) OR phantom events (event sent, DB rolled back).

Outbox solves this: write the business state AND a row in the outbox_message table in one local transaction. A separate dispatcher reads the outbox and publishes to the broker with at-least-once semantics. If the dispatcher fails, retry. If the broker drops the message, retry. The DB row is the source of truth.

4.2 Schema (relational reference)

1
CREATE TABLE outbox_message (
2
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
3
    origin           VARCHAR(255) NOT NULL,    -- producing service (spring.application.name)
4
    routing_key      VARCHAR(255),             -- broker routing key (nullable for fanout)
5
    topic            VARCHAR(255) NOT NULL,    -- exchange/topic name
6
    message_key      VARCHAR(255),             -- aggregate id (Kafka partition key)
7
    payload          TEXT         NOT NULL,    -- JSON-serialized EventMessage<T>
8
    status           VARCHAR(50)  NOT NULL DEFAULT 'PENDING',  -- PENDING | IN_FLIGHT | PUBLISHED | FAILED | ARCHIVED
9
    attempts         INT          NOT NULL DEFAULT 0,
10
    error            TEXT,
11
    occurred_at      TIMESTAMP    NOT NULL DEFAULT NOW(),
12
    next_attempt_at  TIMESTAMP,
13
    published_at     TIMESTAMP,
14
    claimed_at       TIMESTAMP,                -- for stale-claim recovery
15
    claimed_by       VARCHAR(255)              -- instance marker: appName@processId
16
);
17

18
CREATE INDEX idx_outbox_origin_status_occurred
19
    ON outbox_message(origin, status, occurred_at ASC);
20

21
CREATE INDEX idx_outbox_origin_next_attempt
22
    ON outbox_message(origin, status, next_attempt_at);
23

24
-- Partial index on stale-claim recovery (Postgres / SQL Server filtered index)
25
CREATE INDEX idx_outbox_claim_recovery
26
    ON outbox_message(status, claimed_at)
27
    WHERE status = 'IN_FLIGHT';

4.3 Hexagonal port contract

1
// application/port/outbound — port belongs to application, depends only on domain
2
public interface OutboxRepositoryPort {
3
    <T> void save(OutboxMessage<T> message);
4

5
    List<OutboxMessage<String>> claimPendingMessages(
6
        String origin, int batchSize, String claimedBy, Duration claimTtl);
7

8
    void update(OutboxMessage<String> message);
9
}

4.3.1 SDK starter (codehuntersio-sdk-event-messaging)

Reference implementation that ships every contract in this section as a Spring Boot auto-configuration. Use the starter; do NOT re-implement these primitives per service.

Auto-wired beans (each @ConditionalOnMissingBean — override any to customize):

Properties (prefix outbox.):

1
outbox:
2
  enabled: true                                     # default true
3
  batch-size: 50                                    # rows per dispatcher tick
4
  max-attempts: 5                                   # then status=FAILED
5
  task:
6
    fixed-delay: 5000                               # ms between ticks
7
    lock-at-most: 60000
8
    lock-at-least: 10000
9
    lock-name: outbox-${spring.application.name}    # per-service lock — DO NOT collapse to a global name
10
  claim:
11
    ttl-ms: 300000                                  # stale-claim recovery window (5 min)
12
  backoff:
13
    base-delay-ms: 30000                            # 30s, doubled per retry
14
    max-delay-ms: 3600000                           # capped at 1h

Swapping RabbitMQ → Kafka. Provide a KafkaMessagePublisherAdapter implements MessagePublisherPort bean — the SDK skips its Rabbit default via @ConditionalOnMissingBean. Nothing else changes; the use case, the scheduler, the JDBC outbox stay identical.

Hard rule. The starter exposes ONLY OutboxRepositoryPort, OutboxEventFactory (and EventMessageFactory) to application code. Use cases inject those. Never inject MessagePublisherPort into a domain use case — it is the dispatcher’s port, not the producer’s.

4.4 Producer call site

1
@Service
2
@RequiredArgsConstructor
3
public class CreateOrderUseCase implements CreateOrderPort {
4

5
    private final OrderRepository orderRepo;
6
    private final OutboxRepositoryPort outbox;
7
    private final OutboxEventFactory factory;
8
    private final TransactionPort<Order> tx;  // see hexagonal-architecture.md — Transaction port
9

10
    @Override
11
    public Order execute(CreateOrderCommand cmd) {
12
        return tx.run(() -> {
13
            Order order = Order.create(cmd);
14
            orderRepo.save(order);                                  // business write
15
            outbox.save(factory.create(                             // outbox write — same TX
16
                "platform.notifications",
17
                cmd.traceId(),
18
                order.id().toString(),
19
                "OrderCreated",
20
                EventTypes.EVENT,
21
                OrderCreatedPayload.from(order)
22
            ));
23
            return order;
24
        });
25
    }
26
}

Both writes commit together. No broker call here. The dispatcher publishes asynchronously.

4.5 Alternative: Debezium CDC

When you can’t change application code (legacy migration), use Debezium to read the DB transaction log directly and publish to Kafka. The outbox table still exists; Debezium becomes the dispatcher.

Tradeoffs:

• • No application-level dispatcher to maintain.
• • Stronger ordering guarantees per partition key.
• − Operational complexity (Debezium connectors, schema registry, Kafka Connect).
• − Coupling to DB-specific WAL format.

Default: application-level dispatcher (next section). Debezium when scale justifies it.

5. Claim-Based Dispatch

5.1 Why

Multiple instances of a microservice run for HA. Naive dispatcher: each instance polls the outbox and publishes — same row published N times.

ShedLock is not enough on its own: it serializes ticks but a tick that takes 10s + crash mid-publish leaves rows in ambiguous state.

Claim-based dispatch combines both:

1. ShedLock (@SchedulerLock) — only one instance ticks at a time.
2. Claim flow — that instance atomically transitions a batch of rows from PENDING → IN_FLIGHT, stamping claimed_at and claimed_by = appName@pid. If the instance crashes mid-publish, a TTL (default 5 min) lets another instance re-claim the stuck rows.

5.2 Atomic claim SQL (Postgres example)

1
WITH eligible AS (
2
    SELECT id FROM outbox_message
3
    WHERE origin = :origin
4
      AND (
5
        (status = 'PENDING' AND (next_attempt_at IS NULL OR next_attempt_at <= NOW()))
6
        OR (status = 'IN_FLIGHT' AND claimed_at IS NOT NULL AND claimed_at <= NOW() - :claim_ttl)
7
      )
8
    ORDER BY occurred_at ASC
9
    LIMIT :batch_size
10
    FOR UPDATE SKIP LOCKED
11
)
12
UPDATE outbox_message m
13
   SET status = 'IN_FLIGHT', claimed_at = NOW(), claimed_by = :claimed_by
14
  FROM eligible e
15
 WHERE m.id = e.id
16
 RETURNING m.*;

SQL Server equivalent uses OFFSET 0 ROWS FETCH NEXT :limit ROWS ONLY; its atomic claim-and-return analog of RETURNING is UPDATE … WITH (READPAST, UPDLOCK) … OUTPUT inserted.*, which locks and skips contended rows while returning the claimed rows in a single statement. See codehuntersio-sdk-event-messaging OutboxJdbcAdapter for the canonical SQL Server pattern.

5.3 Dispatcher loop

1
public void execute() {
2
    var claimed = outbox.claimPendingMessages(origin, batchSize, instanceId, claimTtl);
3
    for (var msg : claimed) {
4
        try {
5
            publish(msg);          // → broker
6
            markPublished(msg);    // status=PUBLISHED, publishedAt=now, clear claim
7
        } catch (Exception e) {
8
            markRetryOrFailed(msg, e);  // see §6
9
        }
10
        outbox.update(msg);
11
    }
12
}

5.4 ShedLock scheduling

1
@Scheduled(fixedDelayString = "${outbox.task.fixed-delay:5000}")
2
@SchedulerLock(
3
    name = "${outbox.task.lock-name:outbox-${spring.application.name:unknown}}",
4
    lockAtMostFor = "${outbox.task.lock-at-most:60000}",
5
    lockAtLeastFor = "${outbox.task.lock-at-least:10000}"
6
)
7
public void tick() {
8
    sendOutboxMessages.execute();
9
}

Lock name MUST include the service name so each microservice runs independently. Sharing a global lock across services serializes the whole platform.

5.5 Instance ID

1
String instanceId = applicationName + "@" + ManagementFactory.getRuntimeMXBean().getName();
2
// e.g. "platform-service@12345@host42"

Used in claimed_by so post-mortem can pinpoint which pod stalled.

6. Retry, Backoff & Dead-Letter

6.1 Status transitions

1
PENDING ─claim──> IN_FLIGHT ─publish─┬─ ok ──> PUBLISHED
2
                                     └─ err ─┬─ attempts<max ──> PENDING (next_attempt_at = now+backoff)
3
                                             └─ attempts>=max ─> FAILED ─archive/DLT──> ARCHIVED

6.2 Exponential backoff

1
long computeBackoffMs(int attempts) {
2
    int shift = Math.min(attempts - 1, 30);
3
    long delay = baseBackoffMs << shift;       // 30s, 60s, 120s, 240s, ...
4
    return (delay <= 0 || delay > maxBackoffMs) ? maxBackoffMs : delay;
5
}

Defaults: base=30s, max=1h, maxAttempts=5. Tune per criticality.

6.3 Permanent failures (FAILED)

When attempts >= maxAttempts, status flips to FAILED and the row sits in the table. Options:

Recommended: both. Outbox stays as the auditable record; DLQ is the operational handle.

SDK gap. codehuntersio-sdk-event-messaging deliberately stops at FAILED — it does NOT publish to a DLT broker topic. If you need a DLT, add a small downstream job that polls WHERE status = 'FAILED', publishes to your DLT (with error metadata appended), and flips the row to ARCHIVED. Cross-reference: §6.4.

6.4 Dead-letter on consumer side

DLQ messages MUST retain the original envelope + add error metadata (failing service, exception class, stack trace, attempt count, original timestamp).

7. Idempotent Consumers

7.1 Why mandatory

At-least-once = duplicates are inevitable. The producer retried, the broker replayed on rebalance, the consumer crashed after processing but before ack. Without idempotency the consumer corrupts state.

7.2 Strategies (pick one per consumer)

7.3 Idempotency key table

1
CREATE TABLE consumer_processed_event (
2
    event_id      UUID         PRIMARY KEY,
3
    consumer_name VARCHAR(255) NOT NULL,
4
    processed_at  TIMESTAMP    NOT NULL DEFAULT NOW()
5
);
6

7
CREATE INDEX idx_consumer_processed_at
8
    ON consumer_processed_event(consumer_name, processed_at);

Consumer flow:

⚠️ Dedup + @Transactional belong to a use case behind an inbound port — never on the @RabbitListener/@KafkaListener method. The adapter only deserializes and delegates handler.handle(event); orchestration (dedup + transaction + business logic) lives in the use case.

Adapter (listener) — deserialize + delegate only, no @Transactional, no dedup:

1
@RabbitListener(queues = "platform.notifications")
2
public void onMessage(@Payload String json, @Headers Map<String, Object> headers) {
3
    EventMessage<OrderCreatedPayload> event = mapper.readValue(json, new TypeReference<>() {});
4
    handler.handle(event);  // delegates to ProcessOrderCreatedUseCase via inbound port
5
}

Use case (inbound port impl) — owns dedup + transaction:

1
@Transactional
2
public void handle(EventMessage<OrderCreatedPayload> event) {
3
    UUID eventId = UUID.fromString(event.getHeader().getEventId());  // stable id from Header — NEVER traceId
4
    try {
5
        processed.insert(eventId, consumerName);  // unique constraint
6
    } catch (DuplicateKeyException e) {
7
        log.info("Duplicate event {} skipped", eventId);
8
        return;  // ack and move on
9
    }
10
    process(event);  // business logic
11
}

7.4 Retention

Idempotency table grows forever if untouched. Purge with TTL job — typically 7-30 days, sized to retry window.

8. Schema Evolution & Versioning

8.1 Compatibility modes

8.2 What counts as breaking

• Removing or renaming a field.
• Changing a type (string → int, scalar → object).
• Adding a required field (no default).
• Narrowing an enum value set.
• Changing semantic meaning of an existing field.

8.3 Coexistence

Bump schemaVersion (e.g. 1.0.0 → 2.0.0). Producer publishes BOTH versions for the deprecation window (typically 1-2 release cycles). Consumers migrate independently. Producer drops v1 only after telemetry confirms no consumer reads it.

8.4 Schema registry

For Kafka + Avro/Protobuf: use a schema registry (Confluent, Apicurio) with BACKWARD compatibility enforced. Reject incompatible schemas at registration time, not at runtime.

For JSON: enforce via JSON Schema + a /arch-api-style contract check in CI.

8.5 Consumer tolerance

Consumers MUST ignore unknown fields. In Jackson:

1
JsonMapper.builder()
2
    .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
3
    .build();

9. Topology Externalization

9.1 Hard rule

The microservice MUST NOT declare exchanges, queues, topics, partitions, or bindings in code. No @Bean Queue, @Bean Exchange, @Bean Binding, @Bean NewTopic.

Why:

• Multiple services publishing to the same exchange would duplicate the declaration with risk of drift.
• Topology changes require redeploy if encoded in services.
• Operators (RabbitMQ Operator, Strimzi for Kafka) can’t reconcile.

9.2 Where topology lives

9.3 Bootstrap order

Microservice startup MUST NOT block on exchange/topic existence — broker may be reconciled by an operator after deploy. Use lazy declarations if needed (@RabbitListener is fine; it auto-declares queues at first listen, NOT at startup), and rely on broker errors at first publish to alert ops.

9.4 Publish-time caveat

RabbitTemplate.send() to a non-existent exchange is silently dropped unless publisher confirms + mandatory=true are enabled. Outbox PUBLISHED status means “broker accepted bytes”, NOT “queue received them”. Enable publisher confirms for critical events:

1
@Bean
2
public RabbitTemplate rabbitTemplate(ConnectionFactory cf) {
3
    var rt = new RabbitTemplate(cf);
4
    rt.setMandatory(true);
5
    rt.setConfirmCallback((corrData, ack, cause) -> {
6
        if (!ack) log.error("Broker NACK: {}", cause);
7
    });
8
    rt.setReturnsCallback(returned ->
9
        log.error("Unroutable message to {}: {}", returned.getExchange(), returned.getMessage()));
10
    return rt;
11
}

10. RabbitMQ Specifics

10.1 Exchange types

10.2 Publisher adapter pinning

1
public class RabbitMessagePublisherAdapter implements MessagePublisherPort {
2
    private final RabbitTemplate rabbitTemplate;
3

4
    @Override
5
    public void publish(String topic, String routingKey, String payload) {
6
        var props = new MessageProperties();
7
        props.setContentType(MessageProperties.CONTENT_TYPE_JSON);
8
        props.setContentEncoding(StandardCharsets.UTF_8.name());
9
        props.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
10
        rabbitTemplate.send(topic, routingKey, new Message(payload.getBytes(UTF_8), props));
11
    }
12

13
    @Override
14
    public void publish(String topic, String payload) {
15
        publish(topic, "", payload);  // fanout: empty routing key
16
    }
17
}

Pin every message: contentType=application/json, contentEncoding=UTF-8, deliveryMode=PERSISTENT. Use send() (not convertAndSend) so the adapter controls bytes end-to-end.

10.3 Consumer

1
@RabbitListener(queues = "platform.notifications")
2
public void onMessage(@Payload String json, @Headers Map<String, Object> headers) {
3
    EventMessage<OrderCreatedPayload> event = mapper.readValue(json, new TypeReference<>() {});
4
    handler.handle(event);
5
    // auto-ack on return; throw to NACK + DLX
6
}

10.4 Quorum vs classic queues

Use quorum queues for durability + HA (Raft-based, replicated). Classic mirrored queues were removed in RabbitMQ 4.0 — declaring one fails; migrate to quorum queues (or streams for replay).

10.5 Convention: fanout without routing key

Default platform convention for cross-team broadcast:

• Exchange: <domain>.notifications (durable, fanout).
• Queue per consumer service: <domain>.<consumer-service> (durable).
• Binding: routing key "".

Targeted routing (e.g., payment.RAFFLE) uses topic exchanges + explicit routing keys.

11. Kafka Specifics

11.1 Topic design

• Naming: <bounded-context>.<aggregate>.<event-type> — e.g. orders.order.events, payments.payment.events.
• Partitions: based on throughput + parallelism target. Default 12-24. Cannot reduce without rebuild — start conservative.
• Replication: replication-factor=3 minimum for production.
• Retention: time-based (retention.ms=604800000 = 7 days) OR compact for stateful keys.

11.2 Partition key

message_key = aggregate id. Same aggregate always lands on same partition → ordering preserved per aggregate.

1
kafkaTemplate.send(topic, msg.getKey(), msg.getMessage());

11.3 Producer config

1
spring.kafka.producer:
2
  acks: all                          # wait for ISR replication
3
  enable-idempotence: true           # exactly-once-per-partition semantics
4
  retries: 2147483647                # producer-level retry; outbox handles app-level
5
  max-in-flight-requests-per-connection: 5
6
  compression-type: lz4
7
  linger-ms: 5                       # small batching window
8
  batch-size: 32768

11.4 Consumer config

1
spring.kafka.consumer:
2
  group-id: ${spring.application.name}
3
  auto-offset-reset: earliest        # never lose events on first deploy
4
  enable-auto-commit: false          # manual commit after processing
5
  max-poll-records: 100
6
  isolation-level: read_committed    # ignore aborted transactional writes

Manual commit pattern:

1
@KafkaListener(topics = "orders.order.events")
2
public void onMessage(ConsumerRecord<String, String> record, Acknowledgment ack) {
3
    handler.handle(record.value());
4
    ack.acknowledge();  // commit only after successful processing
5
}

11.5 Exactly-once vs at-least-once

Kafka transactions + read_committed give exactly-once within Kafka. Most platforms still need consumer idempotency (§7) because:

• Sink writes to non-Kafka systems (DB, external API).
• Cross-service flows leave Kafka boundary.

Don’t pay the EOS cost unless you’re staying inside Kafka end-to-end. Use at-least-once + idempotency.

12. Choreography vs Orchestration

12.1 Choreography (default for 2-3 service flows)

Services react to events without a central coordinator. Loose coupling, no single point of failure.

1
OrderService ──OrderCreated──> PaymentService ──PaymentCompleted──> ShippingService

Risks: distributed monitoring, hard to reason about end-to-end flow, hidden coupling.

12.2 Orchestration (for complex sagas)

A coordinator (Temporal, Camunda, Spring State Machine) drives the flow, owns the state, handles compensation. Use when:

• ≥4 services involved.
• Long-running (hours/days).
• Compensating actions required (saga).
• Strong audit + observability needs.

See arch-saga-pattern skill + /arch-saga for canonical implementation.

12.3 ADR required

Choreography vs orchestration is an architecturally significant choice. Document in an ADR per adr.md.

13. Tracing Propagation

13.1 W3C Trace Context

The producer copies the current traceId + spanId into broker headers; the consumer rehydrates them on the receiving side so a single trace spans services.

13.2 Spring config

Spring Boot 3+ with Micrometer Tracing handles propagation automatically when:

• io.micrometer:micrometer-tracing-bridge-otel is on the classpath.
• Auto-instrumentation for the broker is enabled.

For RabbitMQ specifically, attach a MessagePostProcessor that copies tracing context:

1
rabbitTemplate.addBeforePublishPostProcessors(tracePropagationMessagePostProcessor);

See obs-tracing.md for full setup.

13.3 Explicit traceId in envelope

Even with W3C headers, the EventMessage.Header.traceId field is the business-level correlation ID — survives across replays, DLQ redrives, and is searchable in audit logs without parsing broker headers. Always populate it.

14. Hexagonal Placement

Domain never imports spring-amqp, spring-kafka, or any broker SDK. The envelope is pure.

15. Testing

15.1 Test pyramid

15.2 Outbox unit test essentials

• claimPendingMessages returns claimed rows with claimed_at, claimed_by set.
• Stale IN_FLIGHT (older than claimTtl) re-claimable.
• PENDING with future next_attempt_at NOT claimed.
• Concurrent claims by 2 instances → no double-claim (use FOR UPDATE SKIP LOCKED test).
• Exponential backoff doubles per attempt, caps at maxBackoffMs.
• attempts >= maxAttempts → status FAILED.

15.3 Consumer idempotency test

1
@Test
2
void duplicate_event_processed_only_once() {
3
    var event = makeEvent("evt-1");
4

5
    consumer.onEvent(event);
6
    consumer.onEvent(event);  // duplicate
7

8
    verify(handler, times(1)).handle(any());
9
}

15.4 Contract testing

Producer publishes → Pact records the schema → Consumer verifies against the same. Catches schema drift before deploy.

16. Operational Metrics & Alerting

16.1 Producer-side metrics

16.2 Consumer-side metrics

16.3 Broker metrics

Implement via Micrometer + Prometheus per obs-metrics.md.

17. Anti-patterns

18. Pre-Merge Checklist

For PRs that publish events, consume events, or touch the outbox:

• Outbox pattern used — no direct broker calls in use case or controller.
• Business write + outbox insert wrapped in same local transaction (Transaction port).
• EventMessage<T> envelope (Header + Body) with traceId, origin, aggregate, eventName.
• Event name in past tense, PascalCase; type is EVENT or COMMAND.
• origin injected via factory from spring.application.name, NEVER set manually.
• No @Bean Queue / Exchange / Binding / NewTopic in microservice code.
• Topology declared externally (definitions.json / Terraform / Operator) — referenced in PR.
• Consumer idempotency in place (§7) — strategy documented.
• Retry policy + backoff configured; maxAttempts set per criticality.
• DLQ/DLT path defined for permanent failures.
• Schema change is backward-compatible OR schemaVersion bumped + coexistence plan.
• FAIL_ON_UNKNOWN_PROPERTIES=false on consumer JsonMapper.
• ShedLock lock name includes spring.application.name.
• Publisher confirms + mandatory=true enabled for critical RabbitMQ events.
• Kafka: acks=all, enable-idempotence=true, manual commit in consumer.
• Tracing context propagated in broker headers (traceparent).
• Metrics added: pending, in_flight stale, failed, dispatch duration, consumer lag.
• Alert thresholds defined for pending backlog + failed rate + consumer lag.
• No PII / secrets in payload OR logs (see security.md §10).
• Tests: outbox JDBC integration (Testcontainers), publisher adapter (Testcontainers broker), consumer idempotency, schema contract.
• ADR drafted if changing broker, topology approach, schema-evolution policy, or choreography↔orchestration.
• /arch-outbox agent run on diff — no findings.