1. The Problem: MQTT 3 Without Shared Subscriptions

⚠️ MQTT 3.1.1 vs MQTT 5.0 — The Missing Feature

MQTT 5.0 introduced $share/group/topic — shared subscriptions that load-balance messages across consumers. MQTT 3.1.1 does NOT have this. Every subscriber gets every message. Period.

# MQTT 3.1.1: Every subscriber gets EVERY message (no load balancing)
#
#                    ┌──────────────┐
#  Publisher ──────► │  MQTT Broker │
#                    └──────┬───────┘
#                           │
#              ┌────────────┼────────────┐
#              ▼            ▼            ▼
#          Instance A   Instance B   Instance C
#          (ALL msgs)   (ALL msgs)   (ALL msgs)
#
# MQTT 5.0 with $share:
#
#                    ┌──────────────┐
#  Publisher ──────► │  MQTT Broker │
#                    └──────┬───────┘
#                           │
#              ┌────────────┼────────────┐
#              ▼            ▼            ▼
#          Instance A   Instance B   Instance C
#          (1/3 msgs)   (1/3 msgs)   (1/3 msgs)
#
# We're stuck with MQTT 3.1.1. This cheatsheet solves it.

Scaling is broken: Adding more instances = more duplicate processing, not less load
Slow subscriber problem: Broker queues grow → memory pressure → OOM or message expiry/loss
No backpressure: MQTT has no flow control — broker pushes, subscriber must keep up or drown
"Just upgrade to MQTT 5" isn't always possible: Legacy brokers, IoT firmware, vendor lock-in, certification requirements

The consequences of doing nothing:

3 instances × 10k msg/sec = 30k duplicate processing operations/sec
Broker memory grows unbounded if any subscriber falls behind
No way to tell the broker "slow down" — it just keeps pushing
Eventually: message loss, broker crash, or subscriber OOM

2. Architecture Decision: The Winner

Option A: MQTT → Internal Queue → Workers

Single MQTT subscriber, bounded in-memory queue, thread pool workers.

⚠️ Single point of failure. No persistence. Messages lost on crash.

✅ Option B: MQTT → RabbitMQ → Workers (RECOMMENDED)

Single MQTT subscriber (bridge) → publishes to RabbitMQ → multiple competing consumers.

RabbitMQ handles: load balancing, backpressure (prefetch), persistence, retry, DLQ
Workers scale horizontally — just add instances
Simple, proven, no custom distribution code

Option C: MQTT → Kafka → Consumers

Consumer groups = shared subscriptions natively. Highest throughput. Event replay.

⚠️ Higher ops complexity. Overkill if < 100k msg/sec.

Criteria	In-Memory	RabbitMQ ✅	Kafka
Simplicity	★★★	★★★	★★
Persistence	✗	✓	✓
Backpressure	Manual	Built-in	Built-in
Scalability	★	★★★	★★★★★
Ops Complexity	Low	Low	High
DLQ / Retry	Manual	Built-in	Manual

# The winning architecture:
#
#  IoT Devices ──► MQTT Broker ──► [ Bridge App ] ──► RabbitMQ ──► Worker 1
#                                  (single instance)              ──► Worker 2
#                                                                 ──► Worker 3
#                                                                       │
#                                                                       ▼
#                                                                   MongoDB

3. Spring Boot + MQTT Setup

Dependencies (build.gradle.kts)

dependencies {
    implementation("org.springframework.boot:spring-boot-starter")
    implementation("org.springframework.integration:spring-integration-mqtt")
    implementation("org.springframework.boot:spring-boot-starter-amqp")
    implementation("org.springframework.boot:spring-boot-starter-data-mongodb")
    implementation("org.springframework.boot:spring-boot-starter-actuator")
    implementation("io.micrometer:micrometer-registry-prometheus")
    implementation("io.github.resilience4j:resilience4j-spring-boot3:2.2.0")
}

application.yml — MQTT Configuration

mqtt:
  broker-url: tcp://mqtt-broker:1883
  # IMPORTANT: clientId must be unique per instance!
  client-id: bridge-${random.uuid}
  username: ${MQTT_USER:}
  password: ${MQTT_PASS:}
  qos: 1
  clean-session: true
  completion-timeout: 5000
  topics:
    - sensor/#
    - device/#
    - control/#

MQTT Configuration Class

package com.example.mqttbridge.config;

import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;

@Configuration
public class MqttConfig {

    @Value("${mqtt.broker-url}")
    private String brokerUrl;

    @Value("${mqtt.client-id}")
    private String clientId;

    @Value("${mqtt.username:}")
    private String username;

    @Value("${mqtt.password:}")
    private String password;

    @Value("${mqtt.qos:1}")
    private int qos;

    @Value("${mqtt.clean-session:true}")
    private boolean cleanSession;

    @Value("${mqtt.topics}")
    private String[] topics;

    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(new String[]{brokerUrl});
        options.setCleanSession(cleanSession);
        options.setAutomaticReconnect(true);          // auto-reconnect!
        options.setConnectionTimeout(10);
        options.setKeepAliveInterval(30);
        options.setMaxInflight(1000);                  // in-flight window
        if (!username.isEmpty()) {
            options.setUserName(username);
            options.setPassword(password.toCharArray());
        }
        factory.setConnectionOptions(options);
        return factory;
    }

    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MqttPahoMessageDrivenChannelAdapter mqttInbound(
            MqttPahoClientFactory factory,
            MessageChannel mqttInputChannel) {
        MqttPahoMessageDrivenChannelAdapter adapter =
            new MqttPahoMessageDrivenChannelAdapter(
                clientId, factory, topics);
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(qos);
        adapter.setOutputChannel(mqttInputChannel);
        return adapter;
    }
}

💡 ClientId Strategy

Each MQTT connection must have a unique clientId. If two clients connect with the same ID, the broker disconnects the first one. Use ${random.uuid} or hostname-based IDs like bridge-${HOSTNAME}.

4. Backpressure Solution: Bounded Internal Buffer

Why a buffer between MQTT and RabbitMQ?

If RabbitMQ is temporarily slow or down, MQTT messages pile up. A bounded buffer prevents OOM and gives you control over what happens when capacity is exceeded: drop oldest, reject, or block.

package com.example.mqttbridge.buffer;

import io.micrometer.core.instrument.Counter;
import io.micrometer.core.instrument.Gauge;
import io.micrometer.core.instrument.MeterRegistry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import jakarta.annotation.PostConstruct;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

@Component
public class MqttMessageBuffer {

    private static final Logger log = LoggerFactory.getLogger(MqttMessageBuffer.class);

    public record BufferedMessage(
        String topic,
        String payload,
        int qos,
        long receivedAt
    ) {}

    public enum OverflowStrategy { DROP_OLDEST, REJECT, BLOCK }

    private final LinkedBlockingQueue<BufferedMessage> queue;
    private final OverflowStrategy strategy;
    private final Counter droppedCounter;
    private final Counter receivedCounter;

