[Bug]: MalformedInputException in MQTT Source connector in Azure K8s #939

mukkchir · 2023-05-05T07:21:49Z

Hi,

I have deployed Strimzi kafka clusters in azure k8s. After deploying the MQTT source connector, I get the following error upon
kubectl describe kctr -n kafka -

Error Trace

java.nio.charset.MalformedInputException: Input length = 1
              at java.base/java.nio.charset.CoderResult.throwException(CoderResult.java:274)
              at java.base/sun.nio.cs.StreamDecoder.implRead(StreamDecoder.java:326)
              at java.base/sun.nio.cs.StreamDecoder.read(StreamDecoder.java:188)
              at java.base/java.io.InputStreamReader.read(InputStreamReader.java:177)
              at java.base/java.io.BufferedReader.fill(BufferedReader.java:162)
              at java.base/java.io.BufferedReader.read(BufferedReader.java:183)
              at scala.io.BufferedSource.$anonfun$iter$2(BufferedSource.scala:41)
              at scala.io.Codec.wrap(Codec.scala:74)
              at scala.io.BufferedSource.$anonfun$iter$1(BufferedSource.scala:41)
              at scala.runtime.java8.JFunction0$mcI$sp.apply(JFunction0$mcI$sp.scala:17)
              at scala.collection.Iterator$$anon$27.next(Iterator.scala:1135)
              at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:637)
              at scala.collection.Iterator$$anon$9.hasNext(Iterator.scala:583)
              at scala.io.Source.hasNext(Source.scala:253)
              at scala.collection.Iterator.isEmpty(Iterator.scala:466)
              at scala.collection.Iterator.isEmpty$(Iterator.scala:466)
              at scala.io.Source.isEmpty(Source.scala:205)
              at scala.collection.IterableOnceOps.mkString(IterableOnce.scala:1165)
              at scala.collection.IterableOnceOps.mkString$(IterableOnce.scala:1164)
              at scala.io.Source.mkString(Source.scala:205)
              at scala.collection.IterableOnceOps.mkString(IterableOnce.scala:1179)
              at scala.collection.IterableOnceOps.mkString$(IterableOnce.scala:1190)
              at scala.io.Source.mkString(Source.scala:205)
              at com.datamountaineer.streamreactor.connect.mqtt.source.MqttSourceTask.start(MqttSourceTask.scala:48)
              at org.apache.kafka.connect.runtime.AbstractWorkerSourceTask.initializeAndStart(AbstractWorkerSourceTask.java:274)
              at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:200)
              at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:257)
              at org.apache.kafka.connect.runtime.AbstractWorkerSourceTask.run(AbstractWorkerSourceTask.java:75)
              at org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:177)
              at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539)
              at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
              at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
              at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
              at java.base/java.lang.Thread.run(Thread.java:833)

MQTT source connector config

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: mqtt-hm-source-connector
  labels:
    strimzi.io/cluster: kafka-connect-cluster
spec:
  class: com.datamountaineer.streamreactor.connect.mqtt.source.MqttSourceConnector
  tasksMax: 1
  config:
    topics: test
    connect.mqtt.clean: false
    connect.mqtt.kcql: INSERT INTO test SELECT * FROM `MY_TOPIC` WITHCONVERTER=`com.datamountaineer.streamreactor.connect.converters.source.JsonSimpleConverter`
    connect.mqtt.converter.throw.on.error: true
    connect.mqtt.client.id: "CLIENT_ID"
    connect.mqtt.hosts: "ssl://mqtt.broker.com:8883" (THIS IS A THIRD PARTY CONNECTION)
    connect.mqtt.ssl.ca.cert: "/mycerts/ca_file.pem"
    connect.mqtt.ssl.cert: "/mycerts/crt_file.pem.crt"
    connect.mqtt.ssl.key: "/mycerts/key_file.pem.key"
    connect.mqtt.service.quality: 1
    connect.progress.enabled: true
    connect.mqtt.log.message: true
    errors.log.include.messages: true
    errors.log.enable: true
    key.converter: org.apache.kafka.connect.storage.StringConverter
    value.converter: org.apache.kafka.connect.json.JsonConverter
    key.converter.schemas.enable: false
    value.converter.schemas.enable: false

