Subscription's errors not handled as expected on session end

[adamcrane]

Expected behavior

Given a Mqtt3RxClient connected with two or more subscriptions
When MQTT session ends unexpectedly
Then the subscriptions onError method should be called
and client should reconnect

Actual behavior

The RxJavaPlugins.onError() is called
and client does not reconnect

To Reproduce

Steps

Connect client to broker with two or more subscriptions
End client's session with broker

Reproducer code

    Mqtt3RxClient client = Mqtt3Client.builder()
            .identifier("id")
            .serverHost(mqttConfig.getHost())
            .serverPort(mqttConfig.getPort())
            .automaticReconnectWithDefaultConfig()
            .buildRx();
   
    Disposable disposable = client.connect(mqttConnect)
            .doOnSuccess(connAck -> log
                    .info("Successfully connected to hivemq: " + connAck.getReturnCode()))
            .ignoreElement()
            .andThen(client.subscribeStreamWith()
                           .topicFilter("a/b/c").qos(MqttQos.AT_LEAST_ONCE).addSubscription()
                           .topicFilter("a/b/c/d").qos(MqttQos.EXACTLY_ONCE).applySubscription()
                           .applySubscribe())
            .flatMapCompletable(Kafka::toKafka)
            .subscribe(
                    () -> System.out.println("Closing connections"),
                x -> System.err.println("Uh Oh we shouldnt get here" + x));

Details

• Affected HiveMQ MQTT Client version(s): 1.1.0
• Used JVM version: 1.8 202b08

[SgtSilvio]

Thank you for your report.
I tried to reproduce your error with the following code:

Mqtt3RxClient client = Mqtt3Client.builder()
        .identifier("id")
        .serverHost("broker.hivemq.com")
        .automaticReconnectWithDefaultConfig()
        .addConnectedListener(context -> System.out.println("connected"))
        .addDisconnectedListener(context -> System.out.println("disconnected"))
        .buildRx();

Disposable disposable = client.connectWith().keepAlive(5).applyConnect()
        .doOnSuccess(connAck -> System.out.println("Successfully connected to hivemq: " + connAck.getReturnCode()))
        .ignoreElement()
        .andThen(client.subscribeStreamWith()
                .topicFilter("a/b/c").qos(MqttQos.AT_LEAST_ONCE)
                .applySubscribe())
        .subscribe(
                p -> System.out.println("Received message"),
                x -> System.err.println("Uh Oh we shouldnt get here" + x));

Results:

1. The client does reconnect as expected (you can see this if you add a ConnectedListener as shown above)
2. I can confirm that RxJavaPlugins.onError is called when you subscribe with 2+ subscriptions. This actually does not affect the operation of the client at all. It is just a log for something that is actually not an error. I will create a bug ticket for removing this log.

As the session expired, it seems that you are using cleanSession=true. Could you share the contents of mqttConnect from your example?
If a session expires, the subscriptions are no longer valid, so you have to re-subscribe. Automatic reconnect does not re-subscribe magically. So your assumption that "we shouldnt get here" is not correct. The error is expected to tell you that the subscription does no longer exist because the session expired.

[SgtSilvio]

quick suggestion: if you add a retry for the subscribe the client will try to re-subscribe

Disposable disposable = client.connectWith().keepAlive(5).applyConnect()
        .doOnSuccess(connAck -> System.out.println("Successfully connected to hivemq: " + connAck.getReturnCode()))
        .ignoreElement()
        .andThen(client.subscribeStreamWith()
                .topicFilter("a/b/c").qos(MqttQos.AT_LEAST_ONCE)
                .applySubscribe()
                .retry())
        .subscribe(
                p -> System.out.println("Received message"),
                x -> System.err.println("Uh Oh we shouldnt get here" + x));

[SgtSilvio]

Another question: Is it expected that the client re-subscribes automatically when the client reconnects automatically? If so, we could add a feature request for that.

[adamcrane]

Thanks for the quick reply. Our connect looks like :

Mqtt3Connect = Mqtt3Connect.builder()
            .simpleAuth(Mqtt3SimpleAuth.builder()
                .username(mqttConfig.getUsername())
                .password(mqttConfig.getPassword().getBytes())
                .cleanSession(false)
                .build())
            .build();

I've tried your suggestions above and am still not able to get the client to reconnect.

And yes it would be great if the client is able to resubscibe when the client reconnects

[SgtSilvio]

How do you observe that the client does not reconnect? Did you add a DisconnectedListener and a ConnectedListener?
Can you log the DisconnectedContext.getSource and getCause and post them here?
I also wonder why your session expires if you connect with cleanSession(false). Can you log ConnAck.isSessionPresent?

[adamcrane]

The initial connect is successful then we manually disconnect the client via the HiveMQ Dashboard. Here is the output when we disconnected the client

disconnected source: CLIENT cause: com.hivemq.client.mqtt.mqtt5.exceptions.Mqtt5ConnAckException: CONNECT failed as CONNACK contained an Error Code: BAD_USER_NAME_OR_PASSWORD.
disconnected source: CLIENT cause: com.hivemq.client.mqtt.mqtt5.exceptions.Mqtt5ConnAckException: CONNECT failed as CONNACK contained an Error Code: BAD_USER_NAME_OR_PASSWORD.
disconnected source: CLIENT cause: com.hivemq.client.mqtt.mqtt5.exceptions.Mqtt5ConnAckException: CONNECT failed as CONNACK contained an Error Code: BAD_USER_NAME_OR_PASSWORD.
disconnected source: CLIENT cause: com.hivemq.client.mqtt.mqtt5.exceptions.Mqtt5ConnAckException: CONNECT failed as CONNACK contained an Error Code: BAD_USER_NAME_OR_PASSWORD.
disconnected source: CLIENT cause: com.hivemq.client.mqtt.mqtt5.exceptions.Mqtt5ConnAckException: CONNECT failed as CONNACK contained an Error Code: BAD_USER_NAME_OR_PASSWORD.
disconnected source: CLIENT cause: com.hivemq.client.mqtt.mqtt5.exceptions.Mqtt5ConnAckException: CONNECT failed as CONNACK contained an Error Code: BAD_USER_NAME_OR_PASSWORD.

[adamcrane]

We were able to dig around the code a bit and found that the clients old MqttClientConfig was being used and not it's MqttConnect. So we moved our simpleAuth from the MqttConnect object to the MqttClient object. With this move we were able to successfully reconnect.

[SgtSilvio]

The initial connect is successful then we manually disconnect the client via the HiveMQ Dashboard. Here is the output when we disconnected the client

disconnected source: CLIENT cause: com.hivemq.client.mqtt.mqtt5.exceptions.Mqtt5ConnAckException: CONNECT failed as CONNACK contained an Error Code: BAD_USER_NAME_OR_PASSWORD.
disconnected source: CLIENT cause: com.hivemq.client.mqtt.mqtt5.exceptions.Mqtt5ConnAckException: CONNECT failed as CONNACK contained an Error Code: BAD_USER_NAME_OR_PASSWORD.
disconnected source: CLIENT cause: com.hivemq.client.mqtt.mqtt5.exceptions.Mqtt5ConnAckException: CONNECT failed as CONNACK contained an Error Code: BAD_USER_NAME_OR_PASSWORD.
disconnected source: CLIENT cause: com.hivemq.client.mqtt.mqtt5.exceptions.Mqtt5ConnAckException: CONNECT failed as CONNACK contained an Error Code: BAD_USER_NAME_OR_PASSWORD.
disconnected source: CLIENT cause: com.hivemq.client.mqtt.mqtt5.exceptions.Mqtt5ConnAckException: CONNECT failed as CONNACK contained an Error Code: BAD_USER_NAME_OR_PASSWORD.
disconnected source: CLIENT cause: com.hivemq.client.mqtt.mqtt5.exceptions.Mqtt5ConnAckException: CONNECT failed as CONNACK contained an Error Code: BAD_USER_NAME_OR_PASSWORD.

Thank you for identifying another bug. This should be a Mqtt3ConnAckException. I will create the bug tickets now and they will be solved in the version 1.1.1

[adamcrane]

Thanks for the help @SgtSilvio! Closing the issue now.