add logging for test

Author: brkyvz

extras/kinesis-asl/pom.xml

@@ -70,6 +70,12 @@
       <version>${aws.kinesis.producer.version}</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>com.google.protobuf</groupId>
+      <artifactId>protobuf-java</artifactId>
+      <version>2.6.1</version>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>org.mockito</groupId>
       <artifactId>mockito-core</artifactId>

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

@@ -70,6 +70,7 @@ private[kinesis] class KinesisRecordProcessor[T](
    *   in the DStream
    */
   override def processRecords(batch: List[Record], checkpointer: IRecordProcessorCheckpointer) {
+    logInfo(s"Received batch: $batch")
     if (!receiver.isStopped()) {
       try {
         receiver.addRecords(shardId, batch)

extras/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisStreamSuite.scala

@@ -78,8 +78,8 @@ abstract class KinesisStreamTests(aggregateTestData: Boolean) extends KinesisFun
     if (testUtils != null) {
       // Delete the Kinesis stream as well as the DynamoDB table generated by
       // Kinesis Client Library when consuming the stream
-      testUtils.deleteStream()
-      testUtils.deleteDynamoDBTable(appName)
+      // testUtils.deleteStream()
+      // testUtils.deleteDynamoDBTable(appName)
     }
   }
 
@@ -182,7 +182,7 @@ abstract class KinesisStreamTests(aggregateTestData: Boolean) extends KinesisFun
     val collected = new mutable.HashSet[Int] with mutable.SynchronizedSet[Int]
     stream.map { bytes => new String(bytes).toInt }.foreachRDD { rdd =>
       collected ++= rdd.collect()
-      logInfo("Collected = " + rdd.collect().toSeq.mkString(", "))
+      logInfo("Collected = " + collected.mkString(", "))
     }
     ssc.start()
 
@@ -207,7 +207,7 @@ abstract class KinesisStreamTests(aggregateTestData: Boolean) extends KinesisFun
     val collected = new mutable.HashSet[Int] with mutable.SynchronizedSet[Int]
     stream.foreachRDD { rdd =>
       collected ++= rdd.collect()
-      logInfo("Collected = " + rdd.collect().toSeq.mkString(", "))
+      logInfo("Collected = " + collected.mkString(", "))
     }
     ssc.start()
 

extras/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisTestUtils.scala

@@ -139,7 +139,6 @@ private[kinesis] class KinesisTestUtils extends Logging {
       }
     }
 
-
     logInfo(s"Pushed $testData:\n\t ${shardIdToSeqNumbers.mkString("\n\t")}")
     shardIdToSeqNumbers.toMap
   }