Strimzi kafka config

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: messaging
spec:
  kafka:
    version: 3.4.0
    replicas: 3
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: false
      - name: tls
        port: 9093
        type: internal
        tls: true
      - name: external
        port: 9094
        type: nodeport
        tls: false
    config:
      offsets.topic.replication.factor: 1
      transaction.state.log.replication.factor: 1
      transaction.state.log.min.isr: 1
      default.replication.factor: 1
      min.insync.replicas: 1
      inter.broker.protocol.version: "3.4"
    storage:
      type: jbod
      volumes:
      - id: 0
        type: persistent-claim
        size: 10Gi
        deleteClaim: false
  zookeeper:
    replicas: 3
    storage:
      type: persistent-claim
      size: 10Gi
      deleteClaim: false
  entityOperator:
    topicOperator: {}
    userOperator: {}

Steps to reproduce in azure k8s

kubectl create namespace kafka
kubectl create -f 'https://strimzi.io/install/latest?namespace=kafka' -n kafka
kubectl create -f ABOVE_KAFKA_CONFIG.yaml -n kafka
kubectl create -f ABOVE_MQTT_SOURCE_CONFIG.yaml -n kafka
kubectl get kctr -n kafka (connector status appears not ready)
kubectl describe kctr -n kafka (here the error trace can be seen)
kubectl delete kctr MQTT_CONNECTOR_NAME -n kafka (to re-attach the connector with changes its better to delete the existing connector and repeat step-6)

Previously used versions

kafka - 3.2.0
Strimzi cluster operator - 0.29.0
stream reactor kafka connector version- 4.0.0

Currently used versions

Kafka - 3.4.0
Strimzi operator- 0.34.0
stream reactor kafka connector version- 4.2.0

In both of these versions, the same error above persisted. I think I'm missing some config but not sure what it is. Any guidance will be greatly helpful.

The text was updated successfully, but these errors were encountered:

davidsloan · 2023-05-10T18:21:40Z

@mukkchir thanks for the detailed report. Is it possible that your default charset is set to something different to UTF-8?

Please could you try this build that I have prepared, with a possible fix.

Let me know how it goes.

mukkchir · 2023-05-12T10:07:07Z

@davidsloan you are really a lifesaver and huge kudos to you for giving me a quick fix. It's working!
I'm now able to connect to the stream successfully. I would like to keep this ticket open for a few more days as I'm doing some tests on kafka and the connector you provided. I will report back to this thread in case of any errors/exceptions.

mukkchir · 2023-05-15T08:35:59Z

@davidsloan The connector is constantly looping over "error handling message on topic" and re-subscribing to topic with error description below:


	2023-05-15 08:24:04,304 DEBUG [mqtt-hm-source-connector|task-0] Message received on topic [TOPIC_NAME]. Message id =[3] , isDuplicate=false, payload=ENCRYPTED_DATA (com.datamountaineer.streamreactor.connect.mqtt.source.MqttManager) [MQTT Call: afaf2189-2fc7-4288-8ed4-0c90846c5ef3]