    public MqttMessageBuffer(MeterRegistry registry) {
        int capacity = 10_000; // configurable
        this.queue = new LinkedBlockingQueue<>(capacity);
        this.strategy = OverflowStrategy.DROP_OLDEST;

        this.droppedCounter = Counter.builder("mqtt.bridge.dropped")
            .description("Messages dropped due to full buffer")
            .register(registry);
        this.receivedCounter = Counter.builder("mqtt.messages.received")
            .description("Total MQTT messages received")
            .register(registry);
        Gauge.builder("mqtt.bridge.queue.size", queue, LinkedBlockingQueue::size)
            .description("Current buffer queue depth")
            .register(registry);
    }

    public void offer(String topic, String payload, int qos) {
        receivedCounter.increment();
        BufferedMessage msg = new BufferedMessage(topic, payload, qos, System.currentTimeMillis());

        if (!queue.offer(msg)) {
            switch (strategy) {
                case DROP_OLDEST -> {
                    queue.poll(); // remove oldest
                    queue.offer(msg);
                    droppedCounter.increment();
                    log.warn("Buffer full — dropped oldest message. Queue size: {}", queue.size());
                }
                case REJECT -> {
                    droppedCounter.increment();
                    log.warn("Buffer full — rejected message on topic: {}", topic);
                }
                case BLOCK -> {
                    try {
                        queue.put(msg); // blocks until space available
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        droppedCounter.increment();
                    }
                }
            }
        }
    }

    public BufferedMessage poll(long timeoutMs) throws InterruptedException {
        return queue.poll(timeoutMs, TimeUnit.MILLISECONDS);
    }

    public int size() {
        return queue.size();
    }
}

Wire it to the MQTT channel

package com.example.mqttbridge.handler;

import com.example.mqttbridge.buffer.MqttMessageBuffer;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.stereotype.Component;

@Component
public class MqttInboundHandler {

    private final MqttMessageBuffer buffer;

    public MqttInboundHandler(MqttMessageBuffer buffer) {
        this.buffer = buffer;
    }

    @ServiceActivator(inputChannel = "mqttInputChannel")
    public void handleMessage(Message<?> message) {
        MessageHeaders headers = message.getHeaders();
        String topic = (String) headers.get("mqtt_receivedTopic");
        int qos = (int) headers.getOrDefault("mqtt_receivedQos", 1);
        String payload = message.getPayload().toString();

        buffer.offer(topic, payload, qos);
    }
}

5. The Bridge: MQTT → RabbitMQ

RabbitMQ Configuration

spring:
  rabbitmq:
    host: rabbitmq
    port: 5672
    username: ${RABBITMQ_USER:guest}
    password: ${RABBITMQ_PASS:guest}
    publisher-confirm-type: correlated    # publisher confirms!
    publisher-returns: true

Exchange & Queue Setup

package com.example.mqttbridge.config;

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitConfig {

    public static final String EXCHANGE = "mqtt.messages";
    public static final String QUEUE = "mqtt.messages.processing";
    public static final String DLX = "mqtt.messages.dlx";
    public static final String DLQ = "mqtt.messages.dlq";

    @Bean
    public TopicExchange mqttExchange() {
        return new TopicExchange(EXCHANGE, true, false);
    }

    @Bean
    public Queue processingQueue() {
        return QueueBuilder.durable(QUEUE)
            .withArgument("x-dead-letter-exchange", DLX)
            .withArgument("x-dead-letter-routing-key", "dead")
            .build();
    }

    @Bean
    public Binding processingBinding() {
        return BindingBuilder.bind(processingQueue())
            .to(mqttExchange())
            .with("#"); // all routing keys
    }

    // Dead Letter Queue
    @Bean
    public DirectExchange dlxExchange() {
        return new DirectExchange(DLX, true, false);
    }

    @Bean
    public Queue dlq() {
        return QueueBuilder.durable(DLQ).build();
    }

    @Bean
    public Binding dlqBinding() {
        return BindingBuilder.bind(dlq())
            .to(dlxExchange())
            .with("dead");
    }
}

The Bridge Component

package com.example.mqttbridge.bridge;

import com.example.mqttbridge.buffer.MqttMessageBuffer;
import com.example.mqttbridge.buffer.MqttMessageBuffer.BufferedMessage;
import com.example.mqttbridge.config.RabbitConfig;
import io.micrometer.core.instrument.Counter;
import io.micrometer.core.instrument.MeterRegistry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Component;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;

@Component
public class MqttToRabbitBridge {

    private static final Logger log = LoggerFactory.getLogger(MqttToRabbitBridge.class);

    private final MqttMessageBuffer buffer;
    private final RabbitTemplate rabbitTemplate;
    private final Counter publishedCounter;
    private final Counter failedCounter;
    private final AtomicBoolean running = new AtomicBoolean(true);
    private final ExecutorService executor = Executors.newSingleThreadExecutor();

    public MqttToRabbitBridge(
            MqttMessageBuffer buffer,
            RabbitTemplate rabbitTemplate,
            MeterRegistry registry) {
        this.buffer = buffer;
        this.rabbitTemplate = rabbitTemplate;
        this.publishedCounter = Counter.builder("mqtt.bridge.published")
            .description("Messages published to RabbitMQ")
            .register(registry);
        this.failedCounter = Counter.builder("mqtt.bridge.failed")
            .description("Failed publishes to RabbitMQ")
            .register(registry);
    }

    @EventListener(ApplicationReadyEvent.class)
    public void start() {
        executor.submit(this::bridgeLoop);
        log.info("MQTT→RabbitMQ bridge started");
    }

    private void bridgeLoop() {
        while (running.get()) {
            try {
                BufferedMessage msg = buffer.poll(1000);
                if (msg == null) continue;

                // MQTT topic → RabbitMQ routing key
                // sensor/temperature/room1 → sensor.temperature.room1
                String routingKey = msg.topic().replace('/', '.');

                rabbitTemplate.convertAndSend(
                    RabbitConfig.EXCHANGE,
                    routingKey,
                    msg.payload(),
                    m -> {
                        m.getMessageProperties().setDeliveryMode(
                            MessageDeliveryMode.PERSISTENT);
                        m.getMessageProperties().setHeader(
                            "mqtt_topic", msg.topic());
                        m.getMessageProperties().setHeader(
                            "mqtt_qos", msg.qos());
                        m.getMessageProperties().setTimestamp(
                            new java.util.Date(msg.receivedAt()));
                        return m;
                    });
                publishedCounter.increment();

            } catch (Exception e) {
                failedCounter.increment();
                log.error("Bridge publish failed: {}", e.getMessage());
                try { Thread.sleep(1000); } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }
    }

    public void stop() {
        running.set(false);
        executor.shutdown();
    }
}

6. RabbitMQ Consumer Workers

🎯 This is where backpressure happens

prefetch=10 means each consumer only gets 10 unacked messages at a time. RabbitMQ won't push more until the consumer acks. This IS the backpressure mechanism.

Worker application.yml

spring:
  rabbitmq:
    host: rabbitmq
    port: 5672
    username: ${RABBITMQ_USER:guest}
    password: ${RABBITMQ_PASS:guest}
    listener:
      simple:
        prefetch: 10                    # ← BACKPRESSURE!
        concurrency: 5                  # min threads
        max-concurrency: 20             # max threads
        acknowledge-mode: manual        # explicit ack
        default-requeue-rejected: false # send to DLQ, don't requeue
        retry:
          enabled: true
          initial-interval: 1000        # 1s
          max-interval: 30000           # 30s max
          multiplier: 2.0               # exponential backoff
          max-attempts: 3               # then → DLQ

Consumer Component

package com.example.mqttworker.consumer;

import com.rabbitmq.client.Channel;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Timer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.UUID;

@Component
public class MqttMessageConsumer {

