MQTT with Spring Boot and Spring Integration

Spring Boot has no MQTT client of its own, but Spring Integration does: the spring-integration-mqtt module wraps Eclipse Paho in channel adapters, so incoming MQTT messages become Spring Message objects on a channel and anything you send to an outbound channel is published to the broker. This tutorial builds a small service that subscribes, publishes through a gateway interface and can be tested end to end from the browser.

Dependencies

Spring Boot's dependency management covers the Spring Integration version, so you do not pin it yourself:

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-integration</artifactId>
</dependency>
<dependency>
  <groupId>org.springframework.integration</groupId>
  <artifactId>spring-integration-mqtt</artifactId>
</dependency>

The module brings in the Paho mqttv3 client for MQTT 3.1.1. If you want the MQTT 5 adapters, also add org.eclipse.paho:org.eclipse.paho.mqttv5.client. The examples below use Spring Boot 3 or later and Java 17.

Spring Integration is the right choice when MQTT is one of several integration points and you want channels, transformers and error handling in one model. If your service only needs a client and you want reactive streams or full MQTT 5 control, the HiveMQ MQTT Client wrapped in a plain Spring bean is a lighter alternative.

Configuration properties

Keep broker settings out of the code. A small record bound with @ConfigurationProperties keeps them typed:

mqtt:
  url: ssl://broker.connectmq.com:8883
  username: ${MQTT_USER}
  password: ${MQTT_PASS}
  client-id: orders-service-${random.uuid}
  inbound-topic: testmqtt/spring/in/#
  outbound-topic: testmqtt/spring/out

@ConfigurationProperties(prefix = "mqtt")
public record MqttProperties(String url, String username, String password,
                             String clientId, String inboundTopic, String outboundTopic) {}

Enable it with @EnableConfigurationProperties(MqttProperties.class) on a configuration class. Use ssl:// with port 8883 for TLS and tcp:// with 1883 for plain connections during local development.

The client factory

All adapters get their connections from an MqttPahoClientFactory. This is where TLS, credentials, keep alive and reconnect behaviour are configured once:

@Configuration
@EnableConfigurationProperties(MqttProperties.class)
@IntegrationComponentScan
public class MqttConfig {

    @Bean
    public MqttPahoClientFactory mqttClientFactory(MqttProperties props) {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(new String[] { props.url() });
        options.setUserName(props.username());
        options.setPassword(props.password().toCharArray());
        options.setAutomaticReconnect(true);
        options.setCleanSession(true);
        options.setKeepAliveInterval(30);
        options.setConnectionTimeout(10);

        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setConnectionOptions(options);
        return factory;
    }
}

MqttConnectOptions is the regular Paho class, so every Paho option — Last Will, custom SSLSocketFactory, MQTT version — is available. See MQTT in Java with Eclipse Paho for the details of each option.

Inbound: subscribe with a message-driven adapter

MqttPahoMessageDrivenChannelAdapter subscribes to one or more topic filters and pushes every received message to an output channel. A @ServiceActivator then handles it:

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

@Bean
public MessageProducer inbound(MqttPahoClientFactory factory, MqttProperties props) {
    MqttPahoMessageDrivenChannelAdapter adapter =
            new MqttPahoMessageDrivenChannelAdapter(props.clientId() + "-in", factory, props.inboundTopic());
    adapter.setCompletionTimeout(5000);
    adapter.setConverter(new DefaultPahoMessageConverter());
    adapter.setQos(1);
    adapter.setOutputChannel(mqttInputChannel());
    return adapter;
}

@Bean
@ServiceActivator(inputChannel = "mqttInputChannel")
public MessageHandler inboundHandler() {
    return message -> {
        String topic = message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC, String.class);
        System.out.println("received " + topic + ": " + message.getPayload());
    };
}

Useful headers on every inbound message:

The default converter turns payloads into Strings. Call converter.setPayloadAsBytes(true) if you receive binary formats such as CBOR or Protobuf. You can also add or remove topics at runtime with adapter.addTopic(...) and adapter.removeTopic(...).

Errors and slow handlers

With a DirectChannel, your handler runs on Paho's callback thread. That has two consequences worth planning for:

• An exception in the handler is sent to the adapter's error channel if you set one with adapter.setErrorChannel(...). Without it, the exception is logged and, depending on the adapter version and settings, Paho may treat the failure as a lost connection. Always catch and route errors deliberately — for example to a channel that writes failed payloads to a dead-letter topic.
• A slow handler delays the acknowledgement of QoS 1 messages and blocks every other message on that connection. If processing involves a database or an HTTP call, switch mqttInputChannel to an ExecutorChannel backed by a bounded thread pool, so the MQTT thread only hands the message over.

Be aware that handing the message to another thread means the PUBACK is sent before your processing is finished. If the service crashes in between, that message is gone. When at-least-once processing really matters, use adapter.setManualAcks(true) and acknowledge from your handler with the IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK header once the work is committed.

Outbound: publish through a gateway

MqttPahoMessageHandler publishes whatever reaches its input channel. Pair it with a @MessagingGateway so the rest of your code calls a plain Java interface:

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

@Bean
@ServiceActivator(inputChannel = "mqttOutboundChannel")
public MessageHandler mqttOutbound(MqttPahoClientFactory factory, MqttProperties props) {
    MqttPahoMessageHandler handler = new MqttPahoMessageHandler(props.clientId() + "-out", factory);
    handler.setAsync(true);
    handler.setAsyncEvents(true);
    handler.setDefaultTopic(props.outboundTopic());
    handler.setDefaultQos(1);
    return handler;
}

@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
public interface MqttGateway {

    void send(String payload);

    void send(String payload, @Header(MqttHeaders.TOPIC) String topic);

    void send(String payload, @Header(MqttHeaders.TOPIC) String topic,
              @Header(MqttHeaders.QOS) int qos, @Header(MqttHeaders.RETAINED) boolean retained);
}

Now any bean can inject MqttGateway and call gateway.send("{\"id\":42}", "testmqtt/spring/out/orders"). setAsync(true) makes the send return without waiting for the PUBACK; with setAsyncEvents(true) Spring publishes an MqttMessageDeliveredEvent when the broker confirms delivery.

The Java DSL alternative

If you prefer flows to separate beans, the Java DSL expresses the same inbound pipeline in a few lines, including a transformation step:

@Bean
public IntegrationFlow temperatureFlow(MqttPahoClientFactory factory, MqttProperties props) {
    return IntegrationFlow
            .from(new MqttPahoMessageDrivenChannelAdapter(
                    props.clientId() + "-temp", factory, "testmqtt/spring/sensors/+/temp"))
            .<String, Double>transform(Double::valueOf)
            .filter((Double t) -> t > 30.0)
            .handle(m -> System.out.println("too hot: " + m.getPayload()))
            .get();
}

Check which topics a filter like testmqtt/spring/sensors/+/temp matches with the topic matcher before you deploy it.

Connection events and MQTT 5

Spring Integration publishes application events for the MQTT lifecycle. Listening to them is the simplest way to log or alert on connection problems:

@Component
public class MqttEvents {

    @EventListener
    public void onFailure(MqttConnectionFailedEvent event) {
        System.err.println("MQTT connection failed: " + event.getCause());
    }

    @EventListener
    public void onSubscribed(MqttSubscribedEvent event) {
        System.out.println(event.getMessage());
    }
}

For MQTT 5, swap the classes: Mqttv5PahoMessageDrivenChannelAdapter and Mqttv5PahoMessageHandler take a Paho v5 MqttConnectionOptions instead of a factory, and map MQTT 5 user properties to and from message headers. The channels, gateway and handlers stay the same.

Scaling out with shared subscriptions

With a normal subscription, every instance of your service receives every message — three pods means each order is processed three times. MQTT 5 shared subscriptions (also supported as an extension by many brokers on MQTT 3.1.1 connections) fix this: subscribe to $share/orders/testmqtt/spring/in/# and the broker distributes messages across all members of the orders group. Only the topic string in application.yml changes. Read MQTT shared subscriptions for the delivery rules.

Test it end to end

1. Start the application and watch the log for the subscription.
2. Open the online MQTT client, connect to the same broker and publish hello to testmqtt/spring/in/demo. Your service should log received testmqtt/spring/in/demo: hello.
3. Subscribe in the browser to testmqtt/spring/out/#, call an endpoint that uses MqttGateway, and check the message appears with QoS 1.
4. Stop the broker connection (for example by disabling your network briefly) and confirm that a MqttConnectionFailedEvent is logged and the adapter reconnects.

Watch your Spring Boot service's traffic

If messages do not arrive, the Paho exception in the log usually says why — Paho Java errors lists the common ones, and the reason codes table decodes broker refusals.

Spring Integration turns MQTT into just another channel in your Spring Boot application: configure the client factory once, connect inbound adapters to service activators, publish through a gateway, and keep client IDs unique per connection. Once the browser test passes, the same configuration runs in production with only the credentials changed.

Frequently asked questions