Issues receiving messages from Pulsar to MQTT ("Namespace is being unloaded, cannot add topic" and "Invalid unknonwn tag type: 4")

[tonisojandu-sympower]

We have been evaluating MQTT-over-Pulsar (MoP) project for our purposes:
https://github.com/streamnative/mop

We configured our three broker 2.8.1 Pulsar cloud cluster with MoP v2.8.1.26 NAR. We sent both plain String and Avro messages with a Pulsar client and listened for them with MQTT client. We tested the same thing also on single PC setup where all 3 brokers, 3 zookeepers and 3 bookies were initiated from apachepulsar/pulsar-all:2.8.1 Docker image.

We experimented with both:

• Eclipse Paho client: https://www.eclipse.org/paho/index.php?page=clients/java/index.php (We are using Paho with our current Mosquitto server setup)
• FuseSource client suggested in the MoP README.md: https://github.com/fusesource/mqtt-client
  When connecting with Paho client connection, we received an error from the broker:

Exception in thread "main" Connection lost (32109) - java.io.EOFException
    at org.eclipse.paho.client.mqttv3.internal.CommsReceiver.run(CommsReceiver.java:197)
    at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: java.io.EOFException
    at java.base/java.io.DataInputStream.readByte(DataInputStream.java:272)
    at org.eclipse.paho.client.mqttv3.internal.wire.MqttInputStream.readMqttWireMessage(MqttInputStream.java:92)
    at org.eclipse.paho.client.mqttv3.internal.CommsReceiver.run(CommsReceiver.java:137)
    ... 1 more

We received the same with FuseSource client, but it errored out silently and ended up in some internal error loop. Our broker logged the following exception:

java.util.concurrent.CompletionException: org.apache.pulsar.broker.service.BrokerServiceException$ServiceUnitNotReadyException: Namespace is being unloaded, cannot add topic persistent://public/default/moptesting0
    at java.util.concurrent.CompletableFuture.encodeRelay(CompletableFuture.java:367) ~[?:?]
    at java.util.concurrent.CompletableFuture.uniComposeStage(CompletableFuture.java:1108) ~[?:?]
    at java.util.concurrent.CompletableFuture.thenCompose(CompletableFuture.java:2235) [?:?]
    at io.streamnative.pulsar.handlers.mqtt.utils.PulsarTopicUtils.getTopicReference(PulsarTopicUtils.java:63) ~[?:?]
    at io.streamnative.pulsar.handlers.mqtt.utils.PulsarTopicUtils.getOrCreateSubscription(PulsarTopicUtils.java:70) ~[?:?]
    at io.streamnative.pulsar.handlers.mqtt.support.DefaultProtocolMethodProcessorImpl.lambda$doSubscribe$4(DefaultProtocolMethodProcessorImpl.java:351) ~[?:?]
    at java.util.concurrent.CompletableFuture.uniComposeStage(CompletableFuture.java:1106) [?:?]
    at java.util.concurrent.CompletableFuture.thenCompose(CompletableFuture.java:2235) [?:?]
    at io.streamnative.pulsar.handlers.mqtt.support.DefaultProtocolMethodProcessorImpl.doSubscribe(DefaultProtocolMethodProcessorImpl.java:347) [O9qARjCbhPqAltc4SY2aTw/:?]
    at io.streamnative.pulsar.handlers.mqtt.support.DefaultProtocolMethodProcessorImpl.processSubscribe(DefaultProtocolMethodProcessorImpl.java:308) [O9qARjCbhPqAltc4SY2aTw/:?]
    at io.streamnative.pulsar.handlers.mqtt.MQTTInboundHandler.channelRead(MQTTInboundHandler.java:67) [O9qARjCbhPqAltc4SY2aTw/:?]
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379) [io.netty-netty-transport-4.1.66.Final.jar:4.1.66.Final]
    ...
    at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) [io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final]
    at java.lang.Thread.run(Thread.java:829) [?:?]
 Caused by: org.apache.pulsar.broker.service.BrokerServiceException$ServiceUnitNotReadyException: Namespace is being unloaded, cannot add topic persistent://public/default/moptesting0
    at org.apache.pulsar.broker.service.BrokerService.createPersistentTopic(BrokerService.java:1218) ~[org.apache.pulsar-pulsar-broker-2.8.1.jar:2.8.1]
    at org.apache.pulsar.broker.service.BrokerService.lambda$loadOrCreatePersistentTopic$42(BrokerService.java:1190) ~[org.apache.pulsar-pulsar-broker-2.8.1.jar:2.8.1]
    at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:859) ~[?:?]
    at java.util.concurrent.CompletableFuture.uniWhenCompleteStage(CompletableFuture.java:883) ~[?:?]
    at java.util.concurrent.CompletableFuture.whenComplete(CompletableFuture.java:2251) ~[?:?]
    at org.apache.pulsar.broker.service.BrokerService.loadOrCreatePersistentTopic(BrokerService.java:1181) ~[org.apache.pulsar-pulsar-broker-2.8.1.jar:2.8.1]
    at org.apache.pulsar.broker.service.BrokerService.lambda$getTopic$26(BrokerService.java:894) ~[org.apache.pulsar-pulsar-broker-2.8.1.jar:2.8.1]
    at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap$Section.put(ConcurrentOpenHashMap.java:276) ~[org.apache.pulsar-pulsar-common-2.8.1.jar:2.8.1]
    at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap.computeIfAbsent(ConcurrentOpenHashMap.java:129) ~[org.apache.pulsar-pulsar-common-2.8.1.jar:2.8.1]
    at org.apache.pulsar.broker.service.BrokerService.getTopic(BrokerService.java:893) ~[org.apache.pulsar-pulsar-broker-2.8.1.jar:2.8.1]
    at io.streamnative.pulsar.handlers.mqtt.utils.PulsarTopicUtils.lambda$getTopicReference$0(PulsarTopicUtils.java:63) ~[?:?]
    at java.util.concurrent.CompletableFuture.uniComposeStage(CompletableFuture.java:1106) ~[?:?]
    ... 32 more

