MapR [SPARK-737] Calling poll(1000) from paranoidPoll causes batch sc…

…heduling delay (apache#675)
mapr · Nov 6, 2023 · dc615b9 · dc615b9
1 parent 84fcd71
commit dc615b9
Show file tree

Hide file tree

Showing 2 changed files with 2 additions and 2 deletions.
diff --git a/...fka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/DirectKafkaInputDStream.scala b/...fka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/DirectKafkaInputDStream.scala
@@ -170,7 +170,7 @@ private[spark] class DirectKafkaInputDStream[K, V](
    * which would throw off consumer position.  Fix position if this happens.
    */
   private def paranoidPoll(c: Consumer[K, V]): Unit = {
-    val msgs = c.poll(1000)
+    val msgs = c.poll(0)
 
     val newAssignment = c.assignment()
     val parts = if (currentOffsets.size < newAssignment.size()) {

diff --git a/...kafka-0-9/src/main/scala/org/apache/spark/streaming/kafka09/DirectKafkaInputDStream.scala b/...kafka-0-9/src/main/scala/org/apache/spark/streaming/kafka09/DirectKafkaInputDStream.scala
@@ -171,7 +171,7 @@ private[spark] class DirectKafkaInputDStream[K, V](
    * which would throw off consumer position.  Fix position if this happens.
    */
   private def paranoidPoll(c: Consumer[K, V]): Unit = {
-    val msgs = c.poll(1000)
+    val msgs = c.poll(0)
 
     val newAssignment = c.assignment()
     val parts = if (currentOffsets.size < newAssignment.size()) {