You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@artemis.apache.org by "jbertram (via GitHub)" <gi...@apache.org> on 2026/09/10 16:14:22 UTC
[PR] ARTEMIS-6241 mitigate potential races [artemis]
jbertram opened a new pull request, #6688:
URL: https://github.com/apache/artemis/pull/6688
Unify locking so sendToQueue() and the PUBACK/PUBREC/PUBREL/PUBCOMP handlers all synchronize on the same monitor (i.e. this), and move the duplicate-packet-ID check into sendToQueue() itself so the check and the cache mutation happen atomically under that lock. Also removes the now-unused lock field.
Refactor a few methods out of sendToQueue() to increase readability.
Add values to cache instantly to avoid potential races with doing so via o.a.a.a.c.t.Transaction#afterStore.
Lastly, move outgoing protocol logging closer to putting the packet on the wire.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@artemis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@artemis.apache.org
For additional commands, e-mail: issues-help@artemis.apache.org
Re: [PR] ARTEMIS-6241 mitigate potential races [artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic merged PR #6688:
URL: https://github.com/apache/artemis/pull/6688
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@artemis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@artemis.apache.org
For additional commands, e-mail: issues-help@artemis.apache.org
Re: [PR] ARTEMIS-6241 mitigate potential races [artemis]
Posted by "jbertram (via GitHub)" <gi...@apache.org>.
jbertram commented on code in PR #6688:
URL: https://github.com/apache/artemis/pull/6688#discussion_r4021121092
##########
artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java:
##########
@@ -145,125 +143,101 @@ protected void publishToClient(ICoreMessage message, ServerConsumer consumer) th
*
* @param internal if true means on behalf of the broker (skips authorisation) and does not return ack.
*/
- void sendToQueue(MqttPublishMessage message, boolean internal) throws Exception {
- synchronized (lock) {
- if (createProducer) {
- session.getServerSession().addProducer(senderName, MQTTProtocolManagerFactory.MQTT_PROTOCOL_NAME, ServerProducer.ANONYMOUS);
- createProducer = false;
- }
- String topic = message.variableHeader().topicName();
- if (session.getVersion() == MQTTVersion.MQTT_5) {
- Integer alias = MQTTUtil.getProperty(Integer.class, message.variableHeader().properties(), TOPIC_ALIAS);
- if (alias != null) {
- Integer topicAliasMax = session.getProtocolManager().getTopicAliasMaximum();
- if (alias == 0) {
- // [MQTT-3.3.2-8]
- throw new DisconnectException(MQTTReasonCodes.TOPIC_ALIAS_INVALID);
- } else if (topicAliasMax != null && alias > topicAliasMax) {
- // [MQTT-3.3.2-9]
- throw new DisconnectException(MQTTReasonCodes.TOPIC_ALIAS_INVALID);
- }
+ synchronized void sendToQueue(MqttPublishMessage message, boolean internal) throws Exception {
Review Comment:
Agreed. This is a belt-and-suspenders kind of thing. The main race is the instant-add to the cache.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@artemis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@artemis.apache.org
For additional commands, e-mail: issues-help@artemis.apache.org
Re: [PR] ARTEMIS-6241 mitigate potential races [artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic commented on code in PR #6688:
URL: https://github.com/apache/artemis/pull/6688#discussion_r4016408182
##########
artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java:
##########
@@ -145,125 +143,101 @@ protected void publishToClient(ICoreMessage message, ServerConsumer consumer) th
*
* @param internal if true means on behalf of the broker (skips authorisation) and does not return ack.
*/
- void sendToQueue(MqttPublishMessage message, boolean internal) throws Exception {
- synchronized (lock) {
- if (createProducer) {
- session.getServerSession().addProducer(senderName, MQTTProtocolManagerFactory.MQTT_PROTOCOL_NAME, ServerProducer.ANONYMOUS);
- createProducer = false;
- }
- String topic = message.variableHeader().topicName();
- if (session.getVersion() == MQTTVersion.MQTT_5) {
- Integer alias = MQTTUtil.getProperty(Integer.class, message.variableHeader().properties(), TOPIC_ALIAS);
- if (alias != null) {
- Integer topicAliasMax = session.getProtocolManager().getTopicAliasMaximum();
- if (alias == 0) {
- // [MQTT-3.3.2-8]
- throw new DisconnectException(MQTTReasonCodes.TOPIC_ALIAS_INVALID);
- } else if (topicAliasMax != null && alias > topicAliasMax) {
- // [MQTT-3.3.2-9]
- throw new DisconnectException(MQTTReasonCodes.TOPIC_ALIAS_INVALID);
- }
+ synchronized void sendToQueue(MqttPublishMessage message, boolean internal) throws Exception {
Review Comment:
Is this really making a difference?
the only caller for this is this: https://github.com/apache/artemis/blob/29b74aea866fa2858c9129eaa6777cce23a11d27/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTProtocolHandler.java#L187-L188
At that point, that's single threaded, as it belongs to the session.
if you had a reconnect, you would have a new MQTTPublishmanager, as this is created on MQTTSession's constructor.
If you're having issues here, in case of a reconenct, I suspect you would need to recover the PublishManager out of an external holder based on clientSession or anything like that, or make sure that this session is finished and done before being able to publish again?
I'm saying you would need the PublishManager to survive beyond the session somehow to be able to do a proper coordination after reconencts.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@artemis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@artemis.apache.org
For additional commands, e-mail: issues-help@artemis.apache.org
Re: [PR] ARTEMIS-6241 mitigate potential races [artemis]
Posted by "jbertram (via GitHub)" <gi...@apache.org>.
jbertram commented on code in PR #6688:
URL: https://github.com/apache/artemis/pull/6688#discussion_r4021121092
##########
artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java:
##########
@@ -145,125 +143,101 @@ protected void publishToClient(ICoreMessage message, ServerConsumer consumer) th
*
* @param internal if true means on behalf of the broker (skips authorisation) and does not return ack.
*/
- void sendToQueue(MqttPublishMessage message, boolean internal) throws Exception {
- synchronized (lock) {
- if (createProducer) {
- session.getServerSession().addProducer(senderName, MQTTProtocolManagerFactory.MQTT_PROTOCOL_NAME, ServerProducer.ANONYMOUS);
- createProducer = false;
- }
- String topic = message.variableHeader().topicName();
- if (session.getVersion() == MQTTVersion.MQTT_5) {
- Integer alias = MQTTUtil.getProperty(Integer.class, message.variableHeader().properties(), TOPIC_ALIAS);
- if (alias != null) {
- Integer topicAliasMax = session.getProtocolManager().getTopicAliasMaximum();
- if (alias == 0) {
- // [MQTT-3.3.2-8]
- throw new DisconnectException(MQTTReasonCodes.TOPIC_ALIAS_INVALID);
- } else if (topicAliasMax != null && alias > topicAliasMax) {
- // [MQTT-3.3.2-9]
- throw new DisconnectException(MQTTReasonCodes.TOPIC_ALIAS_INVALID);
- }
+ synchronized void sendToQueue(MqttPublishMessage message, boolean internal) throws Exception {
Review Comment:
Agreed. This is kind of a belt-and-suspenders kind of thing. The main race is the instant-add to the cache.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@artemis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@artemis.apache.org
For additional commands, e-mail: issues-help@artemis.apache.org
Re: [PR] ARTEMIS-6241 mitigate potential races [artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic commented on code in PR #6688:
URL: https://github.com/apache/artemis/pull/6688#discussion_r4016408182
##########
artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java:
##########
@@ -145,125 +143,101 @@ protected void publishToClient(ICoreMessage message, ServerConsumer consumer) th
*
* @param internal if true means on behalf of the broker (skips authorisation) and does not return ack.
*/
- void sendToQueue(MqttPublishMessage message, boolean internal) throws Exception {
- synchronized (lock) {
- if (createProducer) {
- session.getServerSession().addProducer(senderName, MQTTProtocolManagerFactory.MQTT_PROTOCOL_NAME, ServerProducer.ANONYMOUS);
- createProducer = false;
- }
- String topic = message.variableHeader().topicName();
- if (session.getVersion() == MQTTVersion.MQTT_5) {
- Integer alias = MQTTUtil.getProperty(Integer.class, message.variableHeader().properties(), TOPIC_ALIAS);
- if (alias != null) {
- Integer topicAliasMax = session.getProtocolManager().getTopicAliasMaximum();
- if (alias == 0) {
- // [MQTT-3.3.2-8]
- throw new DisconnectException(MQTTReasonCodes.TOPIC_ALIAS_INVALID);
- } else if (topicAliasMax != null && alias > topicAliasMax) {
- // [MQTT-3.3.2-9]
- throw new DisconnectException(MQTTReasonCodes.TOPIC_ALIAS_INVALID);
- }
+ synchronized void sendToQueue(MqttPublishMessage message, boolean internal) throws Exception {
Review Comment:
Is this really making a difference?
the only caller for this is this: https://github.com/apache/artemis/blob/29b74aea866fa2858c9129eaa6777cce23a11d27/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTProtocolHandler.java#L187-L188
At that point, that's single threaded, as it belongs to the session.
if you had a reconnect, you would have a new MQTTPublishmanager, as this is created on MQTTSession's constructor.
If you're having issues here, in case of a reconenct, I suspect you would need to recover the public Manager out of an external holder based on clientSession, or make sure that this session is finished and done before being able to publish again?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@artemis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@artemis.apache.org
For additional commands, e-mail: issues-help@artemis.apache.org
Re: [PR] ARTEMIS-6241 mitigate potential races [artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic commented on code in PR #6688:
URL: https://github.com/apache/artemis/pull/6688#discussion_r4019947975
##########
artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java:
##########
@@ -145,125 +143,101 @@ protected void publishToClient(ICoreMessage message, ServerConsumer consumer) th
*
* @param internal if true means on behalf of the broker (skips authorisation) and does not return ack.
*/
- void sendToQueue(MqttPublishMessage message, boolean internal) throws Exception {
- synchronized (lock) {
- if (createProducer) {
- session.getServerSession().addProducer(senderName, MQTTProtocolManagerFactory.MQTT_PROTOCOL_NAME, ServerProducer.ANONYMOUS);
- createProducer = false;
- }
- String topic = message.variableHeader().topicName();
- if (session.getVersion() == MQTTVersion.MQTT_5) {
- Integer alias = MQTTUtil.getProperty(Integer.class, message.variableHeader().properties(), TOPIC_ALIAS);
- if (alias != null) {
- Integer topicAliasMax = session.getProtocolManager().getTopicAliasMaximum();
- if (alias == 0) {
- // [MQTT-3.3.2-8]
- throw new DisconnectException(MQTTReasonCodes.TOPIC_ALIAS_INVALID);
- } else if (topicAliasMax != null && alias > topicAliasMax) {
- // [MQTT-3.3.2-9]
- throw new DisconnectException(MQTTReasonCodes.TOPIC_ALIAS_INVALID);
- }
+ synchronized void sendToQueue(MqttPublishMessage message, boolean internal) throws Exception {
Review Comment:
okay.. the change is valid ... however I don't think you were having a race for the lack of synchronized then, as this is pretty much single threaded.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@artemis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@artemis.apache.org
For additional commands, e-mail: issues-help@artemis.apache.org
Re: [PR] ARTEMIS-6241 mitigate potential races [artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic commented on code in PR #6688:
URL: https://github.com/apache/artemis/pull/6688#discussion_r4016408182
##########
artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java:
##########
@@ -145,125 +143,101 @@ protected void publishToClient(ICoreMessage message, ServerConsumer consumer) th
*
* @param internal if true means on behalf of the broker (skips authorisation) and does not return ack.
*/
- void sendToQueue(MqttPublishMessage message, boolean internal) throws Exception {
- synchronized (lock) {
- if (createProducer) {
- session.getServerSession().addProducer(senderName, MQTTProtocolManagerFactory.MQTT_PROTOCOL_NAME, ServerProducer.ANONYMOUS);
- createProducer = false;
- }
- String topic = message.variableHeader().topicName();
- if (session.getVersion() == MQTTVersion.MQTT_5) {
- Integer alias = MQTTUtil.getProperty(Integer.class, message.variableHeader().properties(), TOPIC_ALIAS);
- if (alias != null) {
- Integer topicAliasMax = session.getProtocolManager().getTopicAliasMaximum();
- if (alias == 0) {
- // [MQTT-3.3.2-8]
- throw new DisconnectException(MQTTReasonCodes.TOPIC_ALIAS_INVALID);
- } else if (topicAliasMax != null && alias > topicAliasMax) {
- // [MQTT-3.3.2-9]
- throw new DisconnectException(MQTTReasonCodes.TOPIC_ALIAS_INVALID);
- }
+ synchronized void sendToQueue(MqttPublishMessage message, boolean internal) throws Exception {
Review Comment:
Is this really making a difference?
the only caller for this is this: https://github.com/apache/artemis/blob/29b74aea866fa2858c9129eaa6777cce23a11d27/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTProtocolHandler.java#L187-L188
At that point, that's single threaded, belong to the session.
if you had a reconnect, you would have a new MQTTPublishmanager, as this is created on MQTTSession's constructor.
If you're having issues here, in case of a reconenct, I suspect you would need to recover the public Manager out of an external holder based on clientSession, or make sure that this session is finished and done before being able to publish again?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@artemis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@artemis.apache.org
For additional commands, e-mail: issues-help@artemis.apache.org
Re: [PR] ARTEMIS-6241 mitigate potential races [artemis]
Posted by "jbertram (via GitHub)" <gi...@apache.org>.
jbertram commented on code in PR #6688:
URL: https://github.com/apache/artemis/pull/6688#discussion_r4018447397
##########
artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java:
##########
@@ -145,125 +143,101 @@ protected void publishToClient(ICoreMessage message, ServerConsumer consumer) th
*
* @param internal if true means on behalf of the broker (skips authorisation) and does not return ack.
*/
- void sendToQueue(MqttPublishMessage message, boolean internal) throws Exception {
- synchronized (lock) {
- if (createProducer) {
- session.getServerSession().addProducer(senderName, MQTTProtocolManagerFactory.MQTT_PROTOCOL_NAME, ServerProducer.ANONYMOUS);
- createProducer = false;
- }
- String topic = message.variableHeader().topicName();
- if (session.getVersion() == MQTTVersion.MQTT_5) {
- Integer alias = MQTTUtil.getProperty(Integer.class, message.variableHeader().properties(), TOPIC_ALIAS);
- if (alias != null) {
- Integer topicAliasMax = session.getProtocolManager().getTopicAliasMaximum();
- if (alias == 0) {
- // [MQTT-3.3.2-8]
- throw new DisconnectException(MQTTReasonCodes.TOPIC_ALIAS_INVALID);
- } else if (topicAliasMax != null && alias > topicAliasMax) {
- // [MQTT-3.3.2-9]
- throw new DisconnectException(MQTTReasonCodes.TOPIC_ALIAS_INVALID);
- }
+ synchronized void sendToQueue(MqttPublishMessage message, boolean internal) throws Exception {
Review Comment:
The modifications to `sendToQueue` have 2 main purposes:
1. Use `synchronized` to align/share the locking with the other methods which deal with the publish cache and other QoS 2 state (e.g. packet ID correlation records).
2. Check the cache and add to the cache under that shared lock.
The issue isn't so much a reconnect as it's simply making sure QoS 2 state is guaranteed to be consistent at all times.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@artemis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@artemis.apache.org
For additional commands, e-mail: issues-help@artemis.apache.org