MapR [MS-925] After upgrade to MEP 6.2 (Spark 2.4.0) can no longer (a…

…pache#588) consume Kafka / MapR Streams.
arjunshroff · Oct 23, 2019 · c5cbbcc · c5cbbcc
1 parent c7e076e
commit c5cbbcc
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 6 deletions.
diff --git a/...rnal/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/ConsumerStrategy.scala b/...rnal/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/ConsumerStrategy.scala
@@ -105,10 +105,11 @@ private case class Subscribe[K, V](
       val shouldSuppress =
         aor != null && aor.asInstanceOf[String].toUpperCase(Locale.ROOT) == "NONE"
       try {
-         if (KafkaUtils.isStreams(currentOffsets.asScala.toMap.map(a => (a._1, a._2.toLong)))) {
+        consumer.poll(0)
+
+        if (KafkaUtils.isStreams(toSeek.asScala.toMap.map(a => (a._1, a._2.toLong)))) {
           KafkaUtils.waitForConsumerAssignment(consumer, toSeek.keySet())
-        } else {
-          consumer.poll(0) }
+        }
       } catch {
         case x: NoOffsetForPartitionException if shouldSuppress =>
           logWarning("Catching NoOffsetForPartitionException since " +

diff --git a/external/kafka-0-9/src/main/scala/org/apache/spark/streaming/kafka09/ConsumerStrategy.scala b/external/kafka-0-9/src/main/scala/org/apache/spark/streaming/kafka09/ConsumerStrategy.scala
@@ -106,10 +106,10 @@ private case class Subscribe[K, V](
       val shouldSuppress =
         aor != null && aor.asInstanceOf[String].toUpperCase(Locale.ROOT) == "NONE"
       try {
-        if (KafkaUtils.isStreams(currentOffsets.asScala.toMap.map(a => (a._1, a._2.toLong)))) {
+        consumer.poll(0)
+
+        if (KafkaUtils.isStreams(toSeek.asScala.toMap.map(a => (a._1, a._2.toLong)))) {
           KafkaUtils.waitForConsumerAssignment(consumer, toSeek.keySet())
-        } else {
-          consumer.poll(0)
         }
       } catch {
         case x: NoOffsetForPartitionException if shouldSuppress =>