title: Bad queries error handling in Spark Cassandra Connector
body: <p>I have a Spark Streaming application that has multiple data flows (DStreams) that write in the same Cassandra table. When testing my application on a large amount of random data, I'm receiving an error from Spark Cassandra Connector that has very little information helpful for debugging. The error looks like this:</p>
  
  <pre><code>java.util.concurrent.ExecutionException: com.datastax.driver.core.exceptions.InvalidQueryException: Key may not be empty
      at com.baynote.shaded.com.google.common.util.concurrent.AbstractFuture$Sync.getValue(AbstractFuture.java:299)
      at com.baynote.shaded.com.google.common.util.concurrent.AbstractFuture$Sync.get(AbstractFuture.java:286)
      at com.baynote.shaded.com.google.common.util.concurrent.AbstractFuture.get(AbstractFuture.java:116)
      at com.datastax.spark.connector.rdd.CassandraJoinRDD$$anonfun$fetchIterator$1.apply(CassandraJoinRDD.scala:268)
      at com.datastax.spark.connector.rdd.CassandraJoinRDD$$anonfun$fetchIterator$1.apply(CassandraJoinRDD.scala:268)
      at scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:371)
      at com.datastax.spark.connector.util.CountingIterator.hasNext(CountingIterator.scala:12)
      at scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:371)
      at org.apache.spark.util.collection.ExternalSorter.insertAll(ExternalSorter.scala:189)
      at org.apache.spark.shuffle.sort.SortShuffleWriter.write(SortShuffleWriter.scala:64)
      at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:73)
      at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)
      at org.apache.spark.scheduler.Task.run(Task.scala:89)
      at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:213)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
      at java.lang.Thread.run(Thread.java:745)
  Caused by: com.datastax.driver.core.exceptions.InvalidQueryException: Key may not be empty
      at com.datastax.driver.core.Responses$Error.asException(Responses.java:136)
      at com.datastax.driver.core.DefaultResultSetFuture.onSet(DefaultResultSetFuture.java:179)
      at com.datastax.driver.core.RequestHandler.setFinalResult(RequestHandler.java:184)
      at com.datastax.driver.core.RequestHandler.access$2500(RequestHandler.java:43)
      at com.datastax.driver.core.RequestHandler$SpeculativeExecution.setFinalResult(RequestHandler.java:798)
      at com.datastax.driver.core.RequestHandler$SpeculativeExecution.onSet(RequestHandler.java:617)
      at com.datastax.driver.core.Connection$Dispatcher.channelRead0(Connection.java:1005)
      at com.datastax.driver.core.Connection$Dispatcher.channelRead0(Connection.java:928)
      at io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)
      at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:308)
      at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:294)
      at io.netty.handler.timeout.IdleStateHandler.channelRead(IdleStateHandler.java:266)
      at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:308)
      at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:294)
      at io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:103)
      at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:308)
      at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:294)
      at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:244)
      at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:308)
      at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:294)
      at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:846)
      at io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:831)
      at io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:346)
      at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:254)
      at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:111)
      ... 1 more
  </code></pre>
  
  <p>The problem with it is that I can't tell which DStream, and which data cause it. I can check every DStream that writes to Cassandra, or write my own data validator, but I'm looking for a more generic solution.</p>
  
  <p>The other problem is that it the error kills the whole job instead of ignoring it and continue writing other data. Basically in case of simple non-spark writing I would catch the exception, log it and continue writing the rest of the data. Is there a way to do something like that in Spark Cassandra Connector?</p>
  
  <p>So is there something I can do about those two problems?</p>