We got around this by manually choosing the one broker where unloading was not being reported. Once we had manually chosen the broker and sent both String and Avro messages, we were able to receive them with our local PC cluster setup. If we experimented with ungraceful broker shutdown. Paho client was unable to fail over to the next URI, provided to it. This can be possibly be a client (configuration) issue. FuseSource client does not seem to provide an option to declare failover URIs. Workaround for both would be to create a failover wrapper around the clients and just round-robin to a working one, but it is a bit tragic that we were unable to use the Paho builtin failover logic.

But more importantly we were unable to receive Pulsar to MQTT messages from our Pulsar cloud cluster. Neither Paho nor FuseSource client received any messages and on message broker tied to the MQTT client logged following exception (note that this may be out of order, since our FluentBit split stack trace into separate log events):

15:02:24.093 [BookKeeperClientWorker-OrderedExecutor-0-0] ERROR org.apache.bookkeeper.common.util.SafeRunnable - Unexpected throwable caught"
java.lang.IllegalArgumentException: Invalid unknonwn tag type: 4"
at org.apache.bookkeeper.mledger.impl.OpReadEntry.lambda$checkReadCompletion$2(OpReadEntry.java:156) ~[org.apache.pulsar-managed-ledger-2.8.1.jar:2.8.1]"
at org.apache.bookkeeper.mledger.util.SafeRun$1.safeRun(SafeRun.java:32) ~[org.apache.pulsar-managed-ledger-2.8.1.jar:2.8.1]"
at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) [io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final]"
at org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers.readEntriesComplete(PersistentDispatcherMultipleConsumers.java:477) ~[org.apache.pulsar-pulsar-broker-2.8.1.jar:2.8.1]"
at org.apache.bookkeeper.common.util.SafeRunnable.run(SafeRunnable.java:36) [org.apache.bookkeeper-bookkeeper-common-4.14.2.jar:4.14.2]"
at org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers.sendMessagesToConsumers(PersistentDispatcherMultipleConsumers.java:546) ~[org.apache.pulsar-pulsar-broker-2.8.1.jar:2.8.1]"
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?]"
at java.lang.Thread.run(Thread.java:829) [?:?]"
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?]"
at io.streamnative.pulsar.handlers.mqtt.utils.PulsarMessageConverter.toMqttMessages(PulsarMessageConverter.java:79) ~[?:?]"
at io.streamnative.pulsar.handlers.mqtt.support.MQTTConsumer.sendMessages(MQTTConsumer.java:43) ~[?:?]"
at io.streamnative.pulsar.handlers.mqtt.support.MQTTConsumer.sendMessages(MQTTConsumer.java:91) ~[?:?]"
at org.apache.pulsar.common.protocol.Commands.deSerializeSingleMessageInBatch(Commands.java:1627) ~[org.apache.pulsar-pulsar-common-2.8.1.jar:2.8.1]"
at org.apache.pulsar.common.api.proto.SingleMessageMetadata.parseFrom(SingleMessageMetadata.java:470) ~[org.apache.pulsar-pulsar-common-2.8.1.jar:2.8.1]"
at org.apache.pulsar.common.api.proto.LightProtoCodec.skipUnknownField(LightProtoCodec.java:270) ~[org.apache.pulsar-pulsar-common-2.8.1.jar:2.8.1]"

