Bringing this thread back as I'm seeing this exception on a production
kafka cluster.
I have two Spark streaming apps reading the same topic. App1 has batch
interval 2secs and app2 has 60secs.
Both apps are running on the same cluster on similar hardware. I see this
exception only in app2 and fairly consistently.
Difference I see between the apps is
App1
spark.streaming.kafka.maxRatePerPartition, 6000
batch interval 2 secs
App2
spark.streaming.kafka.maxRatePerPartition, 10000
batch interval 60 secs
All other kafka/spark related configs are same for both apps.
spark.streaming.kafka.consumer.poll.ms = 4096
spark.streaming.backpressure.enabled = true
Not sure if pre-fetching or caching is messing things up.
16/10/19 14:32:04 WARN TaskSetManager: Lost task 2.0 in stage 1780.0 (TID
12541, ip-10-150-20-200.ec2.internal): java.lang.AssertionError: assertion
failed: Failed to get records for spark-executor-StreamingEventSplitProd
mt_event 6 49091480 after polling for 4096
at scala.Predef$.assert(Predef.scala:170)
at org.apache.spark.streaming.kafka010.CachedKafkaConsumer.
get(CachedKafkaConsumer.scala:74)
at org.apache.spark.streaming.kafka010.KafkaRDD$
KafkaRDDIterator.next(KafkaRDD.scala:227)
at org.apache.spark.streaming.kafka010.KafkaRDD$
KafkaRDDIterator.next(KafkaRDD.scala:193)
at scala.collection.Iterator$$anon$11.next(Iterator.scala:409)
at scala.collection.Iterator$$anon$11.next(Iterator.scala:409)
at scala.collection.Iterator$$anon$21.next(Iterator.scala:838)
On Wed, Sep 7, 2016 at 3:55 PM, Cody Koeninger <c...@koeninger.org> wrote:
> That's not what I would have expected to happen with a lower cache
> setting, but in general disabling the cache isn't something you want
> to do with the new kafka consumer.
>
>
> As far as the original issue, are you seeing those polling errors
> intermittently, or consistently? From your description, it sounds
> like retry is working correctly.
>
>
> On Wed, Sep 7, 2016 at 2:37 PM, Srikanth <srikanth...@gmail.com> wrote:
> > Setting those two results in below exception.
> > No.of executors < no.of partitions. Could that be triggering this?
> >
> > 16/09/07 15:33:13 ERROR Executor: Exception in task 2.0 in stage 2.0
> (TID 9)
> > java.util.ConcurrentModificationException: KafkaConsumer is not safe for
> > multi-threaded access
> > at
> > org.apache.kafka.clients.consumer.KafkaConsumer.
> acquire(KafkaConsumer.java:1430)
> > at
> > org.apache.kafka.clients.consumer.KafkaConsumer.close(
> KafkaConsumer.java:1360)
> > at
> > org.apache.spark.streaming.kafka010.CachedKafkaConsumer$$
> anon$1.removeEldestEntry(CachedKafkaConsumer.scala:128)
> > at java.util.LinkedHashMap.afterNodeInsertion(Unknown Source)
> > at java.util.HashMap.putVal(Unknown Source)
> > at java.util.HashMap.put(Unknown Source)
> > at
> > org.apache.spark.streaming.kafka010.CachedKafkaConsumer$.
> get(CachedKafkaConsumer.scala:158)
> > at
> > org.apache.spark.streaming.kafka010.KafkaRDD$KafkaRDDIterator.<init>(
> KafkaRDD.scala:210)
> > at org.apache.spark.streaming.kafka010.KafkaRDD.compute(
> KafkaRDD.scala:185)
> > at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319)
> > at org.apache.spark.rdd.RDD.iterator(RDD.scala:283)
> > at org.apache.spark.rdd.MapPartitionsRDD.compute(
> MapPartitionsRDD.scala:38)
> > at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319)
> > at org.apache.spark.rdd.RDD.iterator(RDD.scala:283)
> > at org.apache.spark.rdd.MapPartitionsRDD.compute(
> MapPartitionsRDD.scala:38)
> > at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319)
> > at org.apache.spark.rdd.RDD.iterator(RDD.scala:283)
> > at org.apache.spark.rdd.MapPartitionsRDD.compute(
> MapPartitionsRDD.scala:38)
> > at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319)
> > at org.apache.spark.rdd.RDD.iterator(RDD.scala:283)
> > at
> > org.apache.spark.scheduler.ShuffleMapTask.runTask(
> ShuffleMapTask.scala:79)
> > at
> > org.apache.spark.scheduler.ShuffleMapTask.runTask(
> ShuffleMapTask.scala:47)
> > at org.apache.spark.scheduler.Task.run(Task.scala:85)
> > at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274)
> > at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
> > at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
> > at java.lang.Thread.run(Unknown Source)
> >
> >
> > On Wed, Sep 7, 2016 at 3:07 PM, Cody Koeninger <c...@koeninger.org>
> wrote:
> >>
> >> you could try setting
> >>
> >> spark.streaming.kafka.consumer.cache.initialCapacity
> >>
> >> spark.streaming.kafka.consumer.cache.maxCapacity
> >>
> >> to 1
> >>
> >> On Wed, Sep 7, 2016 at 2:02 PM, Srikanth <srikanth...@gmail.com> wrote:
> >> > I had a look at the executor logs and noticed that this exception
> >> > happens
> >> > only when using the cached consumer.
> >> > Every retry is successful. This is consistent.
> >> > One possibility is that the cached consumer is causing the failure as
> >> > retry
> >> > clears it.
> >> > Is there a way to disable cache and test this?
> >> > Again, kafkacat is running fine on the same node.
> >> >
> >> > 16/09/07 16:00:00 INFO Executor: Running task 1.0 in stage 138.0 (TID
> >> > 7849)
> >> > 16/09/07 16:00:00 INFO Executor: Running task 3.0 in stage 138.0 (TID
> >> > 7851
> >> >
> >> > 16/09/07 16:00:00 INFO KafkaRDD: Computing topic mt_event, partition 2
> >> > offsets 57079162 -> 57090330
> >> > 16/09/07 16:00:00 INFO KafkaRDD: Computing topic mt_event, partition 0
> >> > offsets 57098866 -> 57109957
> >> > 16/09/07 16:00:00 INFO Executor: Finished task 3.0 in stage 138.0 (TID
> >> > 7851). 1030 bytes result sent to driver
> >> > 16/09/07 16:00:02 ERROR Executor: Exception in task 1.0 in stage 138.0
> >> > (TID
> >> > 7849)
> >> > java.lang.AssertionError: assertion failed: Failed to get records for
> >> > spark-executor-StreamingPixelCount1 mt_event 0 57100069 after polling
> >> > for
> >> > 2048
> >> > at scala.Predef$.assert(Predef.scala:170)
> >> > at
> >> >
> >> > org.apache.spark.streaming.kafka010.CachedKafkaConsumer.
> get(CachedKafkaConsumer.scala:74)
> >> > at
> >> >
> >> > org.apache.spark.streaming.kafka010.KafkaRDD$KafkaRDDIterator.next(
> KafkaRDD.scala:227)
> >> > at
> >> >
> >> > org.apache.spark.streaming.kafka010.KafkaRDD$KafkaRDDIterator.next(
> KafkaRDD.scala:193)
> >> >
> >> > 16/09/07 16:00:02 INFO CoarseGrainedExecutorBackend: Got assigned task
> >> > 7854
> >> > 16/09/07 16:00:02 INFO Executor: Running task 1.1 in stage 138.0 (TID
> >> > 7854)
> >> > 16/09/07 16:00:02 INFO KafkaRDD: Computing topic mt_event, partition 0
> >> > offsets 57098866 -> 57109957
> >> > 16/09/07 16:00:02 INFO CachedKafkaConsumer: Initial fetch for
> >> > spark-executor-StreamingPixelCount1 mt_event 0 57098866
> >> >
> >> > 16/09/07 16:00:03 INFO Executor: Finished task 1.1 in stage 138.0 (TID
> >> > 7854). 1103 bytes result sent to driver
> >> >
> >> >
> >> >
> >> > On Wed, Aug 24, 2016 at 2:13 PM, Srikanth <srikanth...@gmail.com>
> wrote:
> >> >>
> >> >> Thanks Cody. Setting poll timeout helped.
> >> >> Our network is fine but brokers are not fully provisioned in test
> >> >> cluster.
> >> >> But there isn't enough load to max out on broker capacity.
> >> >> Curious that kafkacat running on the same node doesn't have any
> issues.
> >> >>
> >> >> Srikanth
> >> >>
> >> >> On Tue, Aug 23, 2016 at 9:52 PM, Cody Koeninger <c...@koeninger.org>
> >> >> wrote:
> >> >>>
> >> >>> You can set that poll timeout higher with
> >> >>>
> >> >>> spark.streaming.kafka.consumer.poll.ms
> >> >>>
> >> >>> but half a second is fairly generous. I'd try to take a look at
> >> >>> what's going on with your network or kafka broker during that time.
> >> >>>
> >> >>> On Tue, Aug 23, 2016 at 4:44 PM, Srikanth <srikanth...@gmail.com>
> >> >>> wrote:
> >> >>> > Hello,
> >> >>> >
> >> >>> > I'm getting the below exception when testing Spark 2.0 with Kafka
> >> >>> > 0.10.
> >> >>> >
> >> >>> >> 16/08/23 16:31:01 INFO AppInfoParser: Kafka version : 0.10.0.0
> >> >>> >> 16/08/23 16:31:01 INFO AppInfoParser: Kafka commitId :
> >> >>> >> b8642491e78c5a13
> >> >>> >> 16/08/23 16:31:01 INFO CachedKafkaConsumer: Initial fetch for
> >> >>> >> spark-executor-example mt_event 0 15782114
> >> >>> >> 16/08/23 16:31:01 INFO AbstractCoordinator: Discovered
> coordinator
> >> >>> >> 10.150.254.161:9233 (id: 2147483646 rack: null) for group
> >> >>> >> spark-executor-example.
> >> >>> >> 16/08/23 16:31:02 ERROR Executor: Exception in task 0.0 in stage
> >> >>> >> 1.0
> >> >>> >> (TID
> >> >>> >> 6)
> >> >>> >> java.lang.AssertionError: assertion failed: Failed to get records
> >> >>> >> for
> >> >>> >> spark-executor-example mt_event 0 15782114 after polling for 512
> >> >>> >> at scala.Predef$.assert(Predef.scala:170)
> >> >>> >> at
> >> >>> >>
> >> >>> >>
> >> >>> >> org.apache.spark.streaming.kafka010.CachedKafkaConsumer.
> get(CachedKafkaConsumer.scala:74)
> >> >>> >> at
> >> >>> >>
> >> >>> >>
> >> >>> >> org.apache.spark.streaming.kafka010.KafkaRDD$
> KafkaRDDIterator.next(KafkaRDD.scala:227)
> >> >>> >> at
> >> >>> >>
> >> >>> >>
> >> >>> >> org.apache.spark.streaming.kafka010.KafkaRDD$
> KafkaRDDIterator.next(KafkaRDD.scala:193)
> >> >>> >> at scala.collection.Iterator$$anon$11.next(Iterator.scala:409)
> >> >>> >
> >> >>> >
> >> >>> > I get this error intermittently. Sometimes a few batches are
> >> >>> > scheduled
> >> >>> > and
> >> >>> > run fine. Then I get this error.
> >> >>> > kafkacat is able to fetch from this topic continuously.
> >> >>> >
> >> >>> > Full exception is here --
> >> >>> >
> >> >>> > https://gist.github.com/SrikanthTati/
> c2e95c4ac689cd49aab817e24ec42767
> >> >>> >
> >> >>> > Srikanth
> >> >>
> >> >>
> >> >
> >
> >
>