    private static final Logger log = LoggerFactory.getLogger(MqttMessageConsumer.class);

    private final MessageProcessor processor;
    private final Timer processingTimer;

    public MqttMessageConsumer(MessageProcessor processor, MeterRegistry registry) {
        this.processor = processor;
        this.processingTimer = Timer.builder("rabbitmq.consumer.processing.time")
            .description("Time to process a single message")
            .register(registry);
    }

    @RabbitListener(queues = "mqtt.messages.processing")
    public void consume(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        String correlationId = UUID.randomUUID().toString();

        MDC.put("correlationId", correlationId);
        MDC.put("topic", (String) message.getMessageProperties()
            .getHeader("mqtt_topic"));

        try {
            processingTimer.record(() -> {
                processor.process(
                    (String) message.getMessageProperties()
                        .getHeader("mqtt_topic"),
                    new String(message.getBody())
                );
            });

            channel.basicAck(deliveryTag, false);
            log.debug("Message processed and acked");

        } catch (Exception e) {
            log.error("Processing failed: {}", e.getMessage(), e);
            // reject + don't requeue → goes to DLQ
            channel.basicNack(deliveryTag, false, false);
        } finally {
            MDC.clear();
        }
    }
}

// Simple processor interface
@FunctionalInterface
interface MessageProcessor {
    void process(String topic, String payload);
}

7. MongoDB Integration

Document Model

package com.example.mqttworker.model;

import org.springframework.data.annotation.Id;
import org.springframework.data.mongodb.core.index.CompoundIndex;
import org.springframework.data.mongodb.core.index.Indexed;
import org.springframework.data.mongodb.core.mapping.Document;

import java.time.Instant;

@Document(collection = "mqtt_messages")
@CompoundIndex(name = "topic_ts_idx", def = "{'topic': 1, 'receivedAt': -1}")
public class MqttMessageDoc {

    @Id
    private String id;

    @Indexed
    private String messageId;     // for idempotency (upsert)

    @Indexed
    private String topic;

    private String payload;
    private int qos;

    @Indexed(expireAfter = "30d")  // TTL: auto-delete after 30 days
    private Instant receivedAt;

    private Instant processedAt;
    private boolean processed;

    // constructors, getters, setters
    public MqttMessageDoc() {}

    public MqttMessageDoc(String messageId, String topic, String payload,
                          int qos, Instant receivedAt) {
        this.messageId = messageId;
        this.topic = topic;
        this.payload = payload;
        this.qos = qos;
        this.receivedAt = receivedAt;
        this.processed = false;
    }

    // ... getters/setters omitted for brevity
    public String getId() { return id; }
    public String getMessageId() { return messageId; }
    public String getTopic() { return topic; }
    public String getPayload() { return payload; }
    public int getQos() { return qos; }
    public Instant getReceivedAt() { return receivedAt; }
    public Instant getProcessedAt() { return processedAt; }
    public boolean isProcessed() { return processed; }
    public void setProcessedAt(Instant processedAt) { this.processedAt = processedAt; }
    public void setProcessed(boolean processed) { this.processed = processed; }
}

Bulk Write Service

package com.example.mqttworker.service;

import com.example.mqttworker.model.MqttMessageDoc;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Timer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.mongodb.core.BulkOperations;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.stereotype.Service;

import java.time.Instant;
import java.util.List;

@Service
public class MongoMessageService {

    private static final Logger log = LoggerFactory.getLogger(MongoMessageService.class);

    private final MongoTemplate mongoTemplate;
    private final Timer bulkWriteTimer;

    public MongoMessageService(MongoTemplate mongoTemplate, MeterRegistry registry) {
        this.mongoTemplate = mongoTemplate;
        this.bulkWriteTimer = Timer.builder("mongodb.bulk.write.time")
            .description("Time for bulk write operations")
            .register(registry);
    }

    /**
     * Bulk upsert — idempotent by messageId.
     * Handles redelivery without duplicates.
     */
    public void bulkUpsert(List<MqttMessageDoc> messages) {
        if (messages.isEmpty()) return;

        bulkWriteTimer.record(() -> {
            BulkOperations bulk = mongoTemplate.bulkOps(
                BulkOperations.BulkMode.UNORDERED,
                MqttMessageDoc.class);

            for (MqttMessageDoc msg : messages) {
                Query query = Query.query(
                    Criteria.where("messageId").is(msg.getMessageId()));
                Update update = new Update()
                    .setOnInsert("messageId", msg.getMessageId())
                    .setOnInsert("topic", msg.getTopic())
                    .setOnInsert("payload", msg.getPayload())
                    .setOnInsert("qos", msg.getQos())
                    .setOnInsert("receivedAt", msg.getReceivedAt())
                    .set("processedAt", Instant.now())
                    .set("processed", true);
                bulk.upsert(query, update);
            }

            var result = bulk.execute();
            log.debug("Bulk upsert: {} inserted, {} modified",
                result.getInsertedCount(), result.getModifiedCount());
        });
    }

    /** Single insert (for non-batched processing) */
    public void save(MqttMessageDoc message) {
        mongoTemplate.save(message);
    }
}

MongoDB application.yml

spring:
  data:
    mongodb:
      uri: mongodb://${MONGO_USER:root}:${MONGO_PASS:password}@mongodb:27017/mqtt_data?authSource=admin
      auto-index-creation: true

8. Slow Subscriber Patterns

Pattern 1: Competing Consumers (via RabbitMQ)

Multiple workers consume from the same queue — RabbitMQ distributes round-robin
prefetch limits work-in-progress per consumer
Scale workers up/down based on queue depth (K8s HPA)

Pattern 2: Message Batching

package com.example.mqttworker.batch;

import com.example.mqttworker.model.MqttMessageDoc;
import com.example.mqttworker.service.MongoMessageService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.locks.ReentrantLock;

@Component
public class MessageBatcher {

    private static final Logger log = LoggerFactory.getLogger(MessageBatcher.class);
    private static final int BATCH_SIZE = 100;

    private final MongoMessageService mongoService;
    private final List<MqttMessageDoc> batch = new ArrayList<>();
    private final ReentrantLock lock = new ReentrantLock();

    public MessageBatcher(MongoMessageService mongoService) {
        this.mongoService = mongoService;
    }

    public void add(MqttMessageDoc message) {
        lock.lock();
        try {
            batch.add(message);
            if (batch.size() >= BATCH_SIZE) {
                flush();
            }
        } finally {
            lock.unlock();
        }
    }

