title: Spark Streaming application fails with KafkaException: String exceeds the maximum size or with IllegalArgumentException
body: <p>TL;DR:</p>
  
  <p>My very simple Spark Streaming application fails in the driver with the "KafkaException: String exceeds the maximum size". I see the same exception in the executor but I also found somewhere down the executor's logs an IllegalArgumentException with no other information in it</p>
  
  <p>Full problem:</p>
  
  <p>I'm using Spark Streaming to read some messages from a Kafka topic.
  This is what I'm doing:</p>
  
  <pre><code>val conf = new SparkConf().setAppName("testName")
  val streamingContext = new StreamingContext(new SparkContext(conf), Milliseconds(millis))
  val kafkaParams = Map(
        "metadata.broker.list" -&gt; "somevalidaddresshere:9092",
        "auto.offset.reset" -&gt; "largest"
      )
  val topics = Set("data")
  val stream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
        streamingContext,
        kafkaParams,
        topics
      ).map(_._2) // only need the values not the keys
  </code></pre>
  
  <p>What I'm doing with the Kafka data is only printing it using: </p>
  
  <pre><code>stream.print()
  </code></pre>
  
  <p>My application obviously has more code than this but in order to locate my problem I stripped everything I possibly could from the code</p>
  
  <p>I'm trying to run this code on YARN.
  This is my spark submit line:</p>
  
  <pre><code>./spark-submit --class com.somecompany.stream.MainStream --master yarn --deploy-mode cluster myjar.jar hdfs://some.hdfs.address.here/user/spark/streamconfig.properties
  </code></pre>
  
  <p>The streamconfig.properties file is just a regular properties file which is probably irrelevant to the problem here</p>
  
  <p>After trying to execute the application it fails pretty quickly with the following exception on the driver:</p>
  
  <pre><code>16/05/10 06:15:38 WARN scheduler.TaskSetManager: Lost task 0.0 in stage 0.0 (TID 0, some.hdfs.address.here): kafka.common.KafkaException: String exceeds the maximum size of 32767.
      at kafka.api.ApiUtils$.shortStringLength(ApiUtils.scala:73)
      at kafka.api.TopicData$.headerSize(FetchResponse.scala:107)
      at kafka.api.TopicData.&lt;init&gt;(FetchResponse.scala:113)
      at kafka.api.TopicData$.readFrom(FetchResponse.scala:103)
      at kafka.api.FetchResponse$$anonfun$4.apply(FetchResponse.scala:170)
      at kafka.api.FetchResponse$$anonfun$4.apply(FetchResponse.scala:169)
      at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:251)
      at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:251)
      at scala.collection.immutable.Range.foreach(Range.scala:141)
      at scala.collection.TraversableLike$class.flatMap(TraversableLike.scala:251)
      at scala.collection.AbstractTraversable.flatMap(Traversable.scala:105)
      at kafka.api.FetchResponse$.readFrom(FetchResponse.scala:169)
      at kafka.consumer.SimpleConsumer.fetch(SimpleConsumer.scala:135)
      at org.apache.spark.streaming.kafka.KafkaRDD$KafkaRDDIterator.fetchBatch(KafkaRDD.scala:192)
      at org.apache.spark.streaming.kafka.KafkaRDD$KafkaRDDIterator.getNext(KafkaRDD.scala:208)
      at org.apache.spark.util.NextIterator.hasNext(NextIterator.scala:73)
      at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327)
      at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:308)
      at scala.collection.Iterator$class.foreach(Iterator.scala:727)
      at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
      at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:48)
      at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:103)
      at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:47)
      at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:273)
      at scala.collection.AbstractIterator.to(Iterator.scala:1157)
      at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:265)
      at scala.collection.AbstractIterator.toBuffer(Iterator.scala:1157)
      at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:252)
      at scala.collection.AbstractIterator.toArray(Iterator.scala:1157)
      at org.apache.spark.rdd.RDD$$anonfun$take$1$$anonfun$28.apply(RDD.scala:1328)
      at org.apache.spark.rdd.RDD$$anonfun$take$1$$anonfun$28.apply(RDD.scala:1328)
      at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1869)
      at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1869)
      at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
      at org.apache.spark.scheduler.Task.run(Task.scala:89)
      at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
      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)
  </code></pre>
  
  <p>I don't even see my code in the stack trace</p>
  
  <p>Examining the executor I found the same exception as in the driver but also buried deep down is the following exception:</p>
  
  <pre><code>16/05/10 06:40:47 ERROR executor.Executor: Exception in task 0.0 in stage 2.0 (TID 8)
  java.lang.IllegalArgumentException
      at java.nio.Buffer.limit(Buffer.java:275)
      at kafka.api.FetchResponsePartitionData$.readFrom(FetchResponse.scala:38)
      at kafka.api.TopicData$$anonfun$1.apply(FetchResponse.scala:100)
      at kafka.api.TopicData$$anonfun$1.apply(FetchResponse.scala:98)
      at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
      at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
      at scala.collection.immutable.Range.foreach(Range.scala:141)
      at scala.collection.TraversableLike$class.map(TraversableLike.scala:244)
      at scala.collection.AbstractTraversable.map(Traversable.scala:105)
      at kafka.api.TopicData$.readFrom(FetchResponse.scala:98)
      at kafka.api.FetchResponse$$anonfun$4.apply(FetchResponse.scala:170)
      at kafka.api.FetchResponse$$anonfun$4.apply(FetchResponse.scala:169)
      at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:251)
      at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:251)
      at scala.collection.immutable.Range.foreach(Range.scala:141)
      at scala.collection.TraversableLike$class.flatMap(TraversableLike.scala:251)
      at scala.collection.AbstractTraversable.flatMap(Traversable.scala:105)
      at kafka.api.FetchResponse$.readFrom(FetchResponse.scala:169)
      at kafka.consumer.SimpleConsumer.fetch(SimpleConsumer.scala:135)
      at org.apache.spark.streaming.kafka.KafkaRDD$KafkaRDDIterator.fetchBatch(KafkaRDD.scala:192)
      at org.apache.spark.streaming.kafka.KafkaRDD$KafkaRDDIterator.getNext(KafkaRDD.scala:208)
      at org.apache.spark.util.NextIterator.hasNext(NextIterator.scala:73)
      at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327)
      at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:308)
      at scala.collection.Iterator$class.foreach(Iterator.scala:727)
      at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
      at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:48)
      at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:103)
      at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:47)
      at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:273)
      at scala.collection.AbstractIterator.to(Iterator.scala:1157)
      at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:265)
      at scala.collection.AbstractIterator.toBuffer(Iterator.scala:1157)
      at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:252)
      at scala.collection.AbstractIterator.toArray(Iterator.scala:1157)
      at org.apache.spark.rdd.RDD$$anonfun$take$1$$anonfun$28.apply(RDD.scala:1328)
      at org.apache.spark.rdd.RDD$$anonfun$take$1$$anonfun$28.apply(RDD.scala:1328)
      at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1869)
      at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1869)
      at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
      at org.apache.spark.scheduler.Task.run(Task.scala:89)
      at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
      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)
  </code></pre>
  
  <p>I have no idea what is the IllegalArgument since no information is included</p>
  
  <p>The Spark version my YARN is using is 1.6.0. I also verified my pom contains Spark 1.6.0 and not an earlier version. My scope is "provided"</p>
  
  <p>I manually read the data from the exact same topic and the data there is just plain JSONs. The data there is not huge at all. Definitely smaller than 32767. Also I'm able to read this data using the regular command line consumer so that's weird</p>
  
  <p>Googling this exception sadly didn't provide any useful information</p>
  
  <p>Does anyone have any idea on how to understand what exactly is the problem here?</p>
  
  <p>Thanks in advance</p>