changed debug's to infos

Author: brkyvz

extras/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisBackedBlockRDD.scala

@@ -99,7 +99,7 @@ class KinesisBackedBlockRDD[T: ClassTag](
     val blockId = partition.blockId
 
     def getBlockFromBlockManager(): Option[Iterator[T]] = {
-      logDebug(s"Read partition data of $this from block manager, block $blockId")
+      logInfo(s"Read partition data of $this from block manager, block $blockId")
       blockManager.get(blockId).map(_.data.asInstanceOf[Iterator[T]])
     }
 

extras/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisInputDStream.scala

@@ -54,7 +54,7 @@ private[kinesis] class KinesisInputDStream[T: ClassTag](
       val seqNumRanges = blockInfos.map {
         _.metadataOption.get.asInstanceOf[SequenceNumberRanges] }.toArray
       val isBlockIdValid = blockInfos.map { _.isBlockIdValid() }.toArray
-      logDebug(s"Creating KinesisBackedBlockRDD for $time with ${seqNumRanges.length} " +
+      logInfo(s"Creating KinesisBackedBlockRDD for $time with ${seqNumRanges.length} " +
           s"seq number ranges: ${seqNumRanges.mkString(", ")} ")
       new KinesisBackedBlockRDD(
         context.sc, regionName, endpointUrl, blockIds, seqNumRanges,

extras/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisReceiver.scala

@@ -206,6 +206,7 @@ private[kinesis] class KinesisReceiver[T](
       val dataIterator = records.iterator().asScala.map(messageHandler)
       val metadata = SequenceNumberRange(streamName, shardId,
         records.get(0).getSequenceNumber(), records.get(records.size() - 1).getSequenceNumber())
+      logInfo(s"blockGenerator.multipleDataWithCallback: iterator: $dataIterator, meta: $metadata")
       blockGenerator.addMultipleDataWithCallback(dataIterator, metadata)
 
     }
@@ -231,7 +232,7 @@ private[kinesis] class KinesisReceiver[T](
   private def finalizeRangesForCurrentBlock(blockId: StreamBlockId): Unit = {
     blockIdToSeqNumRanges(blockId) = SequenceNumberRanges(seqNumRangesInCurrentBlock.toArray)
     seqNumRangesInCurrentBlock.clear()
-    logDebug(s"Generated block $blockId has $blockIdToSeqNumRanges")
+    logInfo(s"Generated block $blockId has $blockIdToSeqNumRanges")
   }
 
   /** Store the block along with its associated ranges */

extras/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisRecordProcessor.scala

@@ -70,11 +70,11 @@ private[kinesis] class KinesisRecordProcessor[T](
    *   in the DStream
    */
   override def processRecords(batch: List[Record], checkpointer: IRecordProcessorCheckpointer) {
-    logInfo(s"Received batch: $batch")
+    logInfo(s"Received batch: $batch. Is receiver alive: ${!receiver.isStopped()}")
     if (!receiver.isStopped()) {
       try {
         receiver.addRecords(shardId, batch)
-        logDebug(s"Stored: Worker $workerId stored ${batch.size} records for shardId $shardId")
+        logInfo(s"Stored: Worker $workerId stored ${batch.size} records for shardId $shardId")
 
         /*
          *
@@ -111,7 +111,7 @@ private[kinesis] class KinesisRecordProcessor[T](
            *     more than once.
            */
           logError(s"Exception:  WorkerId $workerId encountered and exception while storing " +
-              " or checkpointing a batch for workerId $workerId and shardId $shardId.", e)
+              s" or checkpointing a batch for workerId $workerId and shardId $shardId.", e)
 
           /* Rethrow the exception to the Kinesis Worker that is managing this RecordProcessor. */
           throw e