[
https://issues.apache.org/jira/browse/AMQ-9592?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17896642#comment-17896642
]
Jean-Baptiste Onofré commented on AMQ-9592:
-------------------------------------------
Is it really an issue ?
QoS 2 means exactly once, e.g. ensuring each message is delivered exactly once
to a client (not lost or duplicated).
To do that, we have 2 messages flow:
# the client send PUBLISH packet
# the broker replies PUBREC packet (when ok). If the client doesn't receive
PUBREC before the timeout, it sends PUBLISH again with DUP flag
# the client replies with PUBREL
# the broker replies with PUBCOMP and we consider the flow completed.
Until the client has received the corresponding PUBREL package, the broker must
ack PUBLIC packets with the same id by sending a PUBREC. It must not cause
duplicate messages to be delivered to any onward recipients.
So can you check that the flow is actually completed on producer side (PUBCOMP
on the client) ?
> Incorrect MQTTv3.1.1 behaviour: QoS2 messages were delivered to unintended
> consumers
> -------------------------------------------------------------------------------------
>
> Key: AMQ-9592
> URL: https://issues.apache.org/jira/browse/AMQ-9592
> Project: ActiveMQ Classic
> Issue Type: Bug
> Components: MQTT
> Affects Versions: 5.18.5, 6.1.3
> Reporter: Ken Liao
> Assignee: Jean-Baptiste Onofré
> Priority: Major
> Fix For: 6.2.0, 6.1.4, 5.18.7
>
>
> To replicate:
> - Run a subscriber with QOS2 to a topic with client ID say "1"and stop it
> - Publish 5 messages to the topic, do not consume these yet.
> - Run another subscriber with QOS2 to the topic with a different client ID
> say "2'.
> - This subscriber with different cliend ID receives 5 messages that were
> intended for the clientID "1". However, this does not mean that client ID "1"
> will not receive the 5 messages when it comes back online. I noticed that
> client ID "1" also received the 5 message.
> Now that both clientID 1 and 2 received messages, subscribing a new clientID
> doesn't receive these message. However, if we send 5 message again to the
> topic and do not consume it for all clientIDs, it would delivered to any new
> subscription.
> For new client Id broker must create new session, so messages sent after
> client 1 disconnected should route to client 1 only since client 2 because it
> was connected after those 5 messages were sent (hence the new session of
> client 2 shouldn't contain old messages).
> Here are sample codes to reproduce:
> The producer Python script
> {code:python}
> import paho.mqtt.client as mqttClient
> import time
> import ssl
> # context = ssl.create_default_context()
> Connected = False
> broker_address= "localhost"
> port = 1883
> user = "admin"
> password = "admin"
> def on_connect(client, userdata, flags, rc):
> if rc == 0:
> print("Connected to broker")
> global Connected #Use global variable
> Connected = True #Signal connection
> else:
> print("Connection failed")
> client = mqttClient.Client(client_id="producer",transport="tcp") #
> client.username_pw_set(user, password=password) #set username and password
> client.on_connect=on_connect
> # client.tls_set_context(context=context)
> client.connect(broker_address, port=port)
> client.loop_start()
> i=0
> while True:
> time.sleep(1)
> i=i+1
> client.publish("test/data", "This is my test msg 123.",2,False)
> print("Sent"+str(i))
> {code}
> The consumer Python script
> {code:python}
> import paho.mqtt.client as mqttClient
> import time
> import ssl
> Connected = False
> broker_address= "localhost"
> port = 1883
> user = "admin"
> password = "admin"
> def on_connect(client, userdata, flags, rc):
> if rc == 0:
> print("Connected to broker")
> global Connected #Use global variable
> Connected = True #Signal connection
> print(flags)
> client.subscribe("test/data",2)
> else:
> print("Connection failed")
> def on_message(client, userdata, msg):
> print(msg.topic+" "+str(msg.payload))
> client =
> mqttClient.Client(client_id="id2",transport="tcp",clean_session=False) #
> client.username_pw_set(user, password=password) #set username and password
> client.on_connect=on_connect
> client.on_message=on_message
> # client.tls_set_context(context=context)
> client.connect(broker_address, port=port)
> client.loop_start()
> while Connected != True:
> time.sleep(0.1)
> time.sleep(1000)
> {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
For further information, visit: https://activemq.apache.org/contact
