Publish Message will send twice

[fuuhoo]

In what version(s) of Spring Integration are you seeing this issue?

spring-integration-mqtt v5.5.20

MQTT Client: org.eclipse.paho.mqttv5.client v1.2.5

Describe the bug

emqx5.6.1当作broker，并且限制频率为100/s，使用spring boot2.7，在循环发送的时候，使用mqttv5协议，会导致发送到1000条左右的时候publish meaasge发送两次的情况，导致broker返回145代码

When using emqx5.6.1 as broker and limiting the frequency to 100/s, using spring boot2.7, and using the mqttv5 protocol in a loop, the publish message will be sent twice when about 1000 messages are sent, causing the broker to return code 145.

To Reproduce

细节查看这个问题：

See this for details:

emqx/emqx#15055 (comment)

Expected behavior

A clear and concise description of what you expected to happen.

Sample

A link to a GitHub repository with a minimal, reproducible sample.

Reports that include a sample will take priority over reports that do not.
At times, we may require a sample, so it is good to try and include a sample up front.

[artembilan]

That version of Spring Integration is out of support. Either way, any chances to have some code from you to reproduce on our side?
Perhaps even currently version in support is failing , too…

[fuuhoo]

That version of Spring Integration is out of support. Either way, any chances to have some code from you to reproduce on our side? Perhaps even currently version in support is failing , too…

5.5.20是否是在java8和springboot2环境下的最新版本?
Is 5.5.20 the latest version available in Java 8 and Spring Boot 2 environments?

这是测试代码

This is the test code

package com.example.mqttdemo.demo;

import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.integration.mqtt.outbound.Mqttv5PahoMessageHandler;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;

@SpringBootTest
class DemoApplicationTests {

    @Autowired
    private DefaultListableBeanFactory beanFactory;
    @Test
    void test4() {
        MqttConnectionOptions options = new MqttConnectionOptions();
        options.setServerURIs(new String[]{"tcp://192.168.128.114:1883"});
        options.setUserName("swzs");
        options.setPassword("swzs".getBytes());
        options.setCleanStart(false);
        options.setAutomaticReconnect(true);
        options.setSessionExpiryInterval(1800L);
        String clientId = "test222";
        BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(Mqttv5PahoMessageHandler.class);
        builder.addConstructorArgValue(options);
        builder.addConstructorArgValue(clientId);
        builder.addPropertyValue("async", false);
        AbstractBeanDefinition beanDefinition = builder.getBeanDefinition();
        beanFactory.registerBeanDefinition("output",beanDefinition );
        Mqttv5PahoMessageHandler messageHandler = beanFactory.getBean("output", Mqttv5PahoMessageHandler.class);

        for (int i = 1; i <= 3000; i++) {
            String msg = "=================================test data indexxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx=====>" + i;
            String topic = "testRate";
            Message<String> mqttMessage = MessageBuilder.withPayload(msg).setHeader(MqttHeaders.TOPIC, topic)
                    .setHeader(MqttHeaders.QOS, 2).build();
            System.out.println("msg====>"+msg);
            try {

                messageHandler.handleMessage(mqttMessage);
            }catch (Exception ex){
                ex.printStackTrace();
            }
        }
    }

}

https://github.com/fuuhoo/mqttdemo

mqtt broker use EMQX Version: 5.6.1 (Open Source)

config

## NOTE:
## This config file overrides data/configs/cluster.hocon,
## and is merged with environment variables which start with 'EMQX_' prefix.
##
## Config changes made from EMQX dashboard UI, management HTTP API, or CLI
## are stored in data/configs/cluster.hocon.
## To avoid confusion, please do not store the same configs in both files.
##
## See https://www.emqx.io/docs/en/latest/configuration/configuration.html for more details.
## Configuration full example can be found in etc/examples

node {
  name = "emqx@127.0.0.1"
  cookie = "emqxsecretcookie11111"
  data_dir = "data"
}

# cluster {
#   name = emqxcl
#   discovery_strategy = manual
# }

## EMQX provides support for two primary log handlers: `file` and `console`, with an additional `audit` handler specifically designed to always direct logs to files.
## The system's default log handling behavior can be configured via the environment variable `EMQX_DEFAULT_LOG_HANDLER`, which accepts the following settings:
##
##   - `file`: Directs log output exclusively to files.
##   - `console`: Channels log output solely to the console.
##
## It's noteworthy that `EMQX_DEFAULT_LOG_HANDLER` is set to `file` when EMQX is initiated via systemd `emqx.service` file.
## In scenarios outside systemd initiation, `console` serves as the default log handler.