2023-05-15 08:24:04,304 ERROR [mqtt-hm-source-connector|task-0] Error handling message with id:3 on topic:TOPIC_NAME (com.datamountaineer.streamreactor.connect.mqtt.source.MqttManager) [MQTT Call: afaf2189-2fc7-4288-8ed4-0c90846c5ef3]
org.apache.kafka.connect.errors.DataException: Struct schemas do not match.
	at org.apache.kafka.connect.data.ConnectSchema.validateValue(ConnectSchema.java:249)
	at org.apache.kafka.connect.data.ConnectSchema.validateValue(ConnectSchema.java:213)
	at org.apache.kafka.connect.data.ConnectSchema.validateValue(ConnectSchema.java:255)
	at org.apache.kafka.connect.data.Struct.put(Struct.java:216)
	at org.apache.kafka.connect.data.Struct.put(Struct.java:203)
	at com.datamountaineer.streamreactor.connect.converters.source.JsonSimpleConverter$.$anonfun$handleObject$2(JsonSimpleConverter.scala:142)
	at scala.collection.immutable.Map$Map1.foreach(Map.scala:258)
	at com.datamountaineer.streamreactor.connect.converters.source.JsonSimpleConverter$.handleObject(JsonSimpleConverter.scala:142)
	at com.datamountaineer.streamreactor.connect.converters.source.JsonSimpleConverter$.convert(JsonSimpleConverter.scala:117)
	at com.datamountaineer.streamreactor.connect.converters.source.JsonSimpleConverter$.$anonfun$handleObject$1(JsonSimpleConverter.scala:135)
	at scala.collection.immutable.List.map(List.scala:246)
	at com.datamountaineer.streamreactor.connect.converters.source.JsonSimpleConverter$.handleObject(JsonSimpleConverter.scala:133)
	at com.datamountaineer.streamreactor.connect.converters.source.JsonSimpleConverter$.convert(JsonSimpleConverter.scala:117)
	at com.datamountaineer.streamreactor.connect.converters.source.JsonSimpleConverter$.$anonfun$handleObject$1(JsonSimpleConverter.scala:135)
	at scala.collection.immutable.List.map(List.scala:246)
    at com.datamountaineer.streamreactor.connect.converters.source.JsonSimpleConverter$.handleObject(JsonSimpleConverter.scala:133)
	at com.datamountaineer.streamreactor.connect.converters.source.JsonSimpleConverter$.convert(JsonSimpleConverter.scala:117)
	at com.datamountaineer.streamreactor.connect.converters.source.JsonSimpleConverter$.convert(JsonSimpleConverter.scala:98)
	at com.datamountaineer.streamreactor.connect.converters.source.JsonSimpleConverter.convert(JsonSimpleConverter.scala:46)
	at com.datamountaineer.streamreactor.connect.mqtt.source.MqttManager.messageArrived(MqttManager.scala:122)
	at org.eclipse.paho.client.mqttv3.internal.CommsCallback.deliverMessage(CommsCallback.java:519)
	at org.eclipse.paho.client.mqttv3.internal.CommsCallback.handleMessage(CommsCallback.java:417)
	at org.eclipse.paho.client.mqttv3.internal.CommsCallback.run(CommsCallback.java:214)
	at java.base/java.lang.Thread.run(Thread.java:833)
2023-05-15 08:24:04,305 WARN [mqtt-hm-source-connector|task-0] Connection lost. Re-connecting is set to true (com.datamountaineer.streamreactor.connect.mqtt.source.MqttManager) [MQTT Call: afaf2189-2fc7-4288-8ed4-0c90846c5ef3]
MqttException (0) - org.apache.kafka.connect.errors.DataException: Struct schemas do not match.
	at org.eclipse.paho.client.mqttv3.internal.CommsCallback.run(CommsCallback.java:228)
	at java.base/java.lang.Thread.run(Thread.java:833)
Caused by: org.apache.kafka.connect.errors.DataException: Struct schemas do not match.
	at org.apache.kafka.connect.data.ConnectSchema.validateValue(ConnectSchema.java:249)
	at org.apache.kafka.connect.data.ConnectSchema.validateValue(ConnectSchema.java:213)
	at org.apache.kafka.connect.data.ConnectSchema.validateValue(ConnectSchema.java:255)
	at org.apache.kafka.connect.data.Struct.put(Struct.java:216)
	at org.apache.kafka.connect.data.Struct.put(Struct.java:203)
	at com.datamountaineer.streamreactor.connect.converters.source.JsonSimpleConverter$.$anonfun$handleObject$2(JsonSimpleConverter.scala:142)
	at scala.collection.immutable.Map$Map1.foreach(Map.scala:258)
	at com.datamountaineer.streamreactor.connect.converters.source.JsonSimpleConverter$.handleObject(JsonSimpleConverter.scala:142)
	at com.datamountaineer.streamreactor.connect.converters.source.JsonSimpleConverter$.convert(JsonSimpleConverter.scala:117)
	at com.datamountaineer.streamreactor.connect.converters.source.JsonSimpleConverter$.$anonfun$handleObject$1(JsonSimpleConverter.scala:135)
	at scala.collection.immutable.List.map(List.scala:246)
	at com.datamountaineer.streamreactor.connect.converters.source.JsonSimpleConverter$.handleObject(JsonSimpleConverter.scala:133)
	at com.datamountaineer.streamreactor.connect.converters.source.JsonSimpleConverter$.convert(JsonSimpleConverter.scala:117)
	at com.datamountaineer.streamreactor.connect.converters.source.JsonSimpleConverter$.$anonfun$handleObject$1(JsonSimpleConverter.scala:135)
	at scala.collection.immutable.List.map(List.scala:246)
	at com.datamountaineer.streamreactor.connect.converters.source.JsonSimpleConverter$.handleObject(JsonSimpleConverter.scala:133)
	at com.datamountaineer.streamreactor.connect.converters.source.JsonSimpleConverter$.convert(JsonSimpleConverter.scala:117)
	at com.datamountaineer.streamreactor.connect.converters.source.JsonSimpleConverter$.convert(JsonSimpleConverter.scala:98)
	at com.datamountaineer.streamreactor.connect.converters.source.JsonSimpleConverter.convert(JsonSimpleConverter.scala:46)
	at com.datamountaineer.streamreactor.connect.mqtt.source.MqttManager.messageArrived(MqttManager.scala:122)
	at org.eclipse.paho.client.mqttv3.internal.CommsCallback.deliverMessage(CommsCallback.java:519)
	at org.eclipse.paho.client.mqttv3.internal.CommsCallback.handleMessage(CommsCallback.java:417)
	at org.eclipse.paho.client.mqttv3.internal.CommsCallback.run(CommsCallback.java:214)
	... 1 more

Not sure what is happening. How to properly handle this in the connector config? Also, what happens if the task.max is set to 2 or more?

davidsloan · 2023-05-15T08:53:07Z

Not sure what is happening. How to properly handle this in the connector config? Also, what happens if the task.max is set to 2 or more?

@mukkchir are you able to send me some samples of your data? This would help troubleshooting this issue. You can find me on our community slack if you are not comfortable sending it on here. It is not obvious to me what the problem is without more information.

mukkchir · 2023-05-15T09:35:04Z

@davidsloan Sure! Here is the sample data on which the error is occurring:

{
   "_id":{
      "$oid":"VARCHAR"
   },
   "capability":"diagnostics",
   "data":{
      "diagnostics":{
         "tire_pressures":[
            {
               "data":{
                  "location":"front_left",
                  "pressure":{
                     "unit":"kilopascals",
                     "value":{
                        "$numberDouble":"230.0"
                     }
                  }
               },
               "timestamp":"2022-11-17T12:33:22.000Z"
            },
            {
               "data":{
                  "location":"front_right",
                  "pressure":{
                     "unit":"kilopascals",
                     "value":{
                        "$numberDouble":"230.0"
                     }
                  }
               },
               "timestamp":"2022-11-17T12:33:19.000Z"
            },
            {
               "data":{
                  "location":"rear_left",
                  "pressure":{
                     "unit":"kilopascals",
                     "value":{
                        "$numberDouble":"257.5"
                     }
                  }
               },
               "timestamp":"2022-11-17T12:33:50.000Z"
            },
            {
               "data":{
                  "location":"rear_right",
                  "pressure":{
                     "unit":"kilopascals",
                     "value":{
                        "$numberDouble":"252.5"
                     }
                  }
               },
               "timestamp":"2022-11-17T12:33:10.000Z"
            }
         ]
      }
   },
   "message_id":"VARCHAR",
   "property":"tire_pressures",
   "version":{
      "$numberInt":"1"
   },
   "num":"VARCHAR"
}

davidsloan · 2023-05-15T12:09:05Z

I'm sorry I'm not seeing a problem with this input

I've put together a quick test here which you can see is working:
#942

Is this message any different from previous messages? Is there any schema change, missing/added fields etc here since the preceding message?

mukkchir · 2023-05-15T12:19:24Z

@davidsloan This is just the sample data. Since the data is encrypted, I'm working on decrypting it. I will post the actual data soon. Is there a config where I can move to next message if the previous message is faulty?

spike83 · 2023-05-16T04:48:00Z

@davidsloan I've got the same issue with the jms source connector. When can we expect a release with the fix included? Thank you very much 🙏.

mukkchir · 2023-05-16T08:52:03Z

@davidsloan seems like its the same data which I posted here.

davidsloan · 2023-05-16T09:47:37Z

@mukkchir this is message id 3, correct? Are message id 1 & 2 any different?

davidsloan · 2023-05-16T12:41:29Z

@mukkchir it seems like the issue might be that a different charset encoding is being configured by default. Please can you try setting this environment variable in the Connect container:

LC_ALL=en_US.UTF-8

And let me know how you get on.

erezblm · 2023-06-10T20:58:30Z

Hey,

@davidsloan
I was getting the exact same error with MalfromedInputException using strimzi and AKS deployment.
Your uploaded package fix the issue for me so thanks a lot for that, waiting for an update for official release.

I tried the AC_ALL=en_US.UTF-8 but it was ignore by the bash at the beginning of the connect:
"bash: warning: setlocale: LC_ALL: cannot change locale (en_US.UTF-8)"

Thanks.

erezblm · 2023-08-16T06:38:14Z

Hi @davidsloan, I'm now using the "sink connector" and I have the same problem with the malformed input.
Is your fix handling only the source connector?
Any idea if an official release will be soon with those fixes?

Thanks a lot, Erez.

sunmeplz · 2023-09-05T12:09:01Z

Im using mqtt connector inside debezium kafka connect docker, the issue was solved by adding LANG=en_US.UTF-8 variable to docker container

davidsloan added source mqtt labels May 11, 2023

davidsloan mentioned this issue May 12, 2023

Ascii Art Loading Fix (All Connectors) #941

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Bug]: MalformedInputException in MQTT Source connector in Azure K8s #939

[Bug]: MalformedInputException in MQTT Source connector in Azure K8s #939

mukkchir commented May 5, 2023

davidsloan commented May 10, 2023

mukkchir commented May 12, 2023

mukkchir commented May 15, 2023 •

edited

Loading

davidsloan commented May 15, 2023

mukkchir commented May 15, 2023 •

edited

Loading

davidsloan commented May 15, 2023 •

edited

Loading

mukkchir commented May 15, 2023 •

edited

Loading

spike83 commented May 16, 2023

mukkchir commented May 16, 2023

davidsloan commented May 16, 2023

davidsloan commented May 16, 2023

erezblm commented Jun 10, 2023 •

edited

Loading

erezblm commented Aug 16, 2023

sunmeplz commented Sep 5, 2023

[Bug]: MalformedInputException in MQTT Source connector in Azure K8s #939

[Bug]: MalformedInputException in MQTT Source connector in Azure K8s #939

Comments

mukkchir commented May 5, 2023

Error Trace

MQTT source connector config

Strimzi kafka config

Steps to reproduce in azure k8s

Previously used versions

Currently used versions

davidsloan commented May 10, 2023

mukkchir commented May 12, 2023

mukkchir commented May 15, 2023 • edited Loading

davidsloan commented May 15, 2023

mukkchir commented May 15, 2023 • edited Loading

davidsloan commented May 15, 2023 • edited Loading

mukkchir commented May 15, 2023 • edited Loading

spike83 commented May 16, 2023

mukkchir commented May 16, 2023

davidsloan commented May 16, 2023

davidsloan commented May 16, 2023

erezblm commented Jun 10, 2023 • edited Loading

erezblm commented Aug 16, 2023

sunmeplz commented Sep 5, 2023

mukkchir commented May 15, 2023 •

edited

Loading

mukkchir commented May 15, 2023 •

edited

Loading

davidsloan commented May 15, 2023 •

edited

Loading

mukkchir commented May 15, 2023 •

edited

Loading

erezblm commented Jun 10, 2023 •

edited

Loading