[jira] [Created] (BAHIR-256) Using MQTT topic or timestamp throws ClassCastException

["Jonas Schmöle (Jira)" <ji...@apache.org>]

Jonas Schmöle created BAHIR-256:
-----------------------------------

             Summary: Using MQTT topic or timestamp throws ClassCastException
                 Key: BAHIR-256
                 URL: https://issues.apache.org/jira/browse/BAHIR-256
             Project: Bahir
          Issue Type: Bug
          Components: Spark Structured Streaming Connectors
    Affects Versions: Spark-2.4.0
            Reporter: Jonas Schmöle


In Spark Structured Streaming, with Spark 2.4.0, I can only use the *payload* and *id* columns of MQTT; using the *topic* or *timestamp* columns throws _ClassCastExceptions_ after the 0th batch. The error can be reproduced by using the [MQTTStreamWordCount.scala|https://github.com/apache/bahir/blob/master/sql-streaming-mqtt/examples/src/main/scala/org/apache/bahir/examples/sql/streaming/mqtt/MQTTStreamWordCount.scala] example as follows:

In line 56, replace 
{code:java}
.load(brokerUrl).selectExpr("CAST(payload AS STRING)").as[String]{code}
with
{code:java}
.load(brokerUrl).selectExpr("CAST(topic AS STRING)").as[String]{code}
or 
{code:java}
 .load(brokerUrl).select("topic").as[String]{code}
 

The full exception is as follows for *topic*:
{code:java}
java.lang.ClassCastException: java.lang.String cannot be cast to org.apache.spark.unsafe.types.UTF8Stringjava.lang.ClassCastException: java.lang.String cannot be cast to org.apache.spark.unsafe.types.UTF8String at org.apache.spark.sql.catalyst.expressions.BaseGenericInternalRow.getUTF8String(rows.scala:46) at org.apache.spark.sql.catalyst.expressions.BaseGenericInternalRow.getUTF8String$(rows.scala:46) at org.apache.spark.sql.catalyst.expressions.GenericInternalRow.getUTF8String(rows.scala:195) at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source) at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43) at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$2.hasNext(WholeStageCodegenExec.scala:636) at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460) at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460) at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:491) at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460) at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage2.agg_doAggregateWithKeys_0$(Unknown Source) at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage2.processNext(Unknown Source) at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43) at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$2.hasNext(WholeStageCodegenExec.scala:636) at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460) at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:125) at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99) at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:55) at org.apache.spark.scheduler.Task.run(Task.scala:123) at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:411) at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360) at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:414) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748)
{code}
A similar exception appears when the *timestamp* column is used: 
{code:java}
java.lang.ClassCastException: java.sql.Timestamp cannot be cast to java.lang.Longjava.lang.ClassCastException: java.sql.Timestamp cannot be cast to java.lang.Long at scala.runtime.BoxesRunTime.unboxToLong(BoxesRunTime.java:107) at org.apache.spark.sql.catalyst.expressions.BaseGenericInternalRow.getLong(rows.scala:42) at org.apache.spark.sql.catalyst.expressions.BaseGenericInternalRow.getLong$(rows.scala:42){code}
 Please also see [https://stackoverflow.com/questions/64825099/apache-spark-sql-get-json-object-java-lang-string-cannot-be-cast-to-org-apache-s,] which I believe is strongly related.



--
This message was sent by Atlassian Jira
(v8.3.4#803005)