    @Scheduled(fixedDelay = 5000) // flush every 5s regardless
    public void scheduledFlush() {
        flush();
    }

    private void flush() {
        lock.lock();
        List<MqttMessageDoc> toFlush;
        try {
            if (batch.isEmpty()) return;
            toFlush = new ArrayList<>(batch);
            batch.clear();
        } finally {
            lock.unlock();
        }
        log.debug("Flushing batch of {} messages", toFlush.size());
        mongoService.bulkUpsert(toFlush);
    }
}

Pattern 3: Priority Processing

Route by topic: control/# → high-priority queue, sensor/# → normal queue
Use separate RabbitMQ queues per priority level
High priority consumers: low prefetch (1-2), immediate processing
Low priority consumers: high prefetch (20+), batch processing

Pattern 4: Shedding (Controlled Drop)

// In the bridge: check queue depth before publishing
public void publishWithShedding(BufferedMessage msg) {
    // Check RabbitMQ queue depth via management API
    long queueDepth = getQueueDepth("mqtt.messages.processing");

    if (queueDepth > 50_000 && isLowPriority(msg.topic())) {
        sheddedCounter.increment();
        log.warn("Shedding low-priority message on topic: {} (queue depth: {})",
            msg.topic(), queueDepth);
        return; // intentionally drop
    }

    // Normal publish
    publishToRabbit(msg);
}

private boolean isLowPriority(String topic) {
    return topic.startsWith("sensor/") || topic.startsWith("telemetry/");
}

⚠️ Better to drop some messages than to crash the entire system. Track and alert on drop rate.

9. Resilience & Error Handling

Resilience4j Configuration

# application.yml
resilience4j:
  circuitbreaker:
    instances:
      rabbitmq:
        sliding-window-size: 10
        failure-rate-threshold: 50
        wait-duration-in-open-state: 30s
        permitted-number-of-calls-in-half-open-state: 3
      mongodb:
        sliding-window-size: 10
        failure-rate-threshold: 50
        wait-duration-in-open-state: 30s
  retry:
    instances:
      rabbitmq:
        max-attempts: 3
        wait-duration: 2s
        exponential-backoff-multiplier: 2
      mongodb:
        max-attempts: 3
        wait-duration: 1s

Resilient Publisher with Fallback

package com.example.mqttbridge.bridge;

import io.github.resilience4j.circuitbreaker.annotation.CircuitBreaker;
import io.github.resilience4j.retry.annotation.Retry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

import java.io.FileWriter;
import java.io.IOException;
import java.io.PrintWriter;
import java.time.Instant;

@Component
public class ResilientPublisher {

    private static final Logger log = LoggerFactory.getLogger(ResilientPublisher.class);

    private final RabbitTemplate rabbitTemplate;

    public ResilientPublisher(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    @CircuitBreaker(name = "rabbitmq", fallbackMethod = "fallbackToFile")
    @Retry(name = "rabbitmq")
    public void publish(String exchange, String routingKey, String payload) {
        rabbitTemplate.convertAndSend(exchange, routingKey, payload);
    }

    /**
     * Fallback: write to local file when RabbitMQ is down.
     * A recovery process can replay these later.
     */
    public void fallbackToFile(String exchange, String routingKey,
                               String payload, Exception ex) {
        log.error("RabbitMQ unavailable, writing to fallback file: {}",
            ex.getMessage());
        try (PrintWriter pw = new PrintWriter(
                new FileWriter("/tmp/mqtt-fallback.jsonl", true))) {
            pw.printf("{"ts":"%s","exchange":"%s","
                + ""routingKey":"%s","payload":%s}%n",
                Instant.now(), exchange, routingKey, payload);
        } catch (IOException ioe) {
            log.error("CRITICAL: Fallback file write failed!", ioe);
        }
    }
}

Health Checks

# application.yml — expose all health indicators
management:
  endpoint:
    health:
      show-details: always
  health:
    rabbit:
      enabled: true
    mongo:
      enabled: true
    # Custom MQTT health check below

package com.example.mqttbridge.health;

import org.springframework.boot.actuate.health.Health;
import org.springframework.boot.actuate.health.HealthIndicator;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.stereotype.Component;

@Component
public class MqttHealthIndicator implements HealthIndicator {

    private final MqttPahoMessageDrivenChannelAdapter adapter;

    public MqttHealthIndicator(MqttPahoMessageDrivenChannelAdapter adapter) {
        this.adapter = adapter;
    }

    @Override
    public Health health() {
        if (adapter.isRunning()) {
            return Health.up()
                .withDetail("clientId", adapter.getBeanName())
                .build();
        }
        return Health.down()
            .withDetail("reason", "MQTT adapter not running")
            .build();
    }
}

Graceful Shutdown

package com.example.mqttbridge.lifecycle;

import com.example.mqttbridge.bridge.MqttToRabbitBridge;
import com.example.mqttbridge.buffer.MqttMessageBuffer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.event.ContextClosedEvent;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Component;

@Component
public class GracefulShutdown {

    private static final Logger log = LoggerFactory.getLogger(GracefulShutdown.class);

    private final MqttToRabbitBridge bridge;
    private final MqttMessageBuffer buffer;

    public GracefulShutdown(MqttToRabbitBridge bridge, MqttMessageBuffer buffer) {
        this.bridge = bridge;
        this.buffer = buffer;
    }

    @EventListener(ContextClosedEvent.class)
    public void onShutdown() {
        log.info("Graceful shutdown: draining {} buffered messages...",
            buffer.size());
        bridge.stop();
        // Give the bridge loop time to drain
        try { Thread.sleep(5000); } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        log.info("Shutdown complete. Remaining in buffer: {}", buffer.size());
    }
}

10. Monitoring & Observability

Key Metrics (all registered via Micrometer)

Metric	Type	Alert When
mqtt.messages.received	Counter	Rate drops to 0 (connection lost)
mqtt.bridge.queue.size	Gauge	> 5000 (consumer falling behind)
mqtt.bridge.dropped	Counter	> 0 (buffer overflow)
mqtt.bridge.published	Counter	Rate diverges from received
rabbitmq.consumer.processing.time	Timer	p99 > 1s
mongodb.bulk.write.time	Timer	p99 > 5s

Actuator & Prometheus Config

# application.yml
management:
  endpoints:
    web:
      exposure:
        include: health,info,metrics,prometheus
  metrics:
    tags:
      application: mqtt-bridge
    export:
      prometheus:
        enabled: true
  endpoint:
    health:
      show-details: always

# Structured logging (logback-spring.xml)
# Use MDC fields: correlationId, topic
logging:
  pattern:
    console: "%d{ISO8601} [%thread] %-5level %logger{36} [%X{correlationId}] [%X{topic}] - %msg%n"

PromQL Alert Rules

# Buffer filling up
- alert: MqttBufferHigh
  expr: mqtt_bridge_queue_size > 5000
  for: 2m
  annotations:
    summary: "MQTT bridge buffer depth {{ $value }}"

# Messages being dropped
- alert: MqttMessagesDropped
  expr: rate(mqtt_bridge_dropped_total[5m]) > 0
  annotations:
    summary: "MQTT bridge is dropping messages"

# Consumer lag (messages in RabbitMQ queue)
- alert: RabbitQueueBacklog
  expr: rabbitmq_queue_messages{queue="mqtt.messages.processing"} > 10000
  for: 5m
  annotations:
    summary: "RabbitMQ processing queue backlog: {{ $value }}"

11. Scaling Strategy

# Deployment Architecture
#
# ┌─────────────────────────────────────────────────────────┐
# │                    MQTT Broker                          │
# │              (Mosquitto / HiveMQ / etc.)                │
# └──────────────────────┬──────────────────────────────────┘
#                        │
#                        ▼
# ┌──────────────────────────────────────┐
# │  Bridge App (SINGLE instance)        │   ← leader election
# │  - Subscribes to all MQTT topics     │      if HA needed
# │  - Bounded buffer (10k)              │
# │  - Publishes to RabbitMQ             │
# └──────────────────────┬───────────────┘
#                        │
#                        ▼
# ┌──────────────────────────────────────┐
# │          RabbitMQ Cluster            │
# │  - Durable queues                   │
# │  - DLQ configured                   │
# │  - Prefetch-based backpressure      │
# └────┬─────────┬─────────┬────────────┘
#      │         │         │
#      ▼         ▼         ▼
# ┌────────┐ ┌────────┐ ┌────────┐
# │Worker 1│ │Worker 2│ │Worker 3│    ← scale horizontally
# │        │ │        │ │        │       (K8s HPA)
# └───┬────┘ └───┬────┘ └───┬────┘
#     │          │          │
#     ▼          ▼          ▼
# ┌──────────────────────────────────────┐
# │       MongoDB Replica Set            │
# └──────────────────────────────────────┘

MQTT Bridge: High Availability

Problem: Both bridges get ALL messages → duplicates in RabbitMQ
Solution 1: Leader election (MongoDB-based lock or Redis)
Solution 2: Partition topics across bridges (bridge-1: sensor/#, bridge-2: control/#)

Leader Election with MongoDB

package com.example.mqttbridge.ha;

import org.springframework.data.mongodb.core.FindAndModifyOptions;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.time.Instant;

@Component
public class MongoLeaderElection {

    private final MongoTemplate mongo;
    private final String instanceId = System.getenv("HOSTNAME");
    private volatile boolean isLeader = false;

    public MongoLeaderElection(MongoTemplate mongo) {
        this.mongo = mongo;
    }

    @Scheduled(fixedDelay = 5_000) // heartbeat every 5s
    public void tryAcquire() {
        Instant now = Instant.now();
        Instant staleThreshold = now.minusSeconds(15); // 3× heartbeat

        // Try to become leader (or renew)
        var result = mongo.findAndModify(
            Query.query(new Criteria().orOperator(
                Criteria.where("_id").is("mqtt-bridge-leader")
                    .and("leaderId").is(instanceId),
                Criteria.where("_id").is("mqtt-bridge-leader")
                    .and("lastHeartbeat").lt(staleThreshold)
            )),
            new Update()
                .set("leaderId", instanceId)
                .set("lastHeartbeat", now),
            FindAndModifyOptions.options().upsert(true).returnNew(true),
            org.bson.Document.class,
            "leader_election"
        );

        isLeader = result != null
            && instanceId.equals(result.getString("leaderId"));
    }

    public boolean isLeader() { return isLeader; }
}

RabbitMQ Workers: Auto-scaling

# Kubernetes HPA based on RabbitMQ queue depth
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: mqtt-worker-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: mqtt-worker
  minReplicas: 2
  maxReplicas: 20
  metrics:
    - type: External
      external:
        metric:
          name: rabbitmq_queue_messages
          selector:
            matchLabels:
              queue: mqtt.messages.processing
        target:
          type: AverageValue
          averageValue: "1000"   # scale up when > 1k msgs per pod

12. Cloud Foundry: HA Bridge with Leader Election

The CF Challenge: In Cloud Foundry you run multiple app instances for HA, but with MQTT 3 every instance subscribes and gets ALL messages → duplicates. Solution: Leader election — only the leader subscribes to MQTT, standby instances take over within seconds if the leader dies.

Architecture in Cloud Foundry

# CF Deployment: 2-3 bridge instances, only leader subscribes to MQTT
#
# ┌─────────────────────────────────────────────────────────┐
# │                  Cloud Foundry                          │
# │                                                         │
# │  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐ │
# │  │ bridge/0     │  │ bridge/1     │  │ bridge/2     │ │
# │  │ ★ LEADER     │  │   STANDBY    │  │   STANDBY    │ │
# │  │ MQTT ✓       │  │ MQTT ✗       │  │ MQTT ✗       │ │
# │  │ → RabbitMQ   │  │ (watching)   │  │ (watching)   │ │
# │  └──────┬───────┘  └──────┬───────┘  └──────┬───────┘ │
# │         │                  │                  │         │
# │         └────────┬─────────┴──────────┬───────┘         │
# │                  ▼                    ▼                  │
# │           ┌────────────┐      ┌────────────┐           │
# │           │ MongoDB    │      │ RabbitMQ   │           │
# │           │ (lock doc) │      │ (messages) │           │
# │           └────────────┘      └────────────┘           │
# └─────────────────────────────────────────────────────────┘
#
# Failover: Leader dies → CF restarts instance
#           Meanwhile standby detects stale lock (30s)
#           Standby promotes to leader, subscribes to MQTT
#           Downtime: ~30-40 seconds max

CF-Aware Instance Identification

package com.example.mqttbridge.cf;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.stereotype.Component;

import java.net.InetAddress;

/**
 * Extracts CF instance identity from VCAP_APPLICATION.
 * Works in CF, Diego cells, and locally (fallback to hostname).
 */
@Component
public class CfInstanceIdentity {

    private final String instanceId;
    private final int instanceIndex;
    private final String appName;

    public CfInstanceIdentity() {
        String vcap = System.getenv("VCAP_APPLICATION");
        if (vcap != null && !vcap.isEmpty()) {
            try {
                ObjectMapper mapper = new ObjectMapper();
                JsonNode node = mapper.readTree(vcap);
                this.instanceId = node.path("instance_id").asText("unknown");
                this.instanceIndex = node.path("instance_index").asInt(0);
                this.appName = node.path("application_name").asText("mqtt-bridge");
            } catch (Exception e) {
                throw new RuntimeException("Failed to parse VCAP_APPLICATION", e);
            }
        } else {
            // Local development fallback
            String hostname;
            try { hostname = InetAddress.getLocalHost().getHostName(); }
            catch (Exception e) { hostname = "local-" + ProcessHandle.current().pid(); }
            this.instanceId = hostname;
            this.instanceIndex = 0;
            this.appName = "mqtt-bridge";
        }
    }

    public String getInstanceId()  { return instanceId; }
    public int getInstanceIndex()  { return instanceIndex; }
    public String getAppName()     { return appName; }
    public String getUniqueId()    { return appName + "/" + instanceIndex + "/" + instanceId; }

    /** MQTT clientId must be unique per instance! */
    public String getMqttClientId() {
        return "bridge-" + appName + "-" + instanceIndex;
    }
}

MongoDB Leader Election (CF-optimized)

package com.example.mqttbridge.cf;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.mongodb.core.FindAndModifyOptions;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import jakarta.annotation.PreDestroy;
import java.time.Instant;

/**
 * MongoDB-based leader election for CF.
 * 
 * Lock document in "leader_election" collection:
 * {
 *   "_id": "mqtt-bridge-leader",
 *   "leaderId": "mqtt-bridge/0/abc-123",
 *   "lastHeartbeat": ISODate("2026-03-27T..."),
 *   "acquiredAt": ISODate("2026-03-27T...")
 * }
 * 
 * Rules:
 * - Heartbeat every 5s (sweet spot: fast failover, low split-brain risk)
 * - Lock expires after 15s (3 missed heartbeats)
 * - On leader loss: MQTT connection torn down
 * - On leader gain: MQTT connection established
 */
@Component
public class CfLeaderElection {

    private static final Logger log = LoggerFactory.getLogger(CfLeaderElection.class);
    private static final String LOCK_ID = "mqtt-bridge-leader";
    private static final long LOCK_TTL_SECONDS = 15; // 3× heartbeat interval

    private final MongoTemplate mongo;
    private final CfInstanceIdentity identity;
    private final MqttConnectionManager mqttManager;
    private volatile boolean isLeader = false;
    private volatile boolean wasLeader = false;

    public CfLeaderElection(MongoTemplate mongo,
                            CfInstanceIdentity identity,
                            MqttConnectionManager mqttManager) {
        this.mongo = mongo;
        this.identity = identity;
        this.mqttManager = mqttManager;
    }

    @Scheduled(fixedDelay = 5_000) // every 5s — sweet spot for failover vs stability
    public void heartbeat() {
        try {
            Instant now = Instant.now();
            Instant staleThreshold = now.minusSeconds(LOCK_TTL_SECONDS);
            String myId = identity.getUniqueId();

            // Attempt to acquire or renew the lock
            var result = mongo.findAndModify(
                Query.query(new Criteria().orOperator(
                    // I already hold the lock → renew
                    Criteria.where("_id").is(LOCK_ID)
                        .and("leaderId").is(myId),
                    // Lock is stale → take over
                    Criteria.where("_id").is(LOCK_ID)
                        .and("lastHeartbeat").lt(staleThreshold)
                )),
                new Update()
                    .set("leaderId", myId)
                    .set("lastHeartbeat", now)
                    .setOnInsert("acquiredAt", now),
                FindAndModifyOptions.options().upsert(true).returnNew(true),
                org.bson.Document.class,
                "leader_election"
            );

            isLeader = result != null && myId.equals(result.getString("leaderId"));

            // State transition: became leader
            if (isLeader && !wasLeader) {
                log.info("🟢 LEADER ACQUIRED — instance {} starting MQTT subscription",
                    identity.getInstanceIndex());
                mqttManager.connect();
            }

            // State transition: lost leadership
            if (!isLeader && wasLeader) {
                log.warn("🔴 LEADER LOST — instance {} stopping MQTT subscription",
                    identity.getInstanceIndex());
                mqttManager.disconnect();
            }

            wasLeader = isLeader;

        } catch (Exception e) {
            log.error("Leader election heartbeat failed", e);
            // On error: if we were leader, stay leader (optimistic)
            // Lock will expire naturally if we're truly dead
        }
    }

    @PreDestroy
    public void onShutdown() {
        // Graceful shutdown: release lock so standby takes over faster
        if (isLeader) {
            try {
                log.info("🔄 Releasing leader lock on shutdown");
                mongo.remove(
                    Query.query(Criteria.where("_id").is(LOCK_ID)
                        .and("leaderId").is(identity.getUniqueId())),
                    "leader_election"
                );
                mqttManager.disconnect();
            } catch (Exception e) {
                log.warn("Failed to release leader lock cleanly", e);
            }
        }
    }

    public boolean isLeader() { return isLeader; }
}

MQTT Connection Manager (Leader-Aware)

package com.example.mqttbridge.cf;

import org.eclipse.paho.client.mqttv3.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

/**
 * Manages MQTT connection lifecycle.
 * Only connects when this instance is the leader.
 * Forwards messages to RabbitMQ.
 */
@Component
public class MqttConnectionManager {

    private static final Logger log = LoggerFactory.getLogger(MqttConnectionManager.class);

    @Value("${mqtt.broker-url}") private String brokerUrl;
    @Value("${mqtt.qos:1}")      private int qos;
    @Value("${mqtt.username:}")   private String username;
    @Value("${mqtt.password:}")   private String password;
    @Value("#{${mqtt.topics}}") private String[] topics; // SpEL to parse YAML list

    private final CfInstanceIdentity identity;
    private final RabbitTemplate rabbit;
    private MqttClient client;
    private volatile boolean connected = false;

    public MqttConnectionManager(CfInstanceIdentity identity,
                                  RabbitTemplate rabbit) {
        this.identity = identity;
        this.rabbit = rabbit;
    }

    public synchronized void connect() {
        if (connected) return;
        try {
            String clientId = identity.getMqttClientId();
            log.info("Connecting to MQTT broker {} as {}", brokerUrl, clientId);

            client = new MqttClient(brokerUrl, clientId);
            MqttConnectOptions opts = new MqttConnectOptions();
            opts.setCleanSession(true); // Important: clean session so we don't get stale messages
            opts.setAutomaticReconnect(false); // WE manage reconnection via leader election
            opts.setConnectionTimeout(10);
            opts.setKeepAliveInterval(30);
            if (username != null && !username.isEmpty()) {
                opts.setUserName(username);
                opts.setPassword(password.toCharArray());
            }

            client.setCallback(new MqttCallback() {
                @Override
                public void connectionLost(Throwable cause) {
                    log.warn("MQTT connection lost: {}", cause.getMessage());
                    connected = false;
                    // Leader election will handle reconnection
                }
                @Override
                public void messageArrived(String topic, MqttMessage message) {
                    forwardToRabbit(topic, message);
                }
                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {}
            });

            client.connect(opts);

            // Subscribe to all configured topics
            int[] qosArr = new int[topics.length];
            java.util.Arrays.fill(qosArr, qos);
            client.subscribe(topics, qosArr);

            connected = true;
            log.info("✅ MQTT connected and subscribed to {} topics", topics.length);

        } catch (MqttException e) {
            log.error("Failed to connect to MQTT", e);
            connected = false;
        }
    }

    public synchronized void disconnect() {
        if (client != null && client.isConnected()) {
            try {
                client.disconnect(5000);
                log.info("MQTT disconnected");
            } catch (MqttException e) {
                log.warn("Error disconnecting MQTT", e);
            }
        }
        connected = false;
    }

    private void forwardToRabbit(String topic, MqttMessage message) {
        try {
            String routingKey = topic.replace('/', '.');
            rabbit.convertAndSend(
                "mqtt.messages",  // exchange
                routingKey,
                message.getPayload(),
                msg -> {
                    msg.getMessageProperties().setHeader("mqtt-topic", topic);
                    msg.getMessageProperties().setHeader("mqtt-qos", message.getQos());
                    msg.getMessageProperties().setHeader("bridge-instance",
                        identity.getUniqueId());
                    msg.getMessageProperties().setDeliveryMode(
                        org.springframework.amqp.core.MessageDeliveryMode.PERSISTENT);
                    return msg;
                }
            );
        } catch (Exception e) {
            log.error("Failed to forward MQTT message to RabbitMQ: topic={}", topic, e);
        }
    }

    public boolean isConnected() { return connected; }
}

CF manifest.yml

# manifest.yml — Deploy bridge with 2 instances for HA
applications:
  - name: mqtt-bridge
    memory: 512M
    instances: 2  # Only leader subscribes, standby is hot-spare
    path: target/mqtt-bridge-0.0.1.jar
    buildpacks:
      - java_buildpack
    services:
      - mongodb-service     # For leader election + data
      - rabbitmq-service    # For message distribution
    env:
      SPRING_PROFILES_ACTIVE: bridge,cloud
      MQTT_BROKER_URL: tcp://mqtt-broker.internal:1883
      MQTT_TOPICS: "sensor/#,device/#,control/#"
      JBP_CONFIG_OPEN_JDK_JRE: '{ jre: { version: 21.+ } }'
    health-check-type: http
    health-check-http-endpoint: /actuator/health

  - name: mqtt-worker
    memory: 768M
    instances: 3  # Scale freely — RabbitMQ distributes messages
    path: target/mqtt-bridge-0.0.1.jar
    buildpacks:
      - java_buildpack
    services:
      - mongodb-service
      - rabbitmq-service
    env:
      SPRING_PROFILES_ACTIVE: worker,cloud
      JBP_CONFIG_OPEN_JDK_JRE: '{ jre: { version: 21.+ } }'
    health-check-type: http
    health-check-http-endpoint: /actuator/health

CF Service Bindings (VCAP_SERVICES)

package com.example.mqttbridge.cf;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Profile;

/**
 * Cloud Foundry auto-configures MongoDB and RabbitMQ from VCAP_SERVICES.
 * Spring Cloud Connectors or java-cfenv handles this automatically.
 * 
 * Add to build.gradle / pom.xml:
 *   implementation 'io.pivotal.cfenv:java-cfenv-boot:3.1.5'
 * 
 * This auto-detects bound services and sets:
 *   spring.data.mongodb.uri=<from VCAP>
 *   spring.rabbitmq.host/port/username/password=<from VCAP>
 * 
 * No manual configuration needed in application.yml for CF!
 */
@Configuration
@Profile("cloud")
public class CfServiceConfig {

    // java-cfenv-boot auto-configures everything.
    // This class exists for documentation and custom overrides.

    // If you need to customize the MongoDB database name:
    // @Bean
    // public CfMongoDbCustomizer mongoCustomizer() {
    //     return uri -> uri + "&retryWrites=false";
    // }
}

Health Endpoint with Leader Status

package com.example.mqttbridge.cf;

import org.springframework.boot.actuate.health.Health;
import org.springframework.boot.actuate.health.HealthIndicator;
import org.springframework.stereotype.Component;

@Component
public class BridgeHealthIndicator implements HealthIndicator {

    private final CfLeaderElection leaderElection;
    private final MqttConnectionManager mqttManager;
    private final CfInstanceIdentity identity;

    public BridgeHealthIndicator(CfLeaderElection leaderElection,
                                  MqttConnectionManager mqttManager,
                                  CfInstanceIdentity identity) {
        this.leaderElection = leaderElection;
        this.mqttManager = mqttManager;
        this.identity = identity;
    }

    @Override
    public Health health() {
        Health.Builder builder = Health.up();
        builder.withDetail("instanceIndex", identity.getInstanceIndex());
        builder.withDetail("instanceId", identity.getInstanceId());
        builder.withDetail("role", leaderElection.isLeader() ? "LEADER" : "STANDBY");
        builder.withDetail("mqttConnected", mqttManager.isConnected());

        // Leader MUST be connected to MQTT
        if (leaderElection.isLeader() && !mqttManager.isConnected()) {
            return builder.down()
                .withDetail("reason", "Leader but MQTT disconnected")
                .build();
        }

        return builder.build();
    }
}

// Response example:
// GET /actuator/health
// {
//   "status": "UP",
//   "components": {
//     "bridge": {
//       "status": "UP",
//       "details": {
//         "instanceIndex": 0,
//         "role": "LEADER",
//         "mqttConnected": true
//       }
//     }
//   }
// }

Auto-Scaling Workers in CF

# CF Autoscaler policy for workers
# cf create-autoscaling-policy mqtt-worker policy.json

{
  "instance_min_count": 2,
  "instance_max_count": 10,
  "scaling_rules": [
    {
      "metric_type": "custom_metric",
      "metric": "rabbitmq_queue_depth",
      "threshold": 5000,
      "operator": ">=",
      "adjustment": "+2",
      "cool_down_secs": 120
    },
    {
      "metric_type": "custom_metric", 
      "metric": "rabbitmq_queue_depth",
      "threshold": 500,
      "operator": "<",
      "adjustment": "-1",
      "cool_down_secs": 300
    },
    {
      "metric_type": "memoryutil",
      "threshold": 80,
      "operator": ">=",
      "adjustment": "+1",
      "cool_down_secs": 120
    }
  ]
}

# Deploy with cf CLI:
cf push -f manifest.yml
cf bind-service mqtt-worker rabbitmq-service
cf bind-service mqtt-worker mongodb-service
cf restage mqtt-worker

# Scale bridge (HA only, not load balancing):
cf scale mqtt-bridge -i 2

# Scale workers (actual processing power):
cf scale mqtt-worker -i 5

⚡ Failover Timeline (5s heartbeat / 15s TTL)

T+0s Leader (instance/0) crashes or cf restart

T+5s Standby (instance/1) heartbeat — lock not yet stale

T+10s Standby heartbeat — lock not yet stale (15s TTL)

T+15s Standby heartbeat — lock IS stale → ACQUIRES LEADERSHIP

T+16s Standby connects to MQTT, subscribes to all topics

T+17s ✅ Messages flowing again. Gap: ~15s

(MQTT QoS 1 + broker retention covers the gap)

T+45s CF Diego restarts instance/0 → comes up as STANDBY

⚠️ Tuning Guide: Heartbeat vs. Split-Brain Risk

Setting	Conservative	★ Recommended	Aggressive	Extreme
Heartbeat	10s	5s	3s	1s
TTL (3× HB)	30s	15s	10s	3s
Failover	~30s	~15s	~10s	~3s
MongoDB Writes/min	6	12	20	60
Split-Brain Risk	🟢 Very Low	🟢 Low	🟡 Medium	🔴 High

🧠 Split-Brain: When Two Leaders Exist Simultaneously

Cause: GC pause (2-5s) + slow MongoDB roundtrip → leader misses heartbeat deadline → standby takes over → original leader resumes, thinks it's still leader
Effect: Both instances subscribe to MQTT → every message forwarded TWICE to RabbitMQ → duplicate processing
Why 5s is safe: JVM GC pauses rarely exceed 500ms with G1/ZGC. A 5s heartbeat tolerates occasional 2-3s spikes. At 1s heartbeat, even a minor GC causes a false failover
Mitigation 1 — Fencing Token: Include a monotonic counter in the lock. Bridge stamps each RabbitMQ message with the token. Workers ignore messages from old tokens
Mitigation 2 — RabbitMQ Dedup: Set x-deduplication-header on messages (MQTT messageId + topic). Requires rabbitmq_message_deduplication plugin
Mitigation 3 — Idempotent Consumers: Workers use UPSERT by messageId in MongoDB — duplicates are harmless

// Fencing Token: add monotonic version to the lock document
// On acquire, increment version. Stamp every RabbitMQ message with it.
// Workers track highest seen token per bridge — reject stale tokens.

// In CfLeaderElection.heartbeat():
var result = mongo.findAndModify(
    query,
    new Update()
        .set("leaderId", myId)
        .set("lastHeartbeat", now)
        .inc("fencingToken", 1),  // ← monotonic counter
    FindAndModifyOptions.options().upsert(true).returnNew(true),
    org.bson.Document.class,
    "leader_election"
);
long token = result.getLong("fencingToken");

// In MqttConnectionManager.forwardToRabbit():
msg.getMessageProperties().setHeader("fencing-token", currentFencingToken);

// In Worker consumer:
long token = (long) msg.getMessageProperties().getHeader("fencing-token");
if (token < lastSeenToken) {
    log.warn("Stale message from old leader (token {}), skipping", token);
    channel.basicAck(tag, false); // ack but don't process
    return;
}
lastSeenToken = token;

💡 Tips for CF Deployment

MQTT clientId: Must be unique per instance. Use bridge-appname-instanceIndex from VCAP_APPLICATION
Clean Session = true: Critical! Otherwise a reconnecting standby gets a backlog of queued messages from the broker
QoS 1: Use QoS 1 on both subscribe and publish. Broker retains messages for 30s gap during failover
No automatic reconnect: Let leader election manage the MQTT lifecycle. Paho auto-reconnect would fight with leader election
Graceful shutdown: @PreDestroy releases the lock immediately — standby takes over in 10s instead of 30s
Bridge instances = 2: More than 2 is waste. Only 1 is active, rest are standby
Worker instances = scale freely: 2 minimum for HA, scale up based on RabbitMQ queue depth

13. Complete Reference Architecture

Complete application.yml (Bridge Profile)

spring:
  profiles:
    active: bridge

---
spring:
  config:
    activate:
      on-profile: bridge

mqtt:
  broker-url: tcp://${MQTT_HOST:localhost}:1883
  client-id: bridge-${HOSTNAME:local}
  username: ${MQTT_USER:}
  password: ${MQTT_PASS:}
  qos: 1
  clean-session: true
  topics:
    - sensor/#
    - device/#
    - control/#

spring.rabbitmq:
  host: ${RABBITMQ_HOST:localhost}
  port: 5672
  username: ${RABBITMQ_USER:guest}
  password: ${RABBITMQ_PASS:guest}
  publisher-confirm-type: correlated
  publisher-returns: true

spring.data.mongodb:
  uri: mongodb://${MONGO_USER:root}:${MONGO_PASS:password}@${MONGO_HOST:localhost}:27017/mqtt_data?authSource=admin

management:
  endpoints.web.exposure.include: health,info,metrics,prometheus
  metrics.tags.application: mqtt-bridge
  endpoint.health.show-details: always

---
spring:
  config:
    activate:
      on-profile: worker

# Worker profile: no MQTT config, only RabbitMQ + MongoDB
spring.rabbitmq:
  host: ${RABBITMQ_HOST:localhost}
  port: 5672
  username: ${RABBITMQ_USER:guest}
  password: ${RABBITMQ_PASS:guest}
  listener:
    simple:
      prefetch: 10
      concurrency: 5
      max-concurrency: 20
      acknowledge-mode: manual
      default-requeue-rejected: false
      retry:
        enabled: true
        initial-interval: 1000
        max-interval: 30000
        multiplier: 2.0
        max-attempts: 3

spring.data.mongodb:
  uri: mongodb://${MONGO_USER:root}:${MONGO_PASS:password}@${MONGO_HOST:localhost}:27017/mqtt_data?authSource=admin

management:
  endpoints.web.exposure.include: health,info,metrics,prometheus
  metrics.tags.application: mqtt-worker
  endpoint.health.show-details: always

Docker Compose (Local Development)

version: "3.8"
services:
  mosquitto:
    image: eclipse-mosquitto:2
    ports: ["1883:1883"]
    volumes:
      - ./mosquitto.conf:/mosquitto/config/mosquitto.conf

  rabbitmq:
    image: rabbitmq:3-management
    ports: ["5672:5672", "15672:15672"]
    environment:
      RABBITMQ_DEFAULT_USER: guest
      RABBITMQ_DEFAULT_PASS: guest

  mongodb:
    image: mongo:7
    ports: ["27017:27017"]
    environment:
      MONGO_INITDB_ROOT_USERNAME: root
      MONGO_INITDB_ROOT_PASSWORD: password

  bridge:
    build: .
    environment:
      SPRING_PROFILES_ACTIVE: bridge
      MQTT_HOST: mosquitto
      RABBITMQ_HOST: rabbitmq
      MONGO_HOST: mongodb
    depends_on: [mosquitto, rabbitmq, mongodb]

  worker:
    build: .
    environment:
      SPRING_PROFILES_ACTIVE: worker
      RABBITMQ_HOST: rabbitmq
      MONGO_HOST: mongodb
    depends_on: [rabbitmq, mongodb]
    deploy:
      replicas: 3

mosquitto.conf (Allow anonymous for dev)

listener 1883
allow_anonymous true
persistence true
persistence_location /mosquitto/data/

Production Checklist

Area	Checklist
MQTT	TLS enabled, unique clientIds, QoS 1+, auth configured
RabbitMQ	Durable queues, DLQ configured, mirrored queues (HA), TLS
MongoDB	Replica set, auth enabled, TTL indexes, connection pooling
Bridge	Single instance OR leader election, buffer size tuned, metrics exposed
Workers	Prefetch tuned, concurrency set, retry + DLQ, graceful shutdown
Monitoring	Prometheus scraping, Grafana dashboards, alerts on queue depth + drops
Security	Secrets in env vars / vault, network policies, mTLS between services

When to consider alternatives

Upgrade to MQTT 5: When your broker supports it and all clients can upgrade — shared subscriptions solve this natively
Switch to Kafka: When you need > 100k msg/sec, event replay, or event sourcing patterns
Keep this architecture: When you need simplicity, reliability, and your message rate is < 100k/sec — RabbitMQ handles this with ease