(hard to track) mqtt sometimes hangs resubscribing with clean session = false #832
Comments
|
Hello, |
|
And ... i think i found something that works reliably. I now do: before connecting the client. This way, the .subscribe() call never blocks more than 1 sec. since the subscribe works (it always works, its just paho that doesn't see it) even if it times out it continues fine. |
kervel
changed the title
|
The issue seems to still exist in version 6.0.0.
Could we check if there is a reconnect with clean session = false, and not subscribe in this case? --- a/kafka-connect-mqtt/src/main/scala/io/lenses/streamreactor/connect/mqtt/source/MqttManager.scala
+++ b/kafka-connect-mqtt/src/main/scala/io/lenses/streamreactor/connect/mqtt/source/MqttManager.scala
@@ -50,6 +50,7 @@ class MqttManager(
client.setCallback(this)
logger.info(s"Connecting to ${settings.connection}")
+ client.setTimeToWait(5000)
client.connect(options)
logger.info(s"Connected to ${settings.connection} as ${settings.clientId}")
@@ -165,9 +166,10 @@ class MqttManager(
val topic = sourceToTopicMap.keySet.toArray
val qos = Array.fill(sourceToTopicMap.keySet.size)(settings.mqttQualityOfService)
- if (reconnect)
+ if (reconnect && !options.isCleanSession())
logger.warn(s"Reconnected. Resubscribing to topic $topic...")
- client.subscribe(topic, qos)
+ else client.subscribe(topic, qos)
+
if (reconnect)
logger.warn(s"Resubscribed to topic $topic with QoS $qos")
else logger.info(s"Subscribed to topic $topic with QoS $qos")Is there a cleaner method? |
|
I could solve the issue by wrapping --- a/kafka-connect-mqtt/src/main/scala/io/lenses/streamreactor/connect/mqtt/source/MqttManager.scala
+++ b/kafka-connect-mqtt/src/main/scala/io/lenses/streamreactor/connect/mqtt/source/MqttManager.scala
@@ -28,6 +28,9 @@ import java.util
import java.util.Base64
import java.util.concurrent.LinkedBlockingQueue
import java.util.concurrent.TimeUnit
+
+import scala.concurrent.ExecutionContext.Implicits.global
+import scala.concurrent.Future
import scala.jdk.CollectionConverters.ListHasAsScala
class MqttManager(
@@ -167,9 +170,14 @@ class MqttManager(
if (reconnect)
logger.warn(s"Reconnected. Resubscribing to topic $topic...")
- client.subscribe(topic, qos)
- if (reconnect)
- logger.warn(s"Resubscribed to topic $topic with QoS $qos")
- else logger.info(s"Subscribed to topic $topic with QoS $qos")
+
+ Future {
+ client.subscribe(topic, qos)
+ if (reconnect)
+ logger.warn(s"Resubscribed to topic $topic with QoS $qos")
+ else logger.info(s"Subscribed to topic $topic with QoS $qos")
+ }
+
+ return
}
} |
|
I believe we're running into this very issue: |
What version of the Stream Reactor are you reporting this issue for?
3.0.1
Are you running the correct version of Kafka/Confluent for the Stream reactor release?
running on v2.7 (i adapted build.gradle to match my kafka version)
What is the expected behaviour?
What was observed?
I found various similar (but not exactly the same) bug reports logged against paho mqttv3. So i modified lenses to use mqttv5 instead, and i cannot reproduce the issue anymore.
The modifications are rather trivial apart from clean session now being called clean start, but i can share them if needed.
What is your connector properties configuration (my-connector.properties)?
mosquitto config
Please provide full log files (redact and sensitive information)
this is on the first start:
this is after a restart:
here, the last log message (subscribed) is missing because it never reaches that line of code. note that i added "okay starting subscriptions" print statement at the beginning of the connectionComplete function.
The text was updated successfully, but these errors were encountered: