Auto assigned subscription id isn't sent to broker

[TDYJeffreyDevloo]

• [x] Bug exists Release Version 1.2.5 ( Master Branch)
• [ ] Bug exists in MQTTv3 Client on Snapshot Version 1.2.6-SNAPSHOT (Develop Branch)
• [x] Bug exists in MQTTv5 Client on Snapshot Version 1.2.6-SNAPSHOT (Develop Branch)

Auto assigned subscription identifier isn't sent to the broker

The existence of the subscription identifier when subscribing using the MQTT5 client is

• Using the wrong list to check for it's existence
• Not setting the auto-incremented value on the subscription options for the subscribe request

A simple demo that demonstrates the issue:

import com.hivemq.client.mqtt.datatypes.MqttQos;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import org.eclipse.paho.mqttv5.client.IMqttToken;
import org.eclipse.paho.mqttv5.client.MqttAsyncClient;
import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
import org.eclipse.paho.mqttv5.common.MqttException;
import org.eclipse.paho.mqttv5.common.MqttMessage;
import org.eclipse.paho.mqttv5.common.MqttSubscription;
import org.eclipse.paho.mqttv5.common.packet.MqttProperties;

public class TestPaho
{

  public static void main(String[] args) throws InterruptedException, MqttException
  {
    MqttAsyncClient serverClient = new MqttAsyncClient("tcp://localhost:1883", "myServer");
    MqttAsyncClient deviceClient = new MqttAsyncClient("tcp://localhost:1883", "myDevice");

    serverClient.connect().waitForCompletion();
    deviceClient.connect().waitForCompletion();

    System.out.println("Connected");

    CountDownLatch receivedLatch = new CountDownLatch(4);

    MqttProperties subProperties1 = new MqttProperties();
    subProperties1.setSubscriptionIdentifiers(List.of(0)); // Bug in paho? This is for publish but subscribe requires it
    # subProperties1.setSubscriptionIdentifier(1);
    serverClient.subscribe(new MqttSubscription("#", MqttQos.AT_LEAST_ONCE.getCode()), null, null, (topic, message) ->
    {
      System.out.println("# - Received message");
      receivedLatch.countDown();
    }, subProperties1);

    MqttProperties subProperties2 = new MqttProperties();
    subProperties2.setSubscriptionIdentifiers(List.of(0));
    # subProperties2.setSubscriptionIdentifier(2);
    serverClient.subscribe(new MqttSubscription("A", MqttQos.AT_LEAST_ONCE.getCode()), null, null, (topic, message) ->
    {
      System.out.println("A - Received message");
      receivedLatch.countDown();
    }, subProperties2);

    System.out.println("A - Sending");
    deviceClient.publish("A", new MqttMessage("1".getBytes(), MqttQos.AT_LEAST_ONCE.getCode(), false, new MqttProperties()))
            .waitForCompletion();

    System.out.println("A - Sent");

    receivedLatch.await();
    deviceClient.close();
    serverClient.close();
  }
}

Any user of the API needs to use MqttProperties#setSubscriptionIdentifier regardless of Automatic Subscription Identifier Assignment. The property it should be checking within a subscribe is the MqttProperties#getSubscriptionIdentifier. Not the list. The doc states:

    /**
     * Subscription Identifiers. (Publish Only)
     * 
     * <p>
     * The Subscription Identifiers are associated with any subscription created or
     * modified as the result of a SUBSCRIBE packet. If a subscription was made with
     * a Subscription Identifier, then any incoming messages that match that
     * subscription will contain the associated subscription identifier, if the
     * incoming message matches multiple subscriptions made by the same client, then
     * it will contain a list of all associated subscription identifiers. This
     * property is ONLY for PUBLISH packets. For a Subscription Identifier sent in a
     * SUBSCRIBE packet, see {@link MqttProperties#getSubscriptionIdentifier()}
     * </p>
     * 
     * @return A {@link List} of Subscription Identifiers.
     */
    public List<Integer> getSubscriptionIdentifiers() {
        return publishSubscriptionIdentifiers;
    }
...

    /**
     * Subscription Identifier. (Subscribe Only)
     * 
     * <p>
     * The Subscription identifier field can be set on a SUBSCRIBE packet and will
     * be returned with any incoming PUBLISH packets that match the associated
     * subscription. This property is ONLY for SUBSCRIBE packets. For Subscription
     * Identifier(s) sent in a PUBLISH packet, see
     * {@link MqttProperties#getSubscriptionIdentifiers()}
     * </p>
     * 
     * @return The Subscription Identifier.
     */
    public Integer getSubscriptionIdentifier() {
        return subscribeSubscriptionIdentifier;
    }

The correct property asside, the newly chosen subId is used internally https://github.com/eclipse/paho.mqtt.java/blob/master/org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/MqttAsyncClient.java#L1308 but never set when sending the SUBSCRIBE message https://github.com/eclipse/paho.mqtt.java/blob/master/org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/MqttAsyncClient.java#L1314

[TDYJeffreyDevloo]

Patch for the issue:

Index: org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/MqttAsyncClient.java
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/MqttAsyncClient.java b/org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/MqttAsyncClient.java
--- a/org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/MqttAsyncClient.java  (revision f4e0db802a4433645ef011e711646a09ec9fae89)
+++ b/org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/MqttAsyncClient.java  (revision 7fcaefef9bf9ed5cfa68af4086f5f77aa47b81cc)
@@ -19,6 +19,7 @@
 import java.io.ByteArrayOutputStream;
 import java.io.DataOutputStream;
 import java.util.Hashtable;
+import java.util.Optional;
 import java.util.Properties;
 import java.util.Timer;
 import java.util.TimerTask;
@@ -1273,12 +1274,8 @@
    public IMqttToken subscribe(MqttSubscription[] subscriptions, Object userContext, MqttActionListener callback,
            IMqttMessageListener messageListener, MqttProperties subscriptionProperties) throws MqttException {

-       int subId = 0;
-       try {
-           subId = subscriptionProperties.getSubscriptionIdentifiers().get(0);
-       } catch (IndexOutOfBoundsException e) {
-           log.fine(CLASS_NAME, "subscribe", "No sub subscription property(s)");
-       }
+       int subId = Optional.ofNullable(subscriptionProperties.getSubscriptionIdentifier())
+           .orElse(0);
        // Automatic Subscription Identifier Assignment is enabled
        if (connOpts.useSubscriptionIdentifiers() && this.mqttConnection.isSubscriptionIdentifiersAvailable()) {

@@ -1294,9 +1291,15 @@
            } else {
                // Automatically assign new ID and link to callback.
                subId = this.mqttSession.getNextSubscriptionIdentifier();
+               subscriptionProperties.setSubscriptionIdentifier(subId);
+           }
+       } else {
+           if (subId != 0) {
+               log.fine(CLASS_NAME, "subscribe", "Subscription identifiers are not available or disabled");
+               subscriptionProperties.setSubscriptionIdentifier(null);
            }
        }
-       
+
        // add message handlers to the list for this client
        for (MqttSubscription subscription : subscriptions) {
            MqttTopicValidator.validate(subscription.getTopic(),