MapR [SPARK-795] - extended logging for Spark Streaming and Structured Str… (apache#729)

Author: vishchenkomapr

external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/ConsumerStrategy.scala

@@ -26,8 +26,8 @@ import org.apache.kafka.clients.admin.Admin
 import org.apache.kafka.clients.consumer.{Consumer, KafkaConsumer}
 import org.apache.kafka.clients.consumer.internals.NoOpConsumerRebalanceListener
 import org.apache.kafka.common.TopicPartition
-
 import org.apache.spark.internal.Logging
+
 import org.apache.spark.kafka010.{KafkaConfigUpdater, KafkaRedactionUtil}
 
 /**
@@ -106,6 +106,7 @@ private[kafka010] case class SubscribeStrategy(topics: Seq[String])
     val updatedKafkaParams = setAuthenticationConfigIfNeeded(kafkaParams)
     val consumer = new KafkaConsumer[Array[Byte], Array[Byte]](updatedKafkaParams)
     consumer.subscribe(topics.asJava)
+    logDebug(s"The consumer has been subscribed to topics: ${topics.mkString(", ")}")
     consumer
   }
 

external/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/ConsumerStrategy.scala

@@ -91,11 +91,11 @@ private case class Subscribe[K, V](
     val updatedKafkaParams = setAuthenticationConfigIfNeeded(kafkaParams)
     val consumer = new KafkaConsumer[K, V](updatedKafkaParams)
     consumer.subscribe(topics)
-    val toSeek = if (currentOffsets.isEmpty) {
-      offsets
-    } else {
-      currentOffsets
-    }
+    logDebug(s"The consumer has been subscribed to topics: ${String.join(", ", topics)}")
+
+    val toSeek = if (currentOffsets.isEmpty) offsets else currentOffsets
+    logDebug(s"TopicPartition-Offset map for seeking: $toSeek")
+
     if (!toSeek.isEmpty) {
       // work around KAFKA-3370 when reset is none
       // poll will throw if no position, i.e. auto offset reset none and no explicit position
@@ -106,6 +106,7 @@ private case class Subscribe[K, V](
         aor != null && aor.asInstanceOf[String].toUpperCase(Locale.ROOT) == "NONE"
       try {
         consumer.poll(0)
+        logDebug("Data fetched for the topics.")
 
         if (KafkaUtils.isStreams(toSeek.asScala.toMap.map(a => (a._1, a._2.toLong)))) {
           KafkaUtils.waitForConsumerAssignment(consumer, toSeek.keySet())
@@ -115,7 +116,10 @@ private case class Subscribe[K, V](
           logWarning("Catching NoOffsetForPartitionException since " +
             ConsumerConfig.AUTO_OFFSET_RESET_CONFIG + " is none.  See KAFKA-3370")
       }
-      toSeek.asScala.foreach { case (topicPartition, offset) =>
+
+      toSeek.asScala.foreach {
+        case (topicPartition, offset) =>
+          logDebug(s"Seeking for topicPartition=$topicPartition with offset=$offset")
           consumer.seek(topicPartition, offset)
       }
       // we've called poll, we must pause or next poll may consume messages and set position
@@ -152,11 +156,11 @@ private case class SubscribePattern[K, V](
     val updatedKafkaParams = setAuthenticationConfigIfNeeded(kafkaParams)
     val consumer = new KafkaConsumer[K, V](updatedKafkaParams)
     consumer.subscribe(pattern, new NoOpConsumerRebalanceListener())
-    val toSeek = if (currentOffsets.isEmpty) {
-      offsets
-    } else {
-      currentOffsets
-    }
+    logDebug(s"The consumer has been subscribed to topics matching a pattern: $pattern")
+
+    val toSeek = if (currentOffsets.isEmpty) offsets else currentOffsets
+    logDebug(s"TopicPartition-Offset map for seeking: $toSeek")
+
     if (!toSeek.isEmpty) {
       // work around KAFKA-3370 when reset is none, see explanation in Subscribe above
       val aor = kafkaParams.get(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG)
@@ -167,13 +171,17 @@ private case class SubscribePattern[K, V](
           KafkaUtils.waitForConsumerAssignment(consumer, toSeek.keySet())
         } else {
           consumer.poll(0)
+          logDebug("Data fetched for the topics.")
         }
       } catch {
         case x: NoOffsetForPartitionException if shouldSuppress =>
           logWarning("Catching NoOffsetForPartitionException since " +
             ConsumerConfig.AUTO_OFFSET_RESET_CONFIG + " is none.  See KAFKA-3370")
       }
-      toSeek.asScala.foreach { case (topicPartition, offset) =>
+
+      toSeek.asScala.foreach {
+        case (topicPartition, offset) =>
+          logDebug(s"Seeking for topicPartition=$topicPartition with offset=$offset")
           consumer.seek(topicPartition, offset)
       }
       // we've called poll, we must pause or next poll may consume messages and set position

external/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/KafkaUtils.scala

@@ -227,13 +227,17 @@ object KafkaUtils extends Logging {
       Thread.sleep(500)
       timeout += 500
     }
+
+    if (timeout >= waitingForAssigmentTimeout) {
+      logError(
+        s"""Consumer assignment wasn't completed within the timeout $waitingForAssigmentTimeout.
+           |Assigned partitions: ${consumer.assignment()}.""".stripMargin)
+    }
   }
 
   // Determine if Apache Kafka is used instead of MapR Streams
   def isStreams(currentOffsets: Map[TopicPartition, Long]): Boolean =
     currentOffsets.keys.map(_.topic()).exists(topic => topic.startsWith("/") && topic.contains(":"))
-
-
 }
 
 object KafkaUtilsPythonHelper {

external/kafka-0-9/src/main/scala/org/apache/spark/streaming/kafka09/ConsumerStrategy.scala

@@ -92,11 +92,11 @@ private case class Subscribe[K, V](
   def onStart(currentOffsets: ju.Map[TopicPartition, jl.Long]): Consumer[K, V] = {
     val consumer = new KafkaConsumer[K, V](kafkaParams)
     consumer.subscribe(topics)
-    val toSeek = if (currentOffsets.isEmpty) {
-      offsets
-    } else {
-      currentOffsets
-    }
+    logDebug(s"The consumer has been subscribed to topics: ${String.join(", ", topics)}")
+
+    val toSeek = if (currentOffsets.isEmpty) offsets else currentOffsets
+    logDebug(s"TopicPartition-Offset map for seeking: $toSeek")
+
     if (!toSeek.isEmpty) {
       // work around KAFKA-3370 when reset is none
       // poll will throw if no position, i.e. auto offset reset none and no explicit position
@@ -107,6 +107,7 @@ private case class Subscribe[K, V](
         aor != null && aor.asInstanceOf[String].toUpperCase(Locale.ROOT) == "NONE"
       try {
         consumer.poll(0)
+        logDebug("Data fetched for the topics.")
 
         if (KafkaUtils.isStreams(toSeek.asScala.toMap.map(a => (a._1, a._2.toLong)))) {
           KafkaUtils.waitForConsumerAssignment(consumer, toSeek.keySet())
@@ -116,7 +117,10 @@ private case class Subscribe[K, V](
           logWarning("Catching NoOffsetForPartitionException since " +
             ConsumerConfig.AUTO_OFFSET_RESET_CONFIG + " is none.  See KAFKA-3370")
       }
-      toSeek.asScala.foreach { case (topicPartition, offset) =>
+
+      toSeek.asScala.foreach {
+        case (topicPartition, offset) =>
+          logDebug(s"Seeking for topicPartition=$topicPartition with offset=$offset")
           consumer.seek(topicPartition, offset)
       }
       // we've called poll, we must pause or next poll may consume messages and set position
@@ -152,11 +156,11 @@ private case class SubscribePattern[K, V](
   def onStart(currentOffsets: ju.Map[TopicPartition, jl.Long]): Consumer[K, V] = {
     val consumer = new KafkaConsumer[K, V](kafkaParams)
     consumer.subscribe(pattern, new NoOpConsumerRebalanceListener())
-    val toSeek = if (currentOffsets.isEmpty) {
-      offsets
-    } else {
-      currentOffsets
-    }
+    logDebug(s"The consumer has been subscribed to topics matching a pattern: $pattern")
+
+    val toSeek = if (currentOffsets.isEmpty) offsets else currentOffsets
+    logDebug(s"TopicPartition-Offset map for seeking: $toSeek")
+
     if (!toSeek.isEmpty) {
       // work around KAFKA-3370 when reset is none, see explanation in Subscribe above
       val aor = kafkaParams.get(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG)
@@ -167,13 +171,17 @@ private case class SubscribePattern[K, V](
           KafkaUtils.waitForConsumerAssignment(consumer, toSeek.keySet())
         } else {
           consumer.poll(0)
+          logDebug("Data fetched for the topics.")
         }
       } catch {
         case x: NoOffsetForPartitionException if shouldSuppress =>
           logWarning("Catching NoOffsetForPartitionException since " +
             ConsumerConfig.AUTO_OFFSET_RESET_CONFIG + " is none.  See KAFKA-3370")
       }
-      toSeek.asScala.foreach { case (topicPartition, offset) =>
+
+      toSeek.asScala.foreach {
+        case (topicPartition, offset) =>
+          logDebug(s"Seeking for topicPartition=$topicPartition with offset=$offset")
           consumer.seek(topicPartition, offset)
       }
       // we've called poll, we must pause or next poll may consume messages and set position

external/kafka-0-9/src/main/scala/org/apache/spark/streaming/kafka09/KafkaUtils.scala

@@ -243,6 +243,12 @@ object KafkaUtils extends Logging {
       Thread.sleep(500)
       timeout += 500
     }
+
+    if (timeout >= waitingForAssigmentTimeout) {
+      logError(
+        s"""Consumer assignment wasn't completed within the timeout $waitingForAssigmentTimeout.
+           |Assigned partitions: ${consumer.assignment()}.""".stripMargin)
+    }
   }
 
   // Determine if Apache Kafka is used instead of MapR Streams