• Are there recommendations on how we should properly choosing the broker with which we should be communicating and any suggestions for MQTT client connection failovers?
• Is Invalid unknonwn tag type: 4 something unrelated to MoP? We have seen other threads around internet discussing similar error. Or are might there be something we have misconfigured in our Pulsar cloud setup since that one is not running as Docker apachepulsar/pulsar-all:2.8.1 images? Additionally our cloud cluster uses SSL (unlike the local PC setup).

Our test client code:

build.gradle.kts

plugins {
  id("java")
  id("java-library")
  kotlin("jvm") version "1.5.21"
  id("com.github.davidmc24.gradle.plugin.avro") version "1.2.1"
}

repositories {
  mavenCentral()
  mavenLocal()
}

dependencies {
  implementation("org.apache.pulsar:pulsar-client-original:2.8.0")
  implementation("org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.2.5")
  implementation("ch.qos.logback:logback-core:1.2.7")
  implementation("ch.qos.logback:logback-classic:1.2.7")
  implementation("org.fusesource.mqtt-client:mqtt-client:1.16")
}

FuseSourceMQTTListener.kt:

package net.sympower.mop.test

import org.fusesource.mqtt.client.MQTT
import org.fusesource.mqtt.client.Message
import org.fusesource.mqtt.client.QoS
import org.fusesource.mqtt.client.Topic

fun main() {
  val mqtt = MQTT()
  mqtt.setHost("sympower-pulsar-broker1", 1883)
  val connection = mqtt.blockingConnection()
  connection.connect()
  val topics = arrayOf(Topic(MQTT_TOPIC_NAME, QoS.AT_LEAST_ONCE))
  connection.subscribe(topics)

  println("Connected")

  while (true) {
    val received: Message = connection.receive()
    println("Got: " + String(received.payload))
  }
}

PahoMQTTListener.kt

package net.sympower.mop.test

import org.eclipse.paho.client.mqttv3.*
import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence

const val MQTT_TOPIC_NAME = "persistent://public/default/moptesting0"

val MQTT_URLS: Array<String> = arrayOf(
  "tcp://127.0.0.1:2883",
  "tcp://127.0.0.1:3883",
  "tcp://127.0.0.1:4883",
)

fun main() {
  val options = MqttConnectOptions()
  options.serverURIs = MQTT_URLS
  options.connectionTimeout = 10
  options.isAutomaticReconnect = true

  val client = MqttClient(MQTT_URLS[0], "paho-mqtt-listener", MqttDefaultFilePersistence())
  client.setCallback(Callback())
  client.connect(options)

  client.subscribe(MQTT_TOPIC_NAME)
}

private class Callback : MqttCallback {
  override fun connectionLost(cause: Throwable) {
    println("Lost connection ${cause.message}")
  }

  override fun messageArrived(topic: String, message: MqttMessage) {
    println("Got from $topic: ${String(message.payload)}")
  }

  override fun deliveryComplete(token: IMqttDeliveryToken) {}

}

[sijie]

@Technoboy- Can you take a look at this issue?

[Technoboy-]

@Technoboy- Can you take a look at this issue?

Yes, I'm working on it.

[Technoboy-]

Hi, @tonisojandu-sympower .
Sorry for the late response.
I try to reproduce in my local env, but the doesn't meet the "Invalid unknonwn tag type: 4 issue.

pulsar broker	pulsar client	mop	mqtt client	result
2.8.1	2.8.1	2.8.1	fusesource.client-1.16	ok
2.8.1	2.8.16	2.8.1	fusesource.client-1.16	ok
2.8.1	2.8.16	2.8.0	fusesource.client-1.16	ok

Anyway, We recommend you to use the same version, like pulsar-broker-2.8.1, pulsar-client-2.8.1,mop-version-2.8.1.

For Namespace is being unloaded, cannot add topic persistent://public/default/moptesting0,
if the log print all the time, this may be a bug with fix apache/pulsar#12650.
But in the normal case, when the broker shutdown or the namespace or namespace bundle is going to unload, the log may exist.

MoP is a protocol handler supported by Pulsar, currently, protocol handler implementations like MoP, AoP, KoP didn't have a service like Pulsar proxy to discover these nodes. but we are working on this.

[tonisojandu-sympower]

Hi, @Technoboy-
Thank you for taking a look. We are our self also unsure where the issue relies, since we did not encounter the Invalid unknonwn tag type: 4 in our Docker cluster, but do see it in our direct installation cluster. As said, SSL is also something that differs between these clusters.

Anyway thank you for taking you time! We will be keeping an eye on apache/pulsar#12650 and tentatively investigate this on our own. If we find anything, we will post it here.