59 · SAP Event Mesh

Connect CAP to Event Mesh, design topic namespaces, publish CloudEvents, handle DLQ, and ack/nack messages.

Episode 59 — SAP Event Mesh — Advanced Topics

What you'll learn

• Event Mesh namespace design and topic hierarchies
• Queue-based vs topic subscription patterns
• Guaranteed delivery and at-least-once semantics
• Dead letter queues and retry policies
• Monitoring Event Mesh in BTP Cockpit

Why this matters

Lesson 28 covered the basics. This lesson covers production patterns — what happens when messages fail, how to design namespaces for multi-service systems, and how to monitor message flows in production.

Namespace and topic hierarchy

Format: <namespace>/<service>/<version>/<entity>/<event>
Example: sap/cap/bookshop/orders/v1/OrderCreated
         ├── sap             ← org root
         ├── cap/bookshop    ← application namespace
         ├── orders          ← service/domain
         ├── v1              ← version (never break consumers)
         └── OrderCreated    ← event name

Namespace is configured in em-config.json:

{
  "emname":    "bookshop-em",
  "namespace": "sap/cap/bookshop",
  "version":   "1.1.0",
  "options":   { "management": true }
}

Queue design patterns

Fan-out (one topic, multiple consumers)

topic: orders/v1/OrderCreated
    ├── queue: fulfillment-consumer    (fulfillment service reads)
    ├── queue: notification-consumer   (notification service reads)
    └── queue: analytics-consumer      (analytics service reads)

Each consumer gets every message, independently.

Competing consumers (one queue, multiple workers)

topic: orders/v1/ProcessQueue → queue: order-processor
    worker-1 ─┐
    worker-2 ─┤ compete for messages (each message delivered to ONE worker)
    worker-3 ─┘

Use for horizontal scaling of a single consumer.

em-config.json — production setup

{
  "emname":    "bookshop-em",
  "namespace": "sap/cap/bookshop",
  "version":   "1.2.0",
  "options": {
    "management":    true,
    "messagingrest": true
  },
  "rules": {
    "topicRules": {
      "publishFilter":   ["*"],          // producers can publish to any topic
      "subscribeFilter": ["*"]           // consumers can subscribe to any topic
    },
    "queueRules": {
      "publishFilter":   ["*"],
      "subscribeFilter": ["*"]
    }
  },
  "services": [
    {
      "name": "bookshop-q1",
      "type": "queue",
      "subscribeFilter": ["sap/cap/bookshop/+/+/+"]   // wildcard subscription
    }
  ]
}

Dead letter queue (DLQ)

When a consumer repeatedly fails to process a message:

topic: orders/v1/OrderCreated
    → queue: fulfillment-consumer
        Message fails 3 times (maxRetries=3)
            → dead-letter-queue: fulfillment-consumer/dlq

messaging.on('sap/cap/bookshop/orders/v1/OrderCreated', async (msg) => {
  try {
    await processOrder(msg.data)
    await msg.ack()
  } catch (err) {
    console.error(`[NACK] Order processing failed:`, err.message)
    await msg.nack()   // Message goes back; after maxRetries → DLQ
  }
})

DLQ consumer for monitoring and replaying

// Consume from DLQ to alert and optionally replay
messaging.on('sap/cap/bookshop/orders/v1/OrderCreated/dlq', async (msg) => {
  await alertTeam({
    subject: 'Order processing failed after retries',
    body:    JSON.stringify(msg.data)
  })
  // Log to database for manual review/replay
  await INSERT.into('FailedMessages').entries({
    topic:   msg.headers['x-original-topic'],
    data:    JSON.stringify(msg.data),
    error:   msg.headers['x-original-error'],
    receivedAt: new Date()
  })
  await msg.ack()   // Ack from DLQ — prevents infinite DLQ growth
})

Message headers (CloudEvents)

messaging.on('...', async (msg) => {
  // Standard CloudEvents headers
  console.log(msg.headers['ce-id'])           // message UUID
  console.log(msg.headers['ce-type'])         // event type
  console.log(msg.headers['ce-source'])       // publishing service
  console.log(msg.headers['ce-time'])         // UTC timestamp

  // SAP-specific
  console.log(msg.headers['x-em-timestamp'])  // Event Mesh receipt time
})

Publishing with metadata

await messaging.emit('sap/cap/bookshop/orders/v1/OrderCreated', {
  ID: order.ID, total: order.total
}, {
  headers: {
    'x-correlation-id': req.headers['x-correlation-id'],
    'x-user-id':        req.user.id
  }
})

Monitoring in BTP Cockpit

BTP Cockpit → Instances → Event Mesh → Manage:
- Messages: queue size, in-flight, DLQ count
- Subscriptions: active consumers, connection state
- Metrics: publish/consume rate, latency P95
- Alerts: configure max queue depth thresholds

Hands-on exercise

1. Create two consumer services that subscribe to the same OrderCreated topic
2. Publish 5 orders and verify both consumers receive all 5
3. Make one consumer fail on even-numbered orders — verify DLQ fills
4. Write a DLQ consumer that inserts failed messages to a FailedEvents table

Checkpoint ✓

• [ ] Namespaces: org/app/service/v1/EventName — include version to avoid breaking changes
• [ ] Fan-out: one topic, multiple queues — each queue gets every message
• [ ] msg.nack() retries; after maxRetries the message moves to DLQ
• [ ] Always ack from DLQ (even just to log it) — prevents infinite growth

Sign in for progress tracking & Q&A →