This is an automated email from the ASF dual-hosted git repository.
Aias00 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shenyu.git
The following commit(s) were added to refs/heads/master by this push:
new 3fc3d84eab fix(mqtt): deduplicate retransmitted qos2 publishes with
inbound in-flight state (#7919) (#7938)
3fc3d84eab is described below
commit 3fc3d84eabaa32e0573eb48778065976e4be12f3
Author: konlue <[email protected]>
AuthorDate: Fri Oct 23 11:21:47 2026 +0800
fix(mqtt): deduplicate retransmitted qos2 publishes with inbound in-flight
state (#7919) (#7938)
* fix(mqtt): deduplicate retransmitted qos2 publishes with inbound
in-flight state (#7919)
* fix(mqtt): clear inbound in-flight ids on new sessions and harden dedup
tests (#7919)
---
.../org/apache/shenyu/protocol/mqtt/Connect.java | 31 ++++--
.../apache/shenyu/protocol/mqtt/Disconnect.java | 5 +-
.../org/apache/shenyu/protocol/mqtt/PubRel.java | 9 ++
.../org/apache/shenyu/protocol/mqtt/Publish.java | 61 ++++++++++--
.../mqtt/repositories/InflightRepository.java | 80 +++++++++++++++
.../shenyu/protocol/mqtt/ClientIdTakeoverTest.java | 3 +
.../protocol/mqtt/ConnectSessionPresentTest.java | 2 +
.../apache/shenyu/protocol/mqtt/ConnectTest.java | 2 +
.../protocol/mqtt/DisconnectDispatchTest.java | 2 +
.../shenyu/protocol/mqtt/MqttFactoryTest.java | 3 +
.../apache/shenyu/protocol/mqtt/PubRelTest.java | 25 +++++
.../apache/shenyu/protocol/mqtt/PublishTest.java | 109 ++++++++++++++++++++-
12 files changed, 313 insertions(+), 19 deletions(-)
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Connect.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Connect.java
index 99d1144c41..1d04db87fa 100644
---
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Connect.java
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Connect.java
@@ -27,6 +27,7 @@ import io.netty.handler.codec.mqtt.MqttVersion;
import org.apache.commons.lang3.StringUtils;
import org.apache.shenyu.common.utils.Singleton;
import org.apache.shenyu.protocol.mqtt.repositories.ChannelRepository;
+import org.apache.shenyu.protocol.mqtt.repositories.InflightRepository;
import org.apache.shenyu.protocol.mqtt.repositories.MqttSession;
import org.apache.shenyu.protocol.mqtt.repositories.SessionRepository;
import org.apache.shenyu.protocol.mqtt.repositories.SubscribeRepository;
@@ -81,13 +82,7 @@ public class Connect extends MessageType {
// record connect and take over the client id: a duplicate clientId
must
// replace the existing connection instead of coexisting with it
(#7918)
- final Channel previous =
Singleton.INST.get(ChannelRepository.class).takeover(ctx.channel(), clientId);
- if (Objects.nonNull(previous) && previous != ctx.channel()) {
- LOG.info("MQTT clientId {} is already connected, closing the
previous connection.", clientId);
- // the replaced connection's will is discarded: the client is
alive on the new connection
- Singleton.INST.get(WillRepository.class).remove(previous);
- previous.close();
- }
+ takeOverClientId(ctx.channel(), clientId);
// store will if present
if (msg.variableHeader().isWillFlag()) {
@@ -118,6 +113,11 @@ public class Connect extends MessageType {
Singleton.INST.get(SubscribeRepository.class)
.add(ctx.channel(), session.getTopicSubscriptions());
}
+ if (!sessionPresent) {
+ // a new session resets the packet id space, so stale inbound ids
must
+ // not suppress new publishes on this connection (#7919)
+ Singleton.INST.get(InflightRepository.class).remove(clientId);
+ }
MqttConnAckMessage ackMessage = MqttMessageBuilders.connAck()
.returnCode(MqttConnectReturnCode.CONNECTION_ACCEPTED)
@@ -127,6 +127,23 @@ public class Connect extends MessageType {
setConnected(ctx.channel(), true);
}
+ /**
+ * Takes over the client id for the new channel, closing the connection
that
+ * owned it before, if any.
+ *
+ * @param channel the channel of the new connection
+ * @param clientId the announced client identifier
+ */
+ private void takeOverClientId(final Channel channel, final String
clientId) {
+ final Channel previous =
Singleton.INST.get(ChannelRepository.class).takeover(channel, clientId);
+ if (Objects.nonNull(previous) && previous != channel) {
+ LOG.info("MQTT clientId {} is already connected, closing the
previous connection.", clientId);
+ // the replaced connection's will is discarded: the client is
alive on the new connection
+ Singleton.INST.get(WillRepository.class).remove(previous);
+ previous.close();
+ }
+ }
+
private void close(final ChannelHandlerContext ctx, final
MqttConnectReturnCode returnCode) {
ctx.writeAndFlush(wrong(returnCode));
ctx.close().addListener(CLOSE_ON_FAILURE);
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Disconnect.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Disconnect.java
index c73ef28e25..3d3c22ea12 100644
---
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Disconnect.java
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Disconnect.java
@@ -21,6 +21,7 @@ import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import org.apache.shenyu.common.utils.Singleton;
import org.apache.shenyu.protocol.mqtt.repositories.ChannelRepository;
+import org.apache.shenyu.protocol.mqtt.repositories.InflightRepository;
import org.apache.shenyu.protocol.mqtt.repositories.MqttSession;
import org.apache.shenyu.protocol.mqtt.repositories.SessionRepository;
import org.apache.shenyu.protocol.mqtt.repositories.SubscribeRepository;
@@ -62,8 +63,10 @@ public class Disconnect extends MessageType {
if (channelRepository.isOwner(clientId, channel)) {
MqttSession session =
Singleton.INST.get(SessionRepository.class).get(clientId);
if (Objects.nonNull(session) && session.isCleanSession()) {
- // A clean-session disconnect discards all stored session
state.
+ // A clean-session disconnect discards all stored session
state,
+ // including the inbound packet ids still being
acknowledged (#7919)
Singleton.INST.get(SessionRepository.class).remove(clientId);
+
Singleton.INST.get(InflightRepository.class).remove(clientId);
}
}
}
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/PubRel.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/PubRel.java
index 516ceb96db..023ca1fe67 100644
---
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/PubRel.java
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/PubRel.java
@@ -22,6 +22,11 @@ import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader;
import io.netty.handler.codec.mqtt.MqttQoS;
+import org.apache.shenyu.common.utils.Singleton;
+import org.apache.shenyu.protocol.mqtt.repositories.ChannelRepository;
+import org.apache.shenyu.protocol.mqtt.repositories.InflightRepository;
+
+import java.util.Objects;
import static io.netty.handler.codec.mqtt.MqttMessageType.PUBCOMP;
@@ -34,6 +39,10 @@ public class PubRel extends MessageType {
@Override
public void pubRel(final ChannelHandlerContext ctx, final MqttMessage msg)
{
MqttMessageIdVariableHeader variableHeader =
(MqttMessageIdVariableHeader) msg.variableHeader();
+ String clientId =
Singleton.INST.get(ChannelRepository.class).get(ctx.channel());
+ if (Objects.nonNull(clientId)) {
+ Singleton.INST.get(InflightRepository.class).release(clientId,
variableHeader.messageId());
+ }
MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(PUBCOMP, false,
MqttQoS.AT_MOST_ONCE, false, 0);
MqttMessage mqttPubCompMessage = new MqttMessage(mqttFixedHeader,
variableHeader);
ctx.writeAndFlush(mqttPubCompMessage);
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Publish.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Publish.java
index e5ad6fc169..5a0482b723 100644
---
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Publish.java
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Publish.java
@@ -31,6 +31,10 @@ import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttPublishVariableHeader;
import io.netty.util.ReferenceCountUtil;
import org.apache.shenyu.common.utils.Singleton;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.shenyu.protocol.mqtt.repositories.ChannelRepository;
+import org.apache.shenyu.protocol.mqtt.repositories.InflightRepository;
import org.apache.shenyu.protocol.mqtt.repositories.SubscribeRepository;
import org.apache.shenyu.protocol.mqtt.repositories.TopicRepository;
import org.apache.shenyu.protocol.mqtt.repositories.WillRepository;
@@ -49,6 +53,8 @@ import static
io.netty.handler.codec.mqtt.MqttMessageType.PUBREC;
*/
public class Publish extends MessageType {
+ private static final Logger LOG = LoggerFactory.getLogger(Publish.class);
+
@Override
public void publish(final ChannelHandlerContext ctx, final
MqttPublishMessage msg) {
if (!isConnected(ctx.channel())) {
@@ -68,25 +74,22 @@ public class Publish extends MessageType {
}
}
int packetId = msg.variableHeader().packetId();
- // The inbound message is released by MqttTransportHandler once
publish returns, retain the payload for the asynchronous send.
- payload.retain();
- CompletableFuture.runAsync(() -> {
- try {
- send(topic, payload, mqttQoS);
- } finally {
- ReferenceCountUtil.safeRelease(payload);
- }
- });
-
switch (mqttQoS.value()) {
case 0:
+ deliverAsync(topic, payload, mqttQoS);
break;
case 1:
+ deliverAsync(topic, payload, mqttQoS);
qos1(ctx, packetId);
break;
case 2:
+ // exactly-once: a retransmitted publish is answered with
PUBREC
+ // but must not be delivered to subscribers again (#7919)
+ if (registerInbound(ctx.channel(), packetId)) {
+ deliverAsync(topic, payload, mqttQoS);
+ }
qos2(ctx, packetId);
break;
default:
@@ -95,6 +98,44 @@ public class Publish extends MessageType {
}
+ /**
+ * Delivers the message to the subscribers of the topic on the common pool.
+ * The inbound message is released by MqttTransportHandler once publish
returns,
+ * retain the payload for the asynchronous send.
+ *
+ * @param topic the topic of the publish
+ * @param payload the payload of the publish
+ * @param mqttQoS the qos level of the publish
+ */
+ private void deliverAsync(final String topic, final ByteBuf payload, final
MqttQoS mqttQoS) {
+ payload.retain();
+ CompletableFuture.runAsync(() -> {
+ try {
+ send(topic, payload, mqttQoS);
+ } finally {
+ ReferenceCountUtil.safeRelease(payload);
+ }
+ });
+ }
+
+ /**
+ * Registers the inbound packet for deduplication. QoS 2 packet ids are
scoped
+ * to the connection, so without a known client the publish is delivered
as before.
+ *
+ * @param channel the channel of the publisher
+ * @param packetId the packet identifier of the inbound publish
+ * @return true when the packet was seen for the first time
+ */
+ private boolean registerInbound(final Channel channel, final int packetId)
{
+ String clientId =
Singleton.INST.get(ChannelRepository.class).get(channel);
+ if (Objects.isNull(clientId)) {
+ // dedup is keyed by the client id: without one, deliver as before
+ LOG.debug("mqtt publish on channel {} has no client id, skipping
qos2 dedup", channel.id());
+ return true;
+ }
+ return Singleton.INST.get(InflightRepository.class).register(clientId,
packetId);
+ }
+
/**
* send PUBACK to the publisher for a qos1 publish.
*/
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/repositories/InflightRepository.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/repositories/InflightRepository.java
new file mode 100644
index 0000000000..7eda991208
--- /dev/null
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/repositories/InflightRepository.java
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shenyu.protocol.mqtt.repositories;
+
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Tracks the inbound packet identifiers currently being acknowledged, so that
+ * a retransmitted (DUP) publish can be answered without re-delivering it
(#7919).
+ * Entries are released by PUBREL, and a new session (clean-session connect or
no
+ * stored session) discards them, since it resets the packet id space.
+ *
+ * <p>The state is keyed by the client id, not by the channel: session state
+ * belongs to the client id per spec, so during a duplicate-clientId takeover
the
+ * replacement channel shares — and must keep honoring — the same in-flight
ids.</p>
+ */
+public class InflightRepository implements BaseRepository<String,
Set<Integer>> {
+
+ private static final Map<String, Set<Integer>> INBOUND_INFLIGHT = new
ConcurrentHashMap<>();
+
+ @Override
+ public void add(final String clientId, final Set<Integer> packetIds) {
+ INBOUND_INFLIGHT.put(clientId, packetIds);
+ }
+
+ @Override
+ public void remove(final String clientId) {
+ INBOUND_INFLIGHT.remove(clientId);
+ }
+
+ @Override
+ public Set<Integer> get(final String clientId) {
+ return INBOUND_INFLIGHT.get(clientId);
+ }
+
+ /**
+ * Registers a packet identifier as being acknowledged.
+ *
+ * @param clientId the client that published the packet
+ * @param packetId the packet identifier of the inbound publish
+ * @return true when the packet was seen for the first time, false on a
retransmission
+ */
+ public boolean register(final String clientId, final int packetId) {
+ return INBOUND_INFLIGHT.computeIfAbsent(clientId, key ->
ConcurrentHashMap.newKeySet()).add(packetId);
+ }
+
+ /**
+ * Releases the packet identifier once the client confirms it with PUBREL.
+ * The client entry itself is dropped when its last in-flight id is
released,
+ * so ids that never publish again do not linger for the lifetime of the
JVM.
+ *
+ * @param clientId the client that published the packet
+ * @param packetId the packet identifier to release
+ */
+ public void release(final String clientId, final int packetId) {
+ Set<Integer> packetIds = INBOUND_INFLIGHT.get(clientId);
+ if (Objects.nonNull(packetIds) && packetIds.remove(packetId) &&
packetIds.isEmpty()) {
+ // two-arg remove: a concurrent register must not be evicted
+ INBOUND_INFLIGHT.remove(clientId, packetIds);
+ }
+ }
+}
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/ClientIdTakeoverTest.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/ClientIdTakeoverTest.java
index 45af4a5ece..9971d57f0c 100644
---
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/ClientIdTakeoverTest.java
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/ClientIdTakeoverTest.java
@@ -30,6 +30,7 @@ import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.handler.codec.mqtt.MqttVersion;
import org.apache.shenyu.common.utils.Singleton;
import org.apache.shenyu.protocol.mqtt.repositories.ChannelRepository;
+import org.apache.shenyu.protocol.mqtt.repositories.InflightRepository;
import org.apache.shenyu.protocol.mqtt.repositories.SessionRepository;
import org.apache.shenyu.protocol.mqtt.repositories.SubscribeRepository;
import org.apache.shenyu.protocol.mqtt.repositories.WillRepository;
@@ -86,6 +87,8 @@ public final class ClientIdTakeoverTest {
Singleton.INST.single(SessionRepository.class, sessionRepository);
Singleton.INST.single(WillRepository.class, willRepository);
Singleton.INST.single(SubscribeRepository.class, subscribeRepository);
+ // Connect reaches this repository on a new session, and Singleton.get
does not create one
+ Singleton.INST.single(InflightRepository.class, new
InflightRepository());
MqttContext mqttContext = new MqttContext();
mqttContext.setUserName(USER_NAME);
mqttContext.setPassword(PASSWORD);
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/ConnectSessionPresentTest.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/ConnectSessionPresentTest.java
index c2b57536ba..a98e4b6859 100644
---
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/ConnectSessionPresentTest.java
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/ConnectSessionPresentTest.java
@@ -34,6 +34,7 @@ import io.netty.util.Attribute;
import io.netty.util.AttributeKey;
import org.apache.shenyu.common.utils.Singleton;
import org.apache.shenyu.protocol.mqtt.repositories.ChannelRepository;
+import org.apache.shenyu.protocol.mqtt.repositories.InflightRepository;
import org.apache.shenyu.protocol.mqtt.repositories.MqttSession;
import org.apache.shenyu.protocol.mqtt.repositories.SessionRepository;
import org.apache.shenyu.protocol.mqtt.repositories.SubscribeRepository;
@@ -75,6 +76,7 @@ public class ConnectSessionPresentTest {
Singleton.INST.single(ChannelRepository.class, new
ChannelRepository());
Singleton.INST.single(SessionRepository.class, new
SessionRepository());
Singleton.INST.single(SubscribeRepository.class, new
SubscribeRepository());
+ Singleton.INST.single(InflightRepository.class, new
InflightRepository());
MqttContext mqttContext = new MqttContext();
mqttContext.setUserName("shenyu");
mqttContext.setPassword("shenyu");
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/ConnectTest.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/ConnectTest.java
index 011b759892..caf31afb35 100644
---
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/ConnectTest.java
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/ConnectTest.java
@@ -30,6 +30,7 @@ import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.handler.codec.mqtt.MqttVersion;
import org.apache.shenyu.common.utils.Singleton;
import org.apache.shenyu.protocol.mqtt.repositories.ChannelRepository;
+import org.apache.shenyu.protocol.mqtt.repositories.InflightRepository;
import org.apache.shenyu.protocol.mqtt.repositories.WillRepository;
import org.apache.shenyu.protocol.mqtt.repositories.WillRepository.WillEntry;
import org.junit.jupiter.api.AfterAll;
@@ -95,6 +96,7 @@ public final class ConnectTest {
willRepository = new WillRepository();
Singleton.INST.single(ChannelRepository.class, channelRepository);
Singleton.INST.single(WillRepository.class, willRepository);
+ Singleton.INST.single(InflightRepository.class, new
InflightRepository());
}
@AfterEach
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/DisconnectDispatchTest.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/DisconnectDispatchTest.java
index c58a59ac96..ad059d8ce3 100644
---
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/DisconnectDispatchTest.java
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/DisconnectDispatchTest.java
@@ -23,6 +23,7 @@ import io.netty.handler.codec.mqtt.MqttMessageBuilders;
import io.netty.handler.codec.mqtt.MqttQoS;
import org.apache.shenyu.common.utils.Singleton;
import org.apache.shenyu.protocol.mqtt.repositories.ChannelRepository;
+import org.apache.shenyu.protocol.mqtt.repositories.InflightRepository;
import org.apache.shenyu.protocol.mqtt.repositories.MqttSession;
import org.apache.shenyu.protocol.mqtt.repositories.SessionRepository;
import org.apache.shenyu.protocol.mqtt.repositories.SubscribeRepository;
@@ -55,6 +56,7 @@ public class DisconnectDispatchTest {
Singleton.INST.single(SessionRepository.class, new
SessionRepository());
Singleton.INST.single(SubscribeRepository.class, new
SubscribeRepository());
Singleton.INST.single(WillRepository.class, new WillRepository());
+ Singleton.INST.single(InflightRepository.class, new
InflightRepository());
ctx = mock(ChannelHandlerContext.class);
channel = mock(Channel.class);
when(ctx.channel()).thenReturn(channel);
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/MqttFactoryTest.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/MqttFactoryTest.java
index f28ff11057..2976e02a85 100644
---
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/MqttFactoryTest.java
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/MqttFactoryTest.java
@@ -40,6 +40,7 @@ import io.netty.handler.codec.mqtt.MqttUnsubscribePayload;
import io.netty.handler.codec.mqtt.MqttVersion;
import org.apache.shenyu.common.utils.Singleton;
import org.apache.shenyu.protocol.mqtt.repositories.ChannelRepository;
+import org.apache.shenyu.protocol.mqtt.repositories.InflightRepository;
import org.apache.shenyu.protocol.mqtt.repositories.SessionRepository;
import org.apache.shenyu.protocol.mqtt.repositories.SubscribeRepository;
import org.apache.shenyu.protocol.mqtt.repositories.WillRepository;
@@ -79,6 +80,8 @@ public final class MqttFactoryTest {
Singleton.INST.single(WillRepository.class, willRepository);
Singleton.INST.single(SessionRepository.class, new
SessionRepository());
Singleton.INST.single(SubscribeRepository.class, new
SubscribeRepository());
+ // Connect reaches this repository on a new session, and Singleton.get
does not create one
+ Singleton.INST.single(InflightRepository.class, new
InflightRepository());
new MqttContext().setUserName(USER_NAME);
new MqttContext().setPassword(PASSWORD);
}
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/PubRelTest.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/PubRelTest.java
index 1acd34c6e4..297e9e77ad 100644
---
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/PubRelTest.java
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/PubRelTest.java
@@ -17,18 +17,23 @@
package org.apache.shenyu.protocol.mqtt;
+import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttQoS;
+import org.apache.shenyu.common.utils.Singleton;
+import org.apache.shenyu.protocol.mqtt.repositories.ChannelRepository;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
/**
* Test cases for PubRel.
@@ -39,6 +44,12 @@ public class PubRelTest {
private final ChannelHandlerContext ctx =
mock(ChannelHandlerContext.class);
+ @BeforeEach
+ public void setUp() {
+ Singleton.INST.single(ChannelRepository.class, new
ChannelRepository());
+ when(ctx.channel()).thenReturn(mock(Channel.class));
+ }
+
@Test
public void testPubRelRespondsWithPubComp() {
MqttFixedHeader pubRelFixedHeader = new
MqttFixedHeader(MqttMessageType.PUBREL, false, MqttQoS.AT_LEAST_ONCE, false, 0);
@@ -52,4 +63,18 @@ public class PubRelTest {
assertEquals(MqttQoS.AT_MOST_ONCE, pubComp.fixedHeader().qosLevel());
assertEquals(PACKET_ID, ((MqttMessageIdVariableHeader)
pubComp.variableHeader()).messageId());
}
+
+ /**
+ * A PUBREL for an unknown packet id is still answered with PUBCOMP
(#7919).
+ */
+ @Test
+ public void testPubRelForUnknownPacketIdStillRespondsWithPubComp() {
+ MqttFixedHeader pubRelFixedHeader = new
MqttFixedHeader(MqttMessageType.PUBREL, false, MqttQoS.AT_LEAST_ONCE, false, 0);
+ MqttMessage pubRel = new MqttMessage(pubRelFixedHeader,
MqttMessageIdVariableHeader.from(PACKET_ID));
+ new PubRel().pubRel(ctx, pubRel);
+
+ ArgumentCaptor<MqttMessage> captor =
ArgumentCaptor.forClass(MqttMessage.class);
+ verify(ctx).writeAndFlush(captor.capture());
+ assertEquals(MqttMessageType.PUBCOMP,
captor.getValue().fixedHeader().messageType());
+ }
}
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/PublishTest.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/PublishTest.java
index 42f72bd51a..adb9a2894b 100644
---
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/PublishTest.java
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/PublishTest.java
@@ -42,6 +42,7 @@ import io.netty.util.CharsetUtil;
import io.netty.util.ReferenceCountUtil;
import org.apache.shenyu.common.utils.Singleton;
import org.apache.shenyu.protocol.mqtt.repositories.ChannelRepository;
+import org.apache.shenyu.protocol.mqtt.repositories.InflightRepository;
import org.apache.shenyu.protocol.mqtt.repositories.SubscribeRepository;
import org.apache.shenyu.protocol.mqtt.repositories.SessionRepository;
import org.apache.shenyu.protocol.mqtt.repositories.TopicRepository;
@@ -67,6 +68,7 @@ import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.mock;
@@ -106,6 +108,10 @@ public final class PublishTest {
private static final int PUBLISHER_PACKET_ID = 12345;
+ private static final int RETRANSMIT_PACKET_ID = 77;
+
+ private static final int BARRIER_PACKET_ID = 78;
+
private static final Duration TIMEOUT = Duration.ofSeconds(5);
private static final Duration POLL_INTERVAL = Duration.ofMillis(10);
@@ -132,6 +138,7 @@ public final class PublishTest {
Singleton.INST.single(TopicRepository.class, TOPIC_REPOSITORY);
Singleton.INST.single(ChannelRepository.class, CHANNEL_REPOSITORY);
Singleton.INST.single(SessionRepository.class, new
SessionRepository());
+ Singleton.INST.single(InflightRepository.class, new
InflightRepository());
new MqttContext().setUserName(USER_NAME);
new MqttContext().setPassword(PASSWORD);
@@ -146,6 +153,7 @@ public final class PublishTest {
public void tearDown() {
MqttPacketIdGenerator.remove(subscriberChannel);
MqttPacketIdGenerator.remove(otherSubscriberChannel);
+ Singleton.INST.get(InflightRepository.class).remove(CLIENT_ID);
publisherChannel.finishAndReleaseAll();
clearSharedState();
@@ -286,7 +294,8 @@ public final class PublishTest {
captureMessages(subscriberChannel, 1);
captureMessages(otherSubscriberChannel, 1);
- publishToSubscribers(MqttQoS.EXACTLY_ONCE);
+ // a conforming client never reuses a packet id before its
acknowledgement
+ publishToSubscribers(MqttQoS.EXACTLY_ONCE, PUBLISHER_PACKET_ID + 1,
false);
List<MqttPublishMessage> messages = captureMessages(subscriberChannel,
2);
assertEquals(1, messages.get(0).variableHeader().packetId());
@@ -339,6 +348,86 @@ public final class PublishTest {
assertEquals(PUBLISHER_PACKET_ID, ((MqttMessageIdVariableHeader)
pubRec.variableHeader()).messageId());
}
+ @Test
+ public void testQos2RetransmitIsPublishedExactlyOnce() {
+ CHANNEL_REPOSITORY.add(publisherChannel, CLIENT_ID);
+ addSubscriber(subscriberChannel, MqttQoS.EXACTLY_ONCE);
+ try {
+ publishToSubscribers(MqttQoS.EXACTLY_ONCE, RETRANSMIT_PACKET_ID,
false);
+ captureMessages(subscriberChannel, 1);
+ assertPubRec(RETRANSMIT_PACKET_ID);
+
+ // the retransmission is answered, but must not be delivered again
(#7919)
+ publishToSubscribers(MqttQoS.EXACTLY_ONCE, RETRANSMIT_PACKET_ID,
true);
+ assertPubRec(RETRANSMIT_PACKET_ID);
+
+ // ordering barrier: a fan-out happens on the common pool, so the
next publish
+ // would queue behind it — if the retransmission had been
delivered, the count
+ // here could only be 3, and times(2) fails deterministically
+ publishToSubscribers(MqttQoS.EXACTLY_ONCE, BARRIER_PACKET_ID,
false);
+ assertPubRec(BARRIER_PACKET_ID);
+ captureMessages(subscriberChannel, 2);
+ } finally {
+ CHANNEL_REPOSITORY.remove(publisherChannel);
+ }
+ }
+
+ /**
+ * A new session (clean-session reconnect) resets the packet id space, so a
+ * stale in-flight id must not suppress a publish of the same id (#7919).
+ */
+ @Test
+ public void testCleanSessionReconnectClearsInflightIds() {
+ CHANNEL_REPOSITORY.add(publisherChannel, CLIENT_ID);
+ addSubscriber(subscriberChannel, MqttQoS.EXACTLY_ONCE);
+ final EmbeddedChannel reconnected = channel(false);
+ CHANNEL_REPOSITORY.add(reconnected, CLIENT_ID);
+ try {
+ publishToSubscribers(MqttQoS.EXACTLY_ONCE, RETRANSMIT_PACKET_ID,
false);
+ captureMessages(subscriberChannel, 1);
+ assertPubRec(RETRANSMIT_PACKET_ID);
+
+ // reconnect with cleanSession=1: a new session starts, the packet
id space resets
+ new Connect().connect(reconnected.pipeline().lastContext(),
connectMessage());
+ assertNotNull(reconnected.readOutbound());
+
+ // the same packet id is legal again on the new session and must
be delivered
+ final MqttFixedHeader fixedHeader = new
MqttFixedHeader(MqttMessageType.PUBLISH, true, MqttQoS.EXACTLY_ONCE, false, 0);
+ final MqttPublishVariableHeader variableHeader = new
MqttPublishVariableHeader(TOPIC, RETRANSMIT_PACKET_ID);
+ new Publish().publish(reconnected.pipeline().lastContext(),
+ new MqttPublishMessage(fixedHeader, variableHeader,
Unpooled.copiedBuffer(PAYLOAD, CharsetUtil.UTF_8)));
+ final MqttMessage pubRec = reconnected.readOutbound();
+ assertEquals(MqttMessageType.PUBREC,
pubRec.fixedHeader().messageType());
+ captureMessages(subscriberChannel, 2);
+ } finally {
+ CHANNEL_REPOSITORY.remove(publisherChannel);
+ CHANNEL_REPOSITORY.remove(reconnected);
+ reconnected.finishAndReleaseAll();
+ }
+ }
+
+ @Test
+ public void testPubRelReleasesInflightForRedelivery() {
+ CHANNEL_REPOSITORY.add(publisherChannel, CLIENT_ID);
+ addSubscriber(subscriberChannel, MqttQoS.EXACTLY_ONCE);
+ try {
+ publishToSubscribers(MqttQoS.EXACTLY_ONCE, RETRANSMIT_PACKET_ID,
false);
+ captureMessages(subscriberChannel, 1);
+ assertPubRec(RETRANSMIT_PACKET_ID);
+
+ new PubRel().pubRel(publisherContext(),
pubRelMessage(RETRANSMIT_PACKET_ID));
+ MqttMessage pubComp = awaitOutbound(publisherChannel);
+ assertEquals(MqttMessageType.PUBCOMP,
pubComp.fixedHeader().messageType());
+
+ // the packet id is free again, so the next publish is delivered
+ publishToSubscribers(MqttQoS.EXACTLY_ONCE, RETRANSMIT_PACKET_ID,
true);
+ captureMessages(subscriberChannel, 2);
+ assertPubRec(RETRANSMIT_PACKET_ID);
+ } finally {
+ CHANNEL_REPOSITORY.remove(publisherChannel);
+ }
+ }
+
private static Stream<Arguments> retainedPayloads() {
byte[] text = "hello 中文".getBytes(StandardCharsets.UTF_8);
byte[] binary = {0, (byte) 0xff, (byte) 0x80, 0x41};
@@ -411,6 +500,24 @@ public final class PublishTest {
new Publish().publish(publisherContext(), new
MqttPublishMessage(fixedHeader, variableHeader, payload));
}
+ private void publishToSubscribers(final MqttQoS qos, final int packetId,
final boolean dup) {
+ final MqttFixedHeader fixedHeader = new
MqttFixedHeader(MqttMessageType.PUBLISH, dup, qos, false, 0);
+ final MqttPublishVariableHeader variableHeader = new
MqttPublishVariableHeader(TOPIC, packetId);
+ new Publish().publish(publisherContext(),
+ new MqttPublishMessage(fixedHeader, variableHeader,
Unpooled.copiedBuffer(PAYLOAD, CharsetUtil.UTF_8)));
+ }
+
+ private void assertPubRec(final int packetId) {
+ final MqttMessage pubRec = awaitOutbound(publisherChannel);
+ assertEquals(MqttMessageType.PUBREC,
pubRec.fixedHeader().messageType());
+ assertEquals(packetId, ((MqttMessageIdVariableHeader)
pubRec.variableHeader()).messageId());
+ }
+
+ private MqttMessage pubRelMessage(final int packetId) {
+ return new MqttMessage(new MqttFixedHeader(MqttMessageType.PUBREL,
false, MqttQoS.AT_MOST_ONCE, false, 0),
+ MqttMessageIdVariableHeader.from(packetId));
+ }
+
private MqttPublishMessage captureMessage(final Channel channel) {
return captureMessages(channel, 1).get(0);
}