## Read more about configs here: https://www.emqx.io/docs/en/latest/configuration/logs.html

log {
    # file {
    #     level = warning
    # }
    # console {
    #     level = warning
    # }
}
dashboard {
    listeners.http {
        bind = 18083
    }
}

listeners.tcp.default {

    # ## Connections per second per listener
    # max_conn_rate = "1000/s"

    ## Incoming messages per second per client
    messages_rate = "100/s"

    # ## Incoming message size in bytes per second per client
    # bytes_rate = "1MB/s"
}

thanks！！！

[artembilan]

The conclusion about versions is correct. But again: we don’t support them as Open Source any more.
Feel free to consider a commercial support : https://spring.io/support!

Is there any chances that you can confirm with your demo that it fails same way even on latest supported version?

[fuuhoo]

The conclusion about versions is correct. But again: we don’t support them as Open Source any more. Feel free to consider a commercial support : https://spring.io/support!

Is there any chances that you can confirm with your demo that it fails same way even on latest supported version?

在springboot:3.4.4,spring-integration-mqtt:6.4.3,org.eclipse.paho.mqttv5.client:1.2.5的环境下，仍然会出现这个问题，我不确定是因为emqx broker的问题还是因为spring-integration-mqtt的问题。我直接使用org.eclipse.paho.mqttv5.client.MqttClient客户端发送时没有问题的（test2），但是我使用spring-integration-mqtt发送的时候，就会出现问题（test1），猜测可能spring-integration-mqtt:非异步发送也是用的异步客户端有关系？

In the environment of springboot:3.4.4,
spring-integration-mqtt:6.4.3,
org.eclipse.paho.mqttv5.client:1.2.5,
this problem still occurs. I am not sure whether it is due to the problem of mqtt broker（emqx5.6.1 opensource） or spring-integration-mqtt , or network problem?. I have no problem when I use org.eclipse.paho.mqttv5.client.MqttClient to send directly (test2), but when I use spring-integration-mqtt to send, there will be a problem (test1). I guess it may be related to spring-integration-mqtt: non-asynchronous sending also uses asynchronous client?

这个是用wireshark的抓包数据
This is the packet capture data of wireshark

v5-25041902.zip

这个是相关的demo代码
This is the demo code
https://github.com/fuuhoo/mqttdemoforSpringboot3

[artembilan]

Thanks for the sample!

I'm not familiar with that emqx and cannot install it on my laptop., however I was able to run your application against eclipse-mosquitto Docker image using Testcontainers.

Neither of tests have failed for me, but I've noticed that the first one (based on Spring Integration) is much faster:

.

Both tests shows successful publishing of those 1000 messages.
I see that you are trying to mimic a blocking publishing via that Mqttv5PahoMessageHandler:

builder.addPropertyValue("async", false);

Even if there is a difference in wait timeout (Mqttv5PahoMessageHandler is 30 seconds, the MqttClient is -1 (meaning no timeout)), I don't think this is a problem, because the first test finishes very fast.
That might mean because MqttAsyncClient uses different threads for publishing.

I see two outcomes from this issue:

1. The second test has to be reworked to MqttAsyncClient with token.waitForCompletion(completionTimeout) to mimic whatever is there in the Mqttv5PahoMessageHandler.
   This way we will be able to confirm that the problem is a bit lower, than it raised here against Spring.

2. Since I cannot reproduce it with Mosquitto, that really might be the problem on the emqx broker side when they don't handle concurrent requests from the same client properly.

[fuuhoo]

emqx作为broker，在没有设置 messages_rate = "100/s"的时候也是没有问题的。在mosquitto作为broker的情况下也是没有问题的，我在mosquitto中没有找到关于针对发送频率的限制的设置。所以这个问题可能是和eqmx的messages_rate设置有关吧。

There is no problem when emqx is used as broker and messages_rate = "100/s" is not set. There is no problem when mosquitto is used as broker. I did not find any setting for limiting the sending frequency in mosquitto. So this problem may be related to the messages_rate setting of eqmx.

[artembilan]

Thanks for confirming!
Closed as Works as